Merge remote-tracking branch 'origin/main' into cxymds/fix-5943-hard-quota-reservations

# Conflicts:
#	rustfs/src/app/object_usecase.rs

crates/ecstore/src/erasure/codec/workspace.rs

@@ -76,6 +76,13 @@ impl ShardBufferPool {
         self.buffers[index] = Some(buf);
     }
 
+    #[cfg(test)]
+    pub(crate) fn stored_allocation(&self, index: usize) -> Option<(*const u8, usize)> {
+        self.buffers
+            .get(index)
+            .and_then(|buf| buf.as_ref().map(|buf| (buf.as_ptr(), buf.capacity())))
+    }
+
     #[cfg(test)]
     fn stored_capacity(&self, index: usize) -> Option<usize> {
         self.buffers.get(index).and_then(|buf| buf.as_ref().map(Vec::capacity))

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

@@ -125,6 +125,11 @@ where
         self.last_verify_duration
     }
 
+    #[cfg(test)]
+    pub(crate) fn inner_ref(&self) -> &R {
+        &self.inner
+    }
+
     /// Read a single (hash+data) block, verify hash, and copy `out.len()` bytes
     /// into `out`. Returns an error if the shard is short, the hash mismatches,
     /// or `out` is larger than one shard. On error `out`'s contents are

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

@@ -25,7 +25,9 @@ use crate::disk::error_reduce::reduce_errs;
 use crate::erasure::codec::workspace::ShardBufferPool;
 use crate::erasure::coding::{BitrotReader, Erasure};
 use crate::io_support::bitrot::DeferredReaderStripeHandle;
-use crate::set_disk::shard_source::{ShardReadCost, ShardStripeSource, StripeReadState};
+use crate::set_disk::shard_source::{
+    INLINE_SHARD_SLOTS, ShardBuffers, ShardErrors, ShardReadCost, ShardStripeSource, StripeReadState,
+};
 use futures::FutureExt;
 use futures::stream::{FuturesUnordered, StreamExt};
 use pin_project_lite::pin_project;
@@ -41,9 +43,6 @@ use tracing::{debug, error, warn};
 
 type ShardReadFuture<'a> = Pin<Box<dyn Future<Output = (usize, ShardReadCost, Result<Vec<u8>, Error>, bool)> + Send + 'a>>;
 
-const INLINE_SHARD_SLOTS: usize = 32;
-type ShardBuffers = SmallVec<[Option<Vec<u8>>; INLINE_SHARD_SLOTS]>;
-type ShardErrors = SmallVec<[Option<Error>; INLINE_SHARD_SLOTS]>;
 type ShardIndexes = SmallVec<[usize; INLINE_SHARD_SLOTS]>;
 type ActiveReaders = SmallVec<[bool; INLINE_SHARD_SLOTS]>;
 
@@ -392,6 +391,7 @@ pub(crate) struct ParallelReader<R> {
     // Request-scoped shard buffers keyed by shard index. Keeping ownership in
     // `ParallelReader` avoids dropping unused parity/backup slot buffers between stripes.
     buffers: ShardBufferPool,
+    stripe_state: Option<Box<StripeReadState>>,
     // Lockstep-path state (verify_reconstruction == true). `engaged[i]` marks
     // readers that participate in each stripe read: all data slots from the
     // start, parity slots only once a data shard is missing/dead. Unengaged
@@ -596,6 +596,7 @@ where
             verify_reconstruction,
             locality_preference_enabled: get_shard_locality_preference_enabled(),
             buffers: ShardBufferPool::new(e.data_shards + e.parity_shards),
+            stripe_state: None,
             engaged,
             deferred_handles: Vec::new(),
             stripe_index: 0,
@@ -700,6 +701,12 @@ where
 {
     #[hotpath::measure(impl_type = "ParallelReader")]
     pub async fn read(&mut self) -> StripeReadOutput {
+        let mut state = StripeReadState::with_slot_count(self.readers.len(), self.data_shards);
+        self.read_into_state(&mut state).await;
+        state.into_parts()
+    }
+
+    async fn read_into_state(&mut self, state: &mut StripeReadState) {
         // On the reconstruction-verifying GET path, read every live shard reader
         // in lockstep so all readers advance one block per stripe and stay
         // mutually aligned. The adaptive data-first path below only reads
@@ -709,12 +716,14 @@ where
         // than the data shards, producing "inconsistent read source shards" and
         // truncating large-object GETs under concurrency (backlog#832).
         if self.verify_reconstruction {
-            return self.read_lockstep().await;
+            self.read_lockstep(state).await;
+            return;
         }
         // if self.readers.len() != self.total_shards {
         //     return Err(io::Error::new(ErrorKind::InvalidInput, "Invalid number of readers"));
         // }
         let num_readers = self.readers.len();
+        state.reset(num_readers, self.data_shards);
 
         let shard_size = if self.offset + self.shard_size > self.shard_file_size {
             self.shard_file_size - self.offset
@@ -723,7 +732,7 @@ where
         };
 
         if shard_size == 0 {
-            return (smallvec![None; num_readers], smallvec![None; num_readers]);
+            return;
         }
 
         // Advance to the next stripe so the following read() computes the correct
@@ -734,8 +743,7 @@ where
         // is only read above to derive `shard_size`, so advancing here is safe.
         self.offset += shard_size;
 
-        let mut shards: ShardBuffers = smallvec![None; num_readers];
+        let (shards, errs) = state.parts_mut();
-        let mut errs: ShardErrors = smallvec![None; num_readers];
         let read_costs = self.read_costs.as_slice();
         let locality_preference_enabled = self.locality_preference_enabled;
         let low_cost_available = self
@@ -882,8 +890,8 @@ where
                 }
 
                 let result_is_err = record_shard_read_result(
-                    &mut shards,
+                    shards,
-                    &mut errs,
+                    errs,
                     &mut retire_readers,
                     &mut success,
                     &mut successful_costs,
@@ -944,8 +952,8 @@ where
                     active_readers[i] = false;
                     completed += 1;
                     if record_shard_read_result(
-                        &mut shards,
+                        shards,
-                        &mut errs,
+                        errs,
                         &mut retire_readers,
                         &mut success,
                         &mut successful_costs,
@@ -957,7 +965,7 @@ where
                         failed += 1;
                     }
                 }
-                retire_abandoned_readers(&mut errs, &mut retire_readers, &active_readers);
+                retire_abandoned_readers(errs, &mut retire_readers, &active_readers);
             }
 
             if let Some(path) = self.metrics_path {
@@ -1001,8 +1009,6 @@ where
         for i in retire_readers {
             self.readers[i] = None;
         }
-
-        (shards, errs)
     }
 
     /// Lockstep stripe read for the reconstruction-verifying GET path.
@@ -1030,18 +1036,18 @@ where
     /// stripe would reintroduce the desync. A parity reader that cannot be
     /// realigned (no pending deferred handle) is likewise retired instead of
     /// being read out of position.
-    async fn read_lockstep(&mut self) -> StripeReadOutput {
+    async fn read_lockstep(&mut self, state: &mut StripeReadState) {
         let num_readers = self.readers.len();
+        state.reset(num_readers, self.data_shards);
         let shard_size = if self.offset + self.shard_size > self.shard_file_size {
             self.shard_file_size - self.offset
         } else {
             self.shard_size
         };
 
-        let mut shards: ShardBuffers = smallvec![None; num_readers];
+        let (shards, errs) = state.parts_mut();
-        let mut errs: ShardErrors = smallvec![None; num_readers];
         if shard_size == 0 {
-            return (shards, errs);
+            return;
         }
 
         // Advance to the next stripe (see the matching note in `read`); the
@@ -1279,8 +1285,6 @@ where
         for i in retire_readers {
             self.readers[i] = None;
         }
-
-        (shards, errs)
     }
 
     /// Attempt to bring an as-yet-unread parity reader into the lockstep read
@@ -1337,10 +1341,20 @@ impl<R> ShardStripeSource for ParallelReader<R>
 where
     R: crate::erasure::coding::ShardSource,
 {
-    async fn read_next_stripe(&mut self) -> StripeReadState {
+    async fn read_next_stripe(&mut self) -> Box<StripeReadState> {
-        let read_quorum = self.data_shards;
+        let mut state = self
-        let (shards, errors) = ParallelReader::read(self).await;
+            .stripe_state
-        StripeReadState::from_parts_with_read_costs(shards, errors, &self.read_costs, read_quorum)
+            .take()
+            .unwrap_or_else(|| Box::new(StripeReadState::with_slot_count(self.readers.len(), self.data_shards)));
+        self.read_into_state(&mut state).await;
+        state
+    }
+
+    fn recycle_stripe(&mut self, mut state: Box<StripeReadState>) {
+        self.recycle_shards(state.shards_mut());
+        state.reset(0, self.data_shards);
+        debug_assert!(self.stripe_state.is_none(), "a stripe cannot be recycled twice");
+        self.stripe_state = Some(state);
     }
 }
 
@@ -1972,13 +1986,18 @@ mod tests {
     type BoxedShardReader = crate::io_support::bitrot::ShardReader;
 
     #[test]
-    fn shard_scratch_stays_inline_through_the_common_limit_and_spills_safely() {
+    fn parallel_reader_keeps_stripe_scratch_out_of_line() {
-        let inline: ShardBuffers = smallvec![None; INLINE_SHARD_SLOTS];
+        eprintln!(
-        assert!(!inline.spilled(), "the common shard-count boundary must not allocate");
+            "parallel_reader={} stripe_state={} cached_state={}",
-
+            std::mem::size_of::<ParallelReader<Cursor<Vec<u8>>>>(),
-        let spilled: ShardBuffers = smallvec![None; INLINE_SHARD_SLOTS + 1];
+            std::mem::size_of::<StripeReadState>(),
-        assert!(spilled.spilled(), "larger supported shard counts must fall back to the heap");
+            std::mem::size_of::<Option<Box<StripeReadState>>>()
-        assert_eq!(spilled.len(), INLINE_SHARD_SLOTS + 1);
+        );
+        assert_eq!(
+            std::mem::size_of::<Option<Box<StripeReadState>>>(),
+            std::mem::size_of::<usize>(),
+            "the request-scoped cache must remain pointer-sized",
+        );
     }
 
     #[tokio::test]
@@ -1997,6 +2016,62 @@ mod tests {
         assert_eq!(errors.len(), TOTAL_SHARDS);
     }
 
+    #[tokio::test]
+    async fn codec_reader_reuses_inline_and_spilled_stripe_scratch_between_reads() {
+        for total_shards in [INLINE_SHARD_SLOTS, INLINE_SHARD_SLOTS + 1] {
+            let data_shards = total_shards - 1;
+            let readers = std::iter::repeat_with(|| None).take(total_shards).collect();
+            let erasure = Erasure::new(data_shards, 1, data_shards * 2);
+            let mut reader: ParallelReader<Cursor<Vec<u8>>> = ParallelReader::new(readers, erasure, 0, data_shards * 2);
+
+            let first = ShardStripeSource::read_next_stripe(&mut reader).await;
+            let first_state = (&*first) as *const StripeReadState;
+            let first_storage = first.scratch_storage();
+            assert_eq!(first_storage.2, total_shards > INLINE_SHARD_SLOTS);
+            assert_eq!(first_storage.3, total_shards > INLINE_SHARD_SLOTS);
+            ShardStripeSource::recycle_stripe(&mut reader, first);
+
+            let second = ShardStripeSource::read_next_stripe(&mut reader).await;
+            let second_storage = second.scratch_storage();
+
+            assert_eq!(
+                (&*second) as *const StripeReadState,
+                first_state,
+                "the request-scoped state must be reused"
+            );
+            assert_eq!(second_storage.0, first_storage.0, "shard slots must reuse their allocation");
+            assert_eq!(second_storage.1, first_storage.1, "error slots must reuse their allocation");
+            assert_eq!(second.into_parts().0.len(), total_shards);
+        }
+    }
+
+    #[tokio::test]
+    async fn codec_reader_returns_shard_allocations_to_the_request_pool() {
+        const SHARD_SIZE: usize = 16;
+        let hash_algo = HashAlgorithm::None;
+        let readers = vec![Some(create_reader(SHARD_SIZE, 2, 0x5a, &hash_algo, false).await)];
+        let erasure = Erasure::new(1, 0, SHARD_SIZE);
+        let mut reader = ParallelReader::new(readers, erasure, 0, SHARD_SIZE * 2);
+
+        let first = ShardStripeSource::read_next_stripe(&mut reader).await;
+        let first_allocation = first
+            .shard_allocation(0)
+            .expect("the first stripe should own its shard allocation");
+        ShardStripeSource::recycle_stripe(&mut reader, first);
+        assert_eq!(
+            reader.buffers.stored_allocation(0),
+            Some(first_allocation),
+            "recycling a stripe must return its shard allocation to the request pool"
+        );
+
+        let second = ShardStripeSource::read_next_stripe(&mut reader).await;
+        assert_eq!(
+            second.shard_allocation(0),
+            Some(first_allocation),
+            "the next stripe must reuse the pooled shard allocation"
+        );
+    }
+
     /// Counts the raw bytes pulled from a shard stream, to prove which shards
     /// a decode path actually touches (backlog#923 call-count evidence).
     struct CountingShardReader {

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

@@ -65,7 +65,7 @@ enum FillPolicy {
 }
 
 impl FillPolicy {
-    fn from_env() -> Self {
+    fn load() -> Self {
         match rustfs_utils::get_env_usize(
             ENV_RUSTFS_GET_CODEC_STREAMING_MAX_INFLIGHT,
             DEFAULT_RUSTFS_GET_CODEC_STREAMING_MAX_INFLIGHT,
@@ -75,6 +75,22 @@ impl FillPolicy {
         }
     }
 
+    fn from_env() -> Self {
+        #[cfg(test)]
+        {
+            Self::load()
+        }
+        #[cfg(not(test))]
+        {
+            Self::cached_core(Self::load)
+        }
+    }
+
+    fn cached_core(load: impl FnOnce() -> Self) -> Self {
+        static CACHED: std::sync::OnceLock<FillPolicy> = std::sync::OnceLock::new();
+        *CACHED.get_or_init(load)
+    }
+
     const fn max_inflight(self) -> usize {
         match self {
             Self::SingleInFlight => 1,
@@ -479,22 +495,30 @@ where
     let mut deferred_error = None;
     let fill_stage_start = get_stage_timer_if_enabled(stage_metrics_enabled);
     let stripe_read_stage_start = get_stage_timer_if_enabled(stage_metrics_enabled);
-    let state = source.read_next_stripe().await;
+    let mut state = source.read_next_stripe().await;
     record_get_stage_duration_if_enabled(metrics_path, GET_STAGE_STRIPE_READ, stripe_read_stage_start);
     let decode_stage_start = get_stage_timer_if_enabled(stage_metrics_enabled);
     let mut output_buf = reusable_buffers.pop().unwrap_or_default();
-    let result =
+    let result = match decode_stripe_into(
-        match decode_stripe_into(metrics_path, stage_metrics_enabled, engine, workspace, state, remaining, &mut output_buf) {
+        metrics_path,
-            Ok(true) => Ok(Some(output_buf)),
+        stage_metrics_enabled,
-            Ok(false) => {
+        engine,
-                reusable_buffers.push(output_buf);
+        workspace,
-                Ok(None)
+        &mut state,
-            }
+        remaining,
-            Err(err) => {
+        &mut output_buf,
-                reusable_buffers.push(output_buf);
+    ) {
-                Err(err)
+        Ok(true) => Ok(Some(output_buf)),
-            }
+        Ok(false) => {
-        };
+            reusable_buffers.push(output_buf);
+            Ok(None)
+        }
+        Err(err) => {
+            reusable_buffers.push(output_buf);
+            Err(err)
+        }
+    };
+    source.recycle_stripe(state);
     record_get_stage_duration_if_enabled(metrics_path, GET_STAGE_DECODE, decode_stage_start);
     if let Ok(Some(first_buf)) = result.as_ref() {
         let mut remaining_after_first = remaining.saturating_sub(first_buf.len());
@@ -503,7 +527,7 @@ where
                 break;
             }
             let stripe_read_stage_start = get_stage_timer_if_enabled(stage_metrics_enabled);
-            let state = source.read_next_stripe().await;
+            let mut state = source.read_next_stripe().await;
             record_get_stage_duration_if_enabled(metrics_path, GET_STAGE_STRIPE_READ, stripe_read_stage_start);
             let decode_stage_start = get_stage_timer_if_enabled(stage_metrics_enabled);
             let mut queued_buf = reusable_buffers.pop().unwrap_or_default();
@@ -512,10 +536,11 @@ where
                 stage_metrics_enabled,
                 engine,
                 workspace,
-                state,
+                &mut state,
                 remaining_after_first,
                 &mut queued_buf,
             );
+            source.recycle_stripe(state);
             record_get_stage_duration_if_enabled(metrics_path, GET_STAGE_DECODE, decode_stage_start);
             match queued_result {
                 Ok(true) => {
@@ -717,7 +742,7 @@ fn decode_stripe_into<E>(
     stage_metrics_enabled: bool,
     engine: &E,
     workspace: &mut E::Workspace,
-    state: StripeReadState,
+    state: &mut StripeReadState,
     remaining: usize,
     output: &mut Vec<u8>,
 ) -> io::Result<bool>
@@ -725,7 +750,7 @@ where
     E: ErasureDecodeEngine,
 {
     output.clear();
-    if state.slots().is_empty() {
+    if state.is_empty() {
         return Ok(false);
     }
     if !state.can_decode() {
@@ -741,13 +766,12 @@ where
         );
         record_get_stage_duration_if_enabled(metrics_path, GET_STAGE_RECONSTRUCT, reconstruct_stage_start);
         let emit_stage_start = get_stage_timer_if_enabled(stage_metrics_enabled);
-        emit_data_shards_into(&state, engine.data_shards(), engine.block_size(), remaining, output)?;
+        emit_data_shards_into(state, engine.data_shards(), engine.block_size(), remaining, output)?;
         record_get_stage_duration_if_enabled(metrics_path, GET_STAGE_EMIT, emit_stage_start);
         return Ok(true);
     }
 
-    let (mut shards, _errs) = state.into_parts();
+    let reconstruct_outcome = match engine.reconstruct_into(state.shards_mut(), workspace) {
-    let reconstruct_outcome = match engine.reconstruct_into(&mut shards, workspace) {
         Ok(outcome) => outcome,
         Err(err) => {
             record_get_stage_duration_if_enabled(metrics_path, GET_STAGE_RECONSTRUCT, reconstruct_stage_start);
@@ -757,7 +781,7 @@ where
     rustfs_io_metrics::record_get_object_reconstruct_outcome(metrics_path, engine.engine_name(), reconstruct_outcome);
     record_get_stage_duration_if_enabled(metrics_path, GET_STAGE_RECONSTRUCT, reconstruct_stage_start);
 
-    if shards.len() < engine.data_shards() {
+    if state.shards_mut().len() < engine.data_shards() {
         return Err(io::Error::new(
             ErrorKind::UnexpectedEof,
             "decoded stripe has fewer shards than data shard count",
@@ -766,7 +790,7 @@ where
 
     let emit_stage_start = get_stage_timer_if_enabled(stage_metrics_enabled);
     reserve_output_capacity(output, engine.block_size().min(remaining));
-    for shard in shards.iter().take(engine.data_shards()) {
+    for shard in state.shards_mut().iter().take(engine.data_shards()) {
         if output.len() >= remaining {
             break;
         }
@@ -806,10 +830,7 @@ fn emit_data_shards_into(
         if output.len() >= remaining {
             break;
         }
-        let Some(slot) = state.slot_by_index(index) else {
+        let Some(shard) = state.data_bytes(index) else {
-            return Err(io::Error::new(ErrorKind::UnexpectedEof, "decoded stripe is missing a data shard"));
-        };
-        let Some(shard) = slot.data_bytes() else {
             return Err(io::Error::new(ErrorKind::UnexpectedEof, "decoded stripe is missing a data shard"));
         };
         let copy_len = shard.len().min(remaining - output.len());
@@ -826,7 +847,7 @@ mod tests {
     };
     use crate::erasure::coding::decode::ParallelReader;
     use crate::erasure::coding::{BitrotReader, BitrotWriter, Erasure};
-    use crate::set_disk::shard_source::{ShardSlot, StripeReadState};
+    use crate::set_disk::shard_source::StripeReadState;
     use rustfs_utils::HashAlgorithm;
     use std::collections::VecDeque;
     use std::future::{pending, poll_fn};
@@ -845,6 +866,13 @@ mod tests {
         read_count: Option<Arc<AtomicUsize>>,
     }
 
+    struct RecordingStripeSource {
+        stripes: VecDeque<StripeReadState>,
+        read_quorum: usize,
+        reads: usize,
+        recycles: usize,
+    }
+
     struct BlockingSource {
         started: Arc<Notify>,
         dropped: Arc<AtomicUsize>,
@@ -899,25 +927,43 @@ mod tests {
 
     #[async_trait::async_trait]
     impl ShardStripeSource for VecStripeSource {
-        async fn read_next_stripe(&mut self) -> StripeReadState {
+        async fn read_next_stripe(&mut self) -> Box<StripeReadState> {
             if let Some(read_count) = &self.read_count {
                 read_count.fetch_add(1, Ordering::SeqCst);
             }
-            self.stripes
+            Box::new(
-                .pop_front()
+                self.stripes
-                .unwrap_or_else(|| StripeReadState::new(Vec::new(), self.read_quorum))
+                    .pop_front()
+                    .unwrap_or_else(|| StripeReadState::from_parts(Vec::new(), Vec::new(), self.read_quorum)),
+            )
+        }
+    }
+
+    #[async_trait::async_trait]
+    impl ShardStripeSource for RecordingStripeSource {
+        async fn read_next_stripe(&mut self) -> Box<StripeReadState> {
+            self.reads += 1;
+            Box::new(
+                self.stripes
+                    .pop_front()
+                    .unwrap_or_else(|| StripeReadState::from_parts(Vec::new(), Vec::new(), self.read_quorum)),
+            )
+        }
+
+        fn recycle_stripe(&mut self, _state: Box<StripeReadState>) {
+            self.recycles += 1;
         }
     }
 
     #[async_trait::async_trait]
     impl ShardStripeSource for BlockingSource {
-        async fn read_next_stripe(&mut self) -> StripeReadState {
+        async fn read_next_stripe(&mut self) -> Box<StripeReadState> {
             let _guard = BlockingSourceDropGuard {
                 dropped: Arc::clone(&self.dropped),
             };
             self.started.notify_one();
             pending::<()>().await;
-            StripeReadState::new(Vec::new(), self.read_quorum)
+            Box::new(StripeReadState::from_parts(Vec::new(), Vec::new(), self.read_quorum))
         }
     }
 
@@ -1090,6 +1136,23 @@ mod tests {
         });
     }
 
+    #[test]
+    fn fill_policy_production_cache_loads_once() {
+        use std::cell::Cell;
+
+        let loads = Cell::new(0);
+        for _ in 0..3 {
+            assert_eq!(
+                FillPolicy::cached_core(|| {
+                    loads.set(loads.get() + 1);
+                    FillPolicy::DualInFlight
+                }),
+                FillPolicy::DualInFlight
+            );
+        }
+        assert_eq!(loads.get(), 1, "the production fill policy must not re-read the environment per reader");
+    }
+
     #[test]
     fn erasure_decode_reader_rejects_invalid_engine_shape() {
         let source = VecStripeSource {
@@ -1689,7 +1752,10 @@ mod tests {
             .pop_front()
             .expect("first stripe should exist");
         let mut source = VecStripeSource {
-            stripes: VecDeque::from([first_state, StripeReadState::new(Vec::new(), erasure.data_shards)]),
+            stripes: VecDeque::from([
+                first_state,
+                StripeReadState::from_parts(Vec::new(), Vec::new(), erasure.data_shards),
+            ]),
             read_quorum: erasure.data_shards,
             read_count: None,
         };
@@ -1724,13 +1790,14 @@ mod tests {
             .stripes
             .pop_front()
             .expect("first stripe should exist");
-        let mut source = VecStripeSource {
+        let mut source = RecordingStripeSource {
             stripes: VecDeque::from([
                 first_state,
-                StripeReadState::new(vec![ShardSlot::data(0, vec![1])], erasure.data_shards),
+                StripeReadState::from_parts(vec![Some(vec![1])], Vec::new(), erasure.data_shards),
             ]),
             read_quorum: erasure.data_shards,
-            read_count: None,
+            reads: 0,
+            recycles: 0,
         };
         let engine = LegacyEcDecodeEngine::new(erasure);
         let mut workspace = engine.prepare_workspace(4).expect("workspace should be prepared");
@@ -1756,6 +1823,8 @@ mod tests {
                 .kind(),
             ErrorKind::Other
         );
+        assert_eq!(source.reads, 2, "the fill must read the primary and queued stripe");
+        assert_eq!(source.recycles, source.reads, "every completed stripe read must be recycled");
     }
 
     #[tokio::test]
@@ -1768,7 +1837,7 @@ mod tests {
                     .stripes
                     .pop_front()
                     .expect("first stripe should exist"),
-                StripeReadState::new(Vec::new(), erasure.data_shards),
+                StripeReadState::from_parts(Vec::new(), Vec::new(), erasure.data_shards),
             ]),
             read_quorum: erasure.data_shards,
             read_count: None,
@@ -2028,17 +2097,11 @@ mod tests {
     }
 
     #[test]
-    fn emit_data_shards_preserves_output_order_for_out_of_order_slots() {
+    fn emit_data_shards_preserves_output_order() {
-        let state = StripeReadState::new(
+        let state =
-            vec![
+            StripeReadState::from_parts(vec![Some(b"ab".to_vec()), Some(b"cd".to_vec()), Some(b"ef".to_vec())], Vec::new(), 2);
-                ShardSlot::data(1, b"cd".to_vec()),
-                ShardSlot::data(0, b"ab".to_vec()),
-                ShardSlot::data(2, b"ef".to_vec()),
-            ],
-            2,
-        );
 
-        let output = emit_data_shards(&state, 3, 6, 5).expect("out-of-order data slots should emit by shard index");
+        let output = emit_data_shards(&state, 3, 6, 5).expect("data slots should emit by shard index");
 
         assert_eq!(output, b"abcde");
     }
@@ -2051,27 +2114,27 @@ mod tests {
         };
         let mut workspace = engine.prepare_workspace(4).expect("workspace should be prepared");
         let mut output = Vec::with_capacity(1);
-        let short_state = StripeReadState::new(vec![ShardSlot::data(0, vec![1, 2, 3, 4])], 1);
+        let mut short_state = StripeReadState::from_parts(vec![Some(vec![1, 2, 3, 4])], Vec::new(), 1);
 
         let err = decode_stripe_into(
             GET_OBJECT_PATH_CODEC_STREAMING,
             false,
             &engine,
             &mut workspace,
-            short_state,
+            &mut short_state,
             8,
             &mut output,
         )
         .expect_err("decoded stripe shorter than data shard count must fail");
         assert_eq!(err.kind(), ErrorKind::UnexpectedEof);
 
-        let missing_state = StripeReadState::from_parts(vec![None, Some(vec![5, 6, 7, 8])], Vec::new(), 1);
+        let mut missing_state = StripeReadState::from_parts(vec![None, Some(vec![5, 6, 7, 8])], Vec::new(), 1);
         let err = decode_stripe_into(
             GET_OBJECT_PATH_CODEC_STREAMING,
             false,
             &engine,
             &mut workspace,
-            missing_state,
+            &mut missing_state,
             8,
             &mut output,
         )
@@ -2082,6 +2145,35 @@ mod tests {
         assert!(output.capacity() >= 32);
     }
 
+    #[test]
+    fn decode_stripe_reconstructs_in_place_without_replacing_slot_storage() {
+        let erasure = Erasure::new(2, 1, 8);
+        let engine = LegacyEcDecodeEngine::new(erasure.clone());
+        let mut workspace = engine.prepare_workspace(4).expect("workspace should be prepared");
+        let encoded = erasure.encode_data(b"abcdefgh").expect("test stripe should encode");
+        let mut shards = encoded.into_iter().map(|shard| Some(shard.to_vec())).collect::<Vec<_>>();
+        shards[0] = None;
+        let mut state = StripeReadState::from_parts(shards, vec![Some(DiskError::FileCorrupt)], 2);
+        let before = state.scratch_storage();
+        let mut output = Vec::new();
+
+        let decoded = decode_stripe_into(
+            GET_OBJECT_PATH_CODEC_STREAMING,
+            false,
+            &engine,
+            &mut workspace,
+            &mut state,
+            8,
+            &mut output,
+        )
+        .expect("degraded stripe should reconstruct");
+
+        assert!(decoded);
+        assert_eq!(output, b"abcdefgh");
+        assert_eq!(state.scratch_storage().0, before.0, "reconstruction must retain shard slot storage");
+        assert_eq!(state.scratch_storage().1, before.1, "unused error storage must not be rebuilt");
+    }
+
     #[tokio::test]
     async fn erasure_decode_reader_reports_short_source() {
         let erasure = Erasure::new(4, 2, 32);

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

@@ -586,9 +586,9 @@ impl Erasure {
             ));
         }
 
-        let shards = self.encode_data_owned(buf)?;
+        let block = self.encode_data_owned_block(buf)?;
         let mut mw = MultiWriter::new(writers, quorum);
-        mw.write(shards).await?;
+        mw.write_block(&block).await?;
         mw.shutdown().await?;
         Ok((reader, total))
     }
@@ -613,13 +613,13 @@ impl Erasure {
             return Ok((reader, 0, Vec::new()));
         }
 
-        let shards = self.encode_data_owned(buf)?;
+        let block = self.encode_data_owned_block(buf)?;
-        let mut inline_shards = Vec::with_capacity(shards.len());
+        let mut inline_shards = Vec::with_capacity(block.shards().len());
-        for shard in shards {
+        for shard in block.shards() {
-            let hash = HashAlgorithm::HighwayHash256S.hash_encode(&shard);
+            let hash = HashAlgorithm::HighwayHash256S.hash_encode(shard);
             let mut encoded = BytesMut::with_capacity(hash.as_ref().len() + shard.len());
             encoded.extend_from_slice(hash.as_ref());
-            encoded.extend_from_slice(&shard);
+            encoded.extend_from_slice(shard);
             inline_shards.push(encoded.freeze());
         }
 
@@ -2164,6 +2164,39 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn cancelling_inline_small_drops_stalled_write() {
+        const BLOCK_SIZE: usize = 16;
+
+        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_plain(
+            StallOnWriteWithSignal {
+                entered: Some(writer_entered_tx),
+                writes: writes.clone(),
+            },
+            BLOCK_SIZE,
+        ))];
+        let erasure = Arc::new(Erasure::new(1, 0, BLOCK_SIZE));
+        let reader = tokio::io::BufReader::new(Cursor::new(vec![0xA5; BLOCK_SIZE - 1]));
+        let encode = tokio::spawn(async move { erasure.encode_inline_small(reader, &mut writers, 1).await });
+
+        tokio::time::timeout(Duration::from_secs(1), writer_entered)
+            .await
+            .expect("inline writer should enter before cancellation")
+            .expect("stalling writer should signal entry");
+        encode.abort();
+        assert!(
+            matches!(encode.await, Err(err) if err.is_cancelled()),
+            "inline encode task should be cancelled"
+        );
+        assert_eq!(
+            writes.load(std::sync::atomic::Ordering::SeqCst),
+            1,
+            "cancellation must drop the stalled write instead of polling it again"
+        );
+    }
+
     #[tokio::test]
     async fn encode_returns_unexpected_eof_for_truncated_limited_reader() {
         let committed = Arc::new(Mutex::new(Vec::new()));
@@ -2395,27 +2428,33 @@ mod tests {
         const DATA_SHARDS: usize = 2;
         const PARITY_SHARDS: usize = 2;
         const BLOCK_SIZE: usize = 64;
-        let payload = b"inline commit payload".to_vec();
         let checksum_algo = HashAlgorithm::HighwayHash256S;
-        let erasure = Arc::new(Erasure::new(DATA_SHARDS, PARITY_SHARDS, BLOCK_SIZE));
+        for uses_legacy in [false, true] {
-        let reader = tokio::io::BufReader::new(Cursor::new(payload.clone()));
+            let erasure = Arc::new(Erasure::new_with_options(DATA_SHARDS, PARITY_SHARDS, BLOCK_SIZE, uses_legacy));
+            for payload in [Vec::new(), vec![0xA5], vec![0x5A; BLOCK_SIZE - 1], vec![0xC3; BLOCK_SIZE]] {
+                let reader = tokio::io::BufReader::new(Cursor::new(payload.clone()));
+                let (_reader, total, inline_shards) = erasure
+                    .clone()
+                    .encode_inline_shards_with_size_hint(reader, payload.len())
+                    .await
+                    .expect("inline shards should encode");
 
-        let (_reader, total, inline_shards) = erasure
+                assert_eq!(total, payload.len());
-            .clone()
+                if payload.is_empty() {
-            .encode_inline_shards_with_size_hint(reader, payload.len())
+                    assert!(inline_shards.is_empty());
-            .await
+                    continue;
-            .expect("inline shards should encode");
+                }
-        let raw_shards = erasure
-            .encode_data_owned(payload.clone())
-            .expect("reference shards should encode");
 
-        assert_eq!(total, payload.len());
+                let raw_shards = erasure.encode_data(&payload).expect("reference shards should encode");
-        assert_eq!(inline_shards.len(), DATA_SHARDS + PARITY_SHARDS);
+                assert_eq!(inline_shards.len(), DATA_SHARDS + PARITY_SHARDS);
-        for (inline, raw) in inline_shards.iter().zip(raw_shards) {
+                for (inline, raw) in inline_shards.iter().zip(raw_shards) {
-            let mut writer = BitrotWriterWrapper::new(CustomWriter::new_inline_buffer(), raw.len(), checksum_algo.clone());
+                    let mut writer =
-            writer.write(&raw).await.expect("reference writer should accept shard");
+                        BitrotWriterWrapper::new(CustomWriter::new_inline_buffer(), raw.len(), checksum_algo.clone());
-            writer.shutdown().await.expect("reference writer should shutdown");
+                    writer.write(&raw).await.expect("reference writer should accept shard");
-            assert_eq!(inline.as_ref(), writer.into_inline_data().expect("reference writer should retain bytes"));
+                    writer.shutdown().await.expect("reference writer should shutdown");
+                    assert_eq!(inline.as_ref(), writer.into_inline_data().expect("reference writer should retain bytes"));
+                }
+            }
         }
     }
 

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

@@ -726,101 +726,37 @@ impl Erasure {
     }
 
     fn encode_data_block_inner(&self, data: &[u8]) -> io::Result<EncodedBlock> {
-        let shard_size_fn = if self.uses_legacy {
+        let mut data_buffer = BytesMut::with_capacity(self.encoded_capacity_for_data_len(data.len()));
-            calc_shard_size_legacy
-        } else {
-            calc_shard_size
-        };
-        let per_shard_size = shard_size_fn(data.len(), self.data_shards);
-        if per_shard_size == 0 {
-            return Ok(EncodedBlock::empty());
-        }
-        let need_total_size = per_shard_size * self.total_shard_count();
-
-        let mut data_buffer = BytesMut::with_capacity(need_total_size);
         data_buffer.extend_from_slice(data);
-        data_buffer.resize(need_total_size, 0u8);
+        self.encode_buffer(data_buffer, data.len())
-
-        {
-            let data_slices: SmallVec<[&mut [u8]; 16]> = data_buffer.chunks_exact_mut(per_shard_size).collect();
-
-            if self.parity_shards > 0 {
-                if self.uses_legacy {
-                    if let Some(encoder) = self.legacy_encoder.as_ref() {
-                        encoder.encode(data_slices)?;
-                    } else {
-                        warn!("parity_shards > 0, uses_legacy but legacy_encoder is None");
-                    }
-                } else if let Some(encoder) = self.encoder.as_ref() {
-                    encoder.encode(data_slices)?;
-                } else {
-                    warn!("parity_shards > 0, but encoder is None");
-                }
-            }
-        }
-
-        Ok(EncodedBlock {
-            data: data_buffer.freeze(),
-            shard_size: per_shard_size,
-        })
     }
 
     /// 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(impl_type = "Erasure")]
     pub fn encode_data_owned(&self, data: Vec<u8>) -> io::Result<Vec<Bytes>> {
-        let shard_size_fn = if self.uses_legacy {
+        self.encode_data_owned_block_inner(data)
-            calc_shard_size_legacy
+            .map(|block| block.into_shards(self.total_shard_count()))
-        } else {
+    }
-            calc_shard_size
-        };
-        let per_shard_size = shard_size_fn(data.len(), self.data_shards);
-        if per_shard_size == 0 {
-            return Ok(vec![Bytes::new(); self.total_shard_count()]);
-        }
-        let need_total_size = per_shard_size * self.total_shard_count();
 
+    #[hotpath::measure(label = "Erasure::encode_data_owned", impl_type = "Erasure")]
+    pub(crate) fn encode_data_owned_block(&self, data: Vec<u8>) -> io::Result<EncodedBlock> {
+        self.encode_data_owned_block_inner(data)
+    }
+
+    fn encode_data_owned_block_inner(&self, data: Vec<u8>) -> io::Result<EncodedBlock> {
+        let data_len = data.len();
         // Try zero-copy: Vec<u8> -> Bytes -> BytesMut (succeeds when refcount == 1)
-        let mut data_buffer = match Bytes::from(data).try_into_mut() {
+        let data_buffer = match Bytes::from(data).try_into_mut() {
-            Ok(mut bm) => {
+            Ok(data_buffer) => data_buffer,
-                bm.resize(need_total_size, 0u8);
-                bm
-            }
             Err(b) => {
                 // Rare path: refcount != 1, fall back to copy
-                let mut bm = BytesMut::with_capacity(need_total_size);
+                let mut data_buffer = BytesMut::with_capacity(self.encoded_capacity_for_data_len(data_len));
-                bm.extend_from_slice(&b);
+                data_buffer.extend_from_slice(&b);
-                bm.resize(need_total_size, 0u8);
+                data_buffer
-                bm
             }
         };
-
+        self.encode_buffer(data_buffer, data_len)
-        {
-            let data_slices: SmallVec<[&mut [u8]; 16]> = data_buffer.chunks_exact_mut(per_shard_size).collect();
-
-            if self.parity_shards > 0 {
-                if self.uses_legacy {
-                    if let Some(encoder) = self.legacy_encoder.as_ref() {
-                        encoder.encode(data_slices)?;
-                    } else {
-                        warn!("parity_shards > 0, uses_legacy but legacy_encoder is None");
-                    }
-                } else if let Some(encoder) = self.encoder.as_ref() {
-                    encoder.encode(data_slices)?;
-                } else {
-                    warn!("parity_shards > 0, but encoder is None");
-                }
-            }
-        }
-
-        let mut data_buffer = data_buffer.freeze();
-        let mut shards = Vec::with_capacity(self.total_shard_count());
-        for _ in 0..self.total_shard_count() {
-            let shard = data_buffer.split_to(per_shard_size);
-            shards.push(shard);
-        }
-
-        Ok(shards)
     }
 
     /// Encode data from an owned `BytesMut` buffer, avoiding the initial copy
@@ -833,16 +769,16 @@ impl Erasure {
     /// `data_len` — so this function never reallocates the buffer.
     #[hotpath::measure(impl_type = "Erasure")]
     pub fn encode_data_bytes_mut(&self, data_buffer: BytesMut, data_len: usize) -> io::Result<Vec<Bytes>> {
-        self.encode_data_bytes_mut_block_inner(data_buffer, data_len)
+        self.encode_buffer(data_buffer, data_len)
             .map(|block| block.into_shards(self.total_shard_count()))
     }
 
     #[hotpath::measure(label = "Erasure::encode_data_bytes_mut", impl_type = "Erasure")]
     pub(crate) fn encode_data_bytes_mut_block(&self, data_buffer: BytesMut, data_len: usize) -> io::Result<EncodedBlock> {
-        self.encode_data_bytes_mut_block_inner(data_buffer, data_len)
+        self.encode_buffer(data_buffer, data_len)
     }
 
-    fn encode_data_bytes_mut_block_inner(&self, mut data_buffer: BytesMut, data_len: usize) -> io::Result<EncodedBlock> {
+    fn encode_buffer(&self, mut data_buffer: BytesMut, data_len: usize) -> io::Result<EncodedBlock> {
         let shard_size_fn = if self.uses_legacy {
             calc_shard_size_legacy
         } else {
@@ -1550,10 +1486,16 @@ mod tests {
     fn encode_data_owned_matches_borrowed_path() {
         for uses_legacy in [false, true] {
             let erasure = Erasure::new_with_options(4, 2, 64, uses_legacy);
-
+            for data in [
-            assert_owned_encode_matches_borrowed(&erasure, Vec::new());
+                Vec::new(),
-            assert_owned_encode_matches_borrowed(&erasure, b"small payload".to_vec());
+                vec![0xA5; 1],
-            assert_owned_encode_matches_borrowed(&erasure, (0_u8..37).collect());
+                b"small payload".to_vec(),
+                (0_u8..37).collect(),
+                vec![0xA5; erasure.block_size - 1],
+                vec![0x5A; erasure.block_size],
+            ] {
+                assert_owned_encode_matches_borrowed(&erasure, data);
+            }
         }
     }
 
@@ -1601,26 +1543,41 @@ mod tests {
 
     #[test]
     fn streaming_encoded_block_uses_one_contiguous_backing_buffer() {
-        let erasure = Erasure::new(8, 8, 64);
+        for uses_legacy in [false, true] {
+            let erasure = Erasure::new_with_options(8, 8, 64, uses_legacy);
 
-        for data_len in [1, 63, 64] {
+            for data_len in [0, 1, 63, 64] {
-            let data = (0..data_len).map(|i| i as u8).collect::<Vec<_>>();
+                let data = (0..data_len).map(|i| i as u8).collect::<Vec<_>>();
-            let expected = erasure.encode_data(&data).expect("public encode should succeed");
+                let expected = erasure.encode_data(&data).expect("public encode should succeed");
-            let borrowed = erasure
+                let borrowed = erasure
-                .encode_data_block(&data)
+                    .encode_data_block(&data)
-                .expect("borrowed streaming encode should succeed");
+                    .expect("borrowed streaming encode should succeed");
-            let owned = erasure
+                let owned = erasure
-                .encode_data_bytes_mut_block(BytesMut::from(&data[..]), data.len())
+                    .encode_data_owned_block(data.clone())
-                .expect("BytesMut streaming encode should succeed");
+                    .expect("owned streaming encode should succeed");
+                let bytes_mut = erasure
+                    .encode_data_bytes_mut_block(BytesMut::from(&data[..]), data.len())
+                    .expect("BytesMut streaming encode should succeed");
 
-            assert!(borrowed.shards().eq(expected.iter().map(Bytes::as_ref)));
+                assert_eq!(borrowed.queued_bytes(), owned.queued_bytes());
-            assert!(owned.shards().eq(expected.iter().map(Bytes::as_ref)));
+                assert_eq!(borrowed.queued_bytes(), bytes_mut.queued_bytes());
-            assert_eq!(borrowed.shards().len(), 16);
-            assert_eq!(borrowed.queued_bytes(), owned.queued_bytes());
 
-            let first = borrowed.shards().next().expect("encoded block should have shards").as_ptr();
+                if data_len == 0 {
-            for (index, shard) in borrowed.shards().enumerate() {
+                    assert!(expected.iter().all(Bytes::is_empty));
-                assert_eq!(shard.as_ptr(), first.wrapping_add(index * shard.len()));
+                    assert!(borrowed.is_empty());
+                    assert!(owned.is_empty());
+                    assert!(bytes_mut.is_empty());
+                    continue;
+                }
+
+                assert!(borrowed.shards().eq(expected.iter().map(Bytes::as_ref)));
+                assert!(owned.shards().eq(expected.iter().map(Bytes::as_ref)));
+                assert!(bytes_mut.shards().eq(expected.iter().map(Bytes::as_ref)));
+                assert_eq!(borrowed.shards().len(), 16);
+                let first = borrowed.shards().next().expect("encoded block should have shards").as_ptr();
+                for (index, shard) in borrowed.shards().enumerate() {
+                    assert_eq!(shard.as_ptr(), first.wrapping_add(index * shard.len()));
+                }
             }
         }
         assert_eq!(

crates/ecstore/src/io_support/bitrot.rs

@@ -56,6 +56,16 @@ pub enum ShardReader {
     Stream(Box<dyn AsyncRead + Send + Sync + Unpin>),
 }
 
+#[cfg(test)]
+impl ShardReader {
+    pub(crate) fn inline_bytes(&self) -> Option<&Bytes> {
+        match self {
+            Self::InMemory(cursor) => Some(cursor.get_ref()),
+            Self::Chunked(_) | Self::Stream(_) => None,
+        }
+    }
+}
+
 impl AsyncRead for ShardReader {
     fn poll_read(self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut tokio::io::ReadBuf<'_>) -> Poll<std::io::Result<()>> {
         match self.get_mut() {
@@ -646,7 +656,7 @@ pub async fn create_bitrot_reader_from_bytes(
 }
 
 #[allow(clippy::too_many_arguments)]
-async fn create_bitrot_reader_from_bytes_with_stage_metrics(
+pub(crate) async fn create_bitrot_reader_from_bytes_with_stage_metrics(
     inline_data: Option<Bytes>,
     disk: Option<&DiskStore>,
     bucket: &str,

crates/ecstore/src/layout/disks_layout.rs

@@ -21,7 +21,8 @@ use tracing::debug;
 
 /// Supported set sizes this is used to find the optimal
 /// single set size.
-const SET_SIZES: [usize; 15] = [2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16];
+pub(crate) const MAX_ERASURE_SET_DRIVE_COUNT: usize = 16;
+const SET_SIZES: [usize; 15] = [2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, MAX_ERASURE_SET_DRIVE_COUNT];
 const ENV_RUSTFS_ERASURE_SET_DRIVE_COUNT: &str = "RUSTFS_ERASURE_SET_DRIVE_COUNT";
 
 #[derive(Deserialize, Debug, Default)]
@@ -327,7 +328,7 @@ fn possible_set_counts(set_size: usize) -> Vec<usize> {
 
 /// checks whether given count is a valid set size for erasure coding.
 fn is_valid_set_size(count: usize) -> bool {
-    count >= SET_SIZES[0] && count <= SET_SIZES[SET_SIZES.len() - 1]
+    count >= SET_SIZES[0] && count <= MAX_ERASURE_SET_DRIVE_COUNT
 }
 
 /// Final set size with all the symmetry accounted for.

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

@@ -54,8 +54,9 @@ use crate::disk::{
 use crate::erasure::coding::BitrotReader;
 use crate::io_support::bitrot::ShardReader;
 use crate::io_support::bitrot::{
-    BitrotReaderStageMetrics, DeferredReaderStripeHandle, adjust_shard_read_params, create_bitrot_reader_with_stage_metrics,
+    BitrotReaderStageMetrics, DeferredReaderStripeHandle, adjust_shard_read_params,
-    create_deferred_bitrot_reader_with_stripe_handle, object_mmap_read_enabled, object_mmap_read_max_length,
+    create_bitrot_reader_from_bytes_with_stage_metrics, create_deferred_bitrot_reader_with_stripe_handle,
+    object_mmap_read_enabled, object_mmap_read_max_length,
 };
 use crate::set_disk::shard_source::ShardReadCost;
 use futures::FutureExt as _;
@@ -1262,13 +1263,13 @@ pub(in crate::set_disk) fn schedule_bitrot_reader_task<'a>(
         return;
     }
 
-    let inline_data = files[idx].data.as_deref();
+    let inline_data = files[idx].data.clone();
     let data_dir = files[idx].data_dir.unwrap_or_default();
     let disk = disks[idx].as_ref();
     let path = format!("{object}/{data_dir}/part.{part_number}");
 
     reader_tasks.push(Box::pin(async move {
-        let result = create_bitrot_reader_with_stage_metrics(
+        let result = create_bitrot_reader_from_bytes_with_stage_metrics(
             inline_data,
             disk,
             bucket,
@@ -1560,14 +1561,14 @@ pub(in crate::set_disk) async fn create_bitrot_readers_until_quorum_all_shards(
     let schedule_stage_start = stage_metrics.map(|_| Instant::now());
     for (idx, disk_op) in disks.iter().enumerate() {
         setup.mark_scheduled(idx);
-        let inline_data = files[idx].data.as_deref();
+        let inline_data = files[idx].data.clone();
         let data_dir = files[idx].data_dir.unwrap_or_default();
         let disk = disk_op.as_ref();
         let path = format!("{object}/{data_dir}/part.{part_number}");
         let checksum_algo = checksum_algo.clone();
 
         reader_tasks.push(async move {
-            let result = create_bitrot_reader_with_stage_metrics(
+            let result = create_bitrot_reader_from_bytes_with_stage_metrics(
                 inline_data,
                 disk,
                 bucket,

crates/ecstore/src/set_disk/mod.rs

@@ -247,8 +247,8 @@ impl SetDisks {
             no_lock: true,
             ..Default::default()
         };
-        let (current, _, _) = self.get_object_fileinfo(bucket, object, &read_opts, true, false).await?;
+        let current = self.get_object_fileinfo(bucket, object, &read_opts, true, false).await?;
-        restore_operation_id_from_metadata(&current.metadata)?
+        restore_operation_id_from_metadata(&current.fi().metadata)?
             .filter(|actual| *actual == expected)
             .ok_or_else(|| Error::other(format!("restore operation id changed before {mode}: expected {expected}")))?;
         Ok(())
@@ -736,41 +736,91 @@ mod transition_matrix_tests;
 
 pub use ops::heal_walk::HealWalkVersion;
 
-pub(in crate::set_disk) enum GetObjectMetadata<T> {
+pub(in crate::set_disk) struct GetObjectFileInfo {
-    Owned(T),
+    owned: Option<OwnedGetObjectFileInfo>,
-    Shared(Arc<T>),
+    shared: Option<Arc<GetObjectMetadataCacheEntry>>,
 }
 
-impl<T> std::ops::Deref for GetObjectMetadata<T> {
+struct OwnedGetObjectFileInfo {
-    type Target = T;
+    fi: FileInfo,
+    parts_metadata: Vec<FileInfo>,
+    online_disks: Vec<Option<DiskStore>>,
+}
 
-    fn deref(&self) -> &Self::Target {
+impl GetObjectFileInfo {
-        match self {
+    fn owned(fi: FileInfo, parts_metadata: Vec<FileInfo>, online_disks: Vec<Option<DiskStore>>) -> Self {
-            Self::Owned(value) => value,
+        Self {
-            Self::Shared(value) => value,
+            owned: Some(OwnedGetObjectFileInfo {
+                fi,
+                parts_metadata,
+                online_disks,
+            }),
+            shared: None,
         }
     }
-}
 
-impl<T: Clone> GetObjectMetadata<T> {
+    fn shared(entry: Arc<GetObjectMetadataCacheEntry>) -> Self {
-    fn into_owned(self) -> T {
+        Self {
-        match self {
+            owned: None,
-            Self::Owned(value) => value,
+            shared: Some(entry),
-            Self::Shared(value) => Arc::try_unwrap(value).unwrap_or_else(|value| (*value).clone()),
         }
     }
-}
 
-type GetObjectFileInfo = (
+    fn fi(&self) -> &FileInfo {
-    GetObjectMetadata<FileInfo>,
+        match (&self.owned, &self.shared) {
-    GetObjectMetadata<Vec<FileInfo>>,
+            (Some(snapshot), None) => &snapshot.fi,
-    GetObjectMetadata<Vec<Option<DiskStore>>>,
+            (None, Some(entry)) => &entry.fi,
-);
+            _ => unreachable!("GET metadata snapshot representation must be exclusive"),
+        }
+    }
+
+    fn parts_metadata(&self) -> &[FileInfo] {
+        match (&self.owned, &self.shared) {
+            (Some(snapshot), None) => &snapshot.parts_metadata,
+            (None, Some(entry)) => &entry.parts_metadata,
+            _ => unreachable!("GET metadata snapshot representation must be exclusive"),
+        }
+    }
+
+    fn online_disks(&self) -> &[Option<DiskStore>] {
+        match (&self.owned, &self.shared) {
+            (Some(snapshot), None) => &snapshot.online_disks,
+            (None, Some(entry)) => &entry.online_disks,
+            _ => unreachable!("GET metadata snapshot representation must be exclusive"),
+        }
+    }
+
+    fn into_owned(self) -> (FileInfo, Vec<FileInfo>, Vec<Option<DiskStore>>) {
+        match (self.owned, self.shared) {
+            (Some(snapshot), None) => {
+                let OwnedGetObjectFileInfo {
+                    fi,
+                    parts_metadata,
+                    online_disks,
+                } = snapshot;
+                (fi, parts_metadata, online_disks)
+            }
+            (None, Some(entry)) => match Arc::try_unwrap(entry) {
+                Ok(entry) => (entry.fi, entry.parts_metadata, entry.online_disks),
+                Err(entry) => (entry.fi.clone(), entry.parts_metadata.clone(), entry.online_disks.clone()),
+            },
+            _ => unreachable!("GET metadata snapshot representation must be exclusive"),
+        }
+    }
+
+    #[cfg(test)]
+    fn has_valid_representation(&self) -> bool {
+        self.owned.is_some() ^ self.shared.is_some()
+    }
+
+    #[cfg(test)]
+    fn shared_entry(&self) -> Option<&Arc<GetObjectMetadataCacheEntry>> {
+        self.shared.as_ref()
+    }
+}
 
 pub(crate) struct PreparedGetObjectMetadata {
-    fi: GetObjectMetadata<FileInfo>,
+    snapshot: GetObjectFileInfo,
-    files: GetObjectMetadata<Vec<FileInfo>>,
-    disks: GetObjectMetadata<Vec<Option<DiskStore>>>,
     object_info: Option<ObjectInfo>,
 }
 
@@ -788,7 +838,7 @@ impl PreparedGetObjectMetadata {
     }
 
     pub(crate) fn read_semantics_identity(&self) -> [u8; 32] {
-        SetDisks::file_info_quorum_hash(&self.fi)
+        SetDisks::file_info_quorum_hash(self.snapshot.fi())
     }
 }
 
@@ -838,10 +888,11 @@ mod prepared_get_object_metadata_tests {
 
     #[tokio::test]
     async fn prepared_metadata_is_consumed_exactly_once() {
+        let snapshot = GetObjectFileInfo::owned(FileInfo::default(), Vec::new(), Vec::new());
+        assert!(snapshot.has_valid_representation());
+        assert!(snapshot.shared_entry().is_none());
         let metadata = PreparedGetObjectMetadata {
-            fi: GetObjectMetadata::Owned(FileInfo::default()),
+            snapshot,
-            files: GetObjectMetadata::Owned(Vec::new()),
-            disks: GetObjectMetadata::Owned(Vec::new()),
             object_info: None,
         };
 
@@ -853,6 +904,44 @@ mod prepared_get_object_metadata_tests {
         assert!(take_prepared_get_object_metadata().is_none());
     }
 
+    #[test]
+    fn cache_hit_consumers_release_snapshot_at_legacy_ownership() {
+        let fi = FileInfo {
+            name: "object".to_owned(),
+            ..Default::default()
+        };
+        let cached = Arc::new(GetObjectMetadataCacheEntry {
+            created_at: Instant::now(),
+            parts_metadata: vec![fi.clone()],
+            fi,
+            online_disks: vec![None],
+            read_quorum: 0,
+        });
+        let snapshot = GetObjectFileInfo::shared(Arc::clone(&cached));
+
+        assert!(snapshot.has_valid_representation());
+        assert!(std::mem::size_of::<GetObjectFileInfo>() >= std::mem::size_of::<OwnedGetObjectFileInfo>());
+        assert!(
+            std::mem::size_of::<GetObjectFileInfo>()
+                <= std::mem::size_of::<OwnedGetObjectFileInfo>() + 2 * std::mem::size_of::<usize>()
+        );
+        assert_eq!(Arc::strong_count(&cached), 2, "a cache hit must add one snapshot reference");
+        assert_eq!(snapshot.fi().name, "object");
+        assert_eq!(snapshot.parts_metadata().len(), 1);
+        assert_eq!(snapshot.online_disks().len(), 1);
+        assert_eq!(Arc::strong_count(&cached), 2, "borrowing consumers must not clone the snapshot");
+
+        let (owned_fi, owned_parts, disks) = snapshot.into_owned();
+        assert_eq!(owned_fi.name, "object");
+        assert_eq!(owned_parts.len(), 1);
+        assert_eq!(disks.len(), 1);
+        assert_eq!(
+            Arc::strong_count(&cached),
+            1,
+            "legacy ownership must release the cache snapshot after cloning its owned inputs"
+        );
+    }
+
     #[tokio::test]
     #[serial_test::serial(body_cache_hook)]
     async fn prepared_reader_reuses_metadata_fanout_exactly_once() {
@@ -1016,12 +1105,10 @@ impl SetDisks {
         object: &str,
         opts: &ObjectOptions,
     ) -> Result<PreparedGetObjectMetadata> {
-        let (fi, files, disks) = self.get_object_fileinfo(bucket, object, opts, true, true).await?;
+        let snapshot = self.get_object_fileinfo(bucket, object, opts, true, true).await?;
-        let object_info = build_get_object_info(&fi, bucket, object, opts.versioned || opts.version_suspended);
+        let object_info = build_get_object_info(snapshot.fi(), bucket, object, opts.versioned || opts.version_suspended);
         Ok(PreparedGetObjectMetadata {
-            fi,
+            snapshot,
-            files,
-            disks,
             object_info: Some(object_info),
         })
     }
@@ -1137,24 +1224,6 @@ pub fn is_deadlock_detection_enabled() -> bool {
 // require process restart to take effect.
 // ============================================================================
 
-/// Check if codec streaming is enabled (base flag).
-///
-/// **Note**: Cached via `OnceLock` — env var changes require process restart.
-/// In test mode, bypasses cache to allow per-test env var overrides.
-fn is_get_codec_streaming_enabled() -> bool {
-    #[cfg(test)]
-    {
-        rustfs_utils::get_env_bool(ENV_RUSTFS_GET_CODEC_STREAMING_ENABLE, DEFAULT_RUSTFS_GET_CODEC_STREAMING_ENABLE)
-    }
-    #[cfg(not(test))]
-    {
-        static CACHED: OnceLock<bool> = OnceLock::new();
-        *CACHED.get_or_init(|| {
-            rustfs_utils::get_env_bool(ENV_RUSTFS_GET_CODEC_STREAMING_ENABLE, DEFAULT_RUSTFS_GET_CODEC_STREAMING_ENABLE)
-        })
-    }
-}
-
 /// Check if multipart codec streaming is enabled.
 ///
 /// When enabled, multipart objects use per-part codec streaming
@@ -1309,22 +1378,6 @@ fn is_multipart_reader_setup_prefetch_enabled() -> bool {
     }
 }
 
-// --- Rollout Percentage Functions ---
-
-fn get_codec_streaming_rollout_pct() -> u32 {
-    #[cfg(test)]
-    {
-        rustfs_utils::get_env_u32(ENV_RUSTFS_GET_CODEC_STREAMING_ROLLOUT_PCT, DEFAULT_RUSTFS_GET_CODEC_STREAMING_ROLLOUT_PCT)
-    }
-    #[cfg(not(test))]
-    {
-        static CACHED: OnceLock<u32> = OnceLock::new();
-        *CACHED.get_or_init(|| {
-            rustfs_utils::get_env_u32(ENV_RUSTFS_GET_CODEC_STREAMING_ROLLOUT_PCT, DEFAULT_RUSTFS_GET_CODEC_STREAMING_ROLLOUT_PCT)
-        })
-    }
-}
-
 fn get_metadata_early_stop_rollout_pct() -> u32 {
     static CACHED: OnceLock<u32> = OnceLock::new();
     *CACHED.get_or_init(|| {
@@ -1359,10 +1412,8 @@ fn is_optimization_enabled_for_request(base_enabled: bool, rollout_pct: u32, buc
     (hash as u32) < rollout_pct
 }
 /// Should this specific request use codec streaming?
-pub fn should_use_codec_streaming(bucket: &str, object: &str) -> bool {
+fn should_use_codec_streaming(config: GetCodecStreamingConfig, bucket: &str, object: &str) -> bool {
-    let base = is_get_codec_streaming_enabled();
+    is_optimization_enabled_for_request(config.enabled, config.rollout_pct, bucket, object)
-    let pct = get_codec_streaming_rollout_pct();
-    is_optimization_enabled_for_request(base, pct, bucket, object)
 }
 
 /// Should this specific request use metadata early-stop?
@@ -1372,20 +1423,6 @@ pub fn should_use_metadata_early_stop(bucket: &str, object: &str) -> bool {
     is_optimization_enabled_for_request(base, pct, bucket, object)
 }
 
-fn get_codec_streaming_min_size() -> usize {
-    if std::env::var_os(ENV_RUSTFS_GET_CODEC_STREAMING_MIN_SIZE).is_some() {
-        return rustfs_utils::get_env_usize(ENV_RUSTFS_GET_CODEC_STREAMING_MIN_SIZE, DEFAULT_RUSTFS_GET_CODEC_STREAMING_MIN_SIZE);
-    }
-
-    match get_codec_streaming_engine() {
-        GetCodecStreamingEngine::Rustfs => rustfs_utils::get_env_usize(
-            ENV_RUSTFS_GET_CODEC_STREAMING_RUSTFS_MIN_SIZE,
-            DEFAULT_RUSTFS_GET_CODEC_STREAMING_RUSTFS_MIN_SIZE,
-        ),
-        GetCodecStreamingEngine::Legacy => DEFAULT_RUSTFS_GET_CODEC_STREAMING_MIN_SIZE,
-    }
-}
-
 fn is_get_codec_streaming_data_blocks_first_enabled() -> bool {
     #[cfg(test)]
     {
@@ -1488,8 +1525,18 @@ enum GetCodecStreamingEngine {
     Rustfs,
 }
 
-fn get_codec_streaming_engine() -> GetCodecStreamingEngine {
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
-    let engine = rustfs_utils::get_env_str(ENV_RUSTFS_GET_CODEC_STREAMING_ENGINE, DEFAULT_RUSTFS_GET_CODEC_STREAMING_ENGINE);
+struct GetCodecStreamingConfig {
+    enabled: bool,
+    rollout: GetCodecStreamingRollout,
+    rollout_pct: u32,
+    body_compat_confirmed: bool,
+    header_compat_confirmed: bool,
+    engine: GetCodecStreamingEngine,
+    min_size: usize,
+}
+
+fn parse_get_codec_streaming_engine(engine: &str) -> GetCodecStreamingEngine {
     match engine.trim() {
         value if value.eq_ignore_ascii_case(GET_CODEC_STREAMING_ENGINE_RUSTFS) => GetCodecStreamingEngine::Rustfs,
         value if value.eq_ignore_ascii_case(GET_CODEC_STREAMING_ENGINE_LEGACY) => GetCodecStreamingEngine::Legacy,
@@ -1497,8 +1544,7 @@ fn get_codec_streaming_engine() -> GetCodecStreamingEngine {
     }
 }
 
-fn get_codec_streaming_rollout() -> GetCodecStreamingRollout {
+fn parse_get_codec_streaming_rollout(rollout: &str) -> GetCodecStreamingRollout {
-    let rollout = rustfs_utils::get_env_str(ENV_RUSTFS_GET_CODEC_STREAMING_ROLLOUT, DEFAULT_RUSTFS_GET_CODEC_STREAMING_ROLLOUT);
     match rollout.trim() {
         // Clean production token. `internal`/`benchmark` remain accepted aliases
         // for backward compatibility; all three opt the fast path in.
@@ -1515,20 +1561,58 @@ fn get_codec_streaming_rollout() -> GetCodecStreamingRollout {
     }
 }
 
-/// Emergency kill-switch (defaults to `true`). Set
+fn load_get_codec_streaming_config() -> GetCodecStreamingConfig {
-/// `RUSTFS_GET_CODEC_STREAMING_BODY_COMPAT_CONFIRMED=false` to force the fast path
+    let engine = parse_get_codec_streaming_engine(&rustfs_utils::get_env_str(
-/// off. Body compatibility is confirmed by the parity e2e net + bench (backlog#1183),
+        ENV_RUSTFS_GET_CODEC_STREAMING_ENGINE,
-/// so this no longer gates enablement — the `..._ROLLOUT` switch does.
+        DEFAULT_RUSTFS_GET_CODEC_STREAMING_ENGINE,
-fn is_get_codec_streaming_body_compat_confirmed() -> bool {
+    ));
-    rustfs_utils::get_env_bool(ENV_RUSTFS_GET_CODEC_STREAMING_BODY_COMPAT_CONFIRMED, true)
+    let min_size = if std::env::var_os(ENV_RUSTFS_GET_CODEC_STREAMING_MIN_SIZE).is_some() {
+        rustfs_utils::get_env_usize(ENV_RUSTFS_GET_CODEC_STREAMING_MIN_SIZE, DEFAULT_RUSTFS_GET_CODEC_STREAMING_MIN_SIZE)
+    } else {
+        match engine {
+            GetCodecStreamingEngine::Rustfs => rustfs_utils::get_env_usize(
+                ENV_RUSTFS_GET_CODEC_STREAMING_RUSTFS_MIN_SIZE,
+                DEFAULT_RUSTFS_GET_CODEC_STREAMING_RUSTFS_MIN_SIZE,
+            ),
+            GetCodecStreamingEngine::Legacy => DEFAULT_RUSTFS_GET_CODEC_STREAMING_MIN_SIZE,
+        }
+    };
+
+    GetCodecStreamingConfig {
+        enabled: rustfs_utils::get_env_bool(ENV_RUSTFS_GET_CODEC_STREAMING_ENABLE, DEFAULT_RUSTFS_GET_CODEC_STREAMING_ENABLE),
+        rollout: parse_get_codec_streaming_rollout(&rustfs_utils::get_env_str(
+            ENV_RUSTFS_GET_CODEC_STREAMING_ROLLOUT,
+            DEFAULT_RUSTFS_GET_CODEC_STREAMING_ROLLOUT,
+        )),
+        rollout_pct: rustfs_utils::get_env_u32(
+            ENV_RUSTFS_GET_CODEC_STREAMING_ROLLOUT_PCT,
+            DEFAULT_RUSTFS_GET_CODEC_STREAMING_ROLLOUT_PCT,
+        ),
+        body_compat_confirmed: rustfs_utils::get_env_bool(ENV_RUSTFS_GET_CODEC_STREAMING_BODY_COMPAT_CONFIRMED, true),
+        header_compat_confirmed: rustfs_utils::get_env_bool(ENV_RUSTFS_GET_CODEC_STREAMING_HEADER_COMPAT_CONFIRMED, true),
+        engine,
+        min_size,
+    }
 }
 
-/// Emergency kill-switch (defaults to `true`). Set
+fn get_codec_streaming_config_cached_core(load: impl FnOnce() -> GetCodecStreamingConfig) -> GetCodecStreamingConfig {
-/// `RUSTFS_GET_CODEC_STREAMING_HEADER_COMPAT_CONFIRMED=false` to force the fast path
+    static CACHED: OnceLock<GetCodecStreamingConfig> = OnceLock::new();
-/// off. Header compatibility is confirmed by the parity e2e net + bench (backlog#1183),
+    *CACHED.get_or_init(load)
-/// so this no longer gates enablement — the `..._ROLLOUT` switch does.
+}
-fn is_get_codec_streaming_header_compat_confirmed() -> bool {
+
-    rustfs_utils::get_env_bool(ENV_RUSTFS_GET_CODEC_STREAMING_HEADER_COMPAT_CONFIRMED, true)
+fn get_codec_streaming_config() -> GetCodecStreamingConfig {
+    #[cfg(test)]
+    {
+        load_get_codec_streaming_config()
+    }
+    #[cfg(not(test))]
+    {
+        get_codec_streaming_config_cached_core(load_get_codec_streaming_config)
+    }
+}
+
+fn get_codec_streaming_engine() -> GetCodecStreamingEngine {
+    get_codec_streaming_config().engine
 }
 
 fn build_get_codec_streaming_decode_engine(erasure: coding::Erasure) -> std::io::Result<CodecStreamingDecodeEngine> {
@@ -1897,43 +1981,43 @@ fn should_prefer_codec_streaming_data_blocks_first_reader_setup(
 fn get_codec_streaming_reader_gate(
     bucket: &str,
     object: &str,
-    range: &Option<HTTPRangeSpec>,
     part_number: Option<usize>,
+    object_class: GetCodecStreamingObjectClass,
     object_info: &ObjectInfo,
     fi: &FileInfo,
     lock_optimization_enabled: bool,
 ) -> GetCodecStreamingGate {
-    let object_class = classify_get_codec_streaming_object_class(range, object_info, fi);
+    let config = get_codec_streaming_config();
 
-    if !is_get_codec_streaming_enabled() {
+    if !config.enabled {
         return GetCodecStreamingGate {
             object_class,
             decision: GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::Disabled),
             prefer_data_blocks_first_reader_setup: false,
         };
     }
-    if !get_codec_streaming_rollout().is_opted_in() {
+    if !config.rollout.is_opted_in() {
         return GetCodecStreamingGate {
             object_class,
             decision: GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::RolloutNotOptedIn),
             prefer_data_blocks_first_reader_setup: false,
         };
     }
-    if !should_use_codec_streaming(bucket, object) {
+    if !should_use_codec_streaming(config, bucket, object) {
         return GetCodecStreamingGate {
             object_class,
             decision: GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::RolloutPctNotSelected),
             prefer_data_blocks_first_reader_setup: false,
         };
     }
-    if !is_get_codec_streaming_body_compat_confirmed() {
+    if !config.body_compat_confirmed {
         return GetCodecStreamingGate {
             object_class,
             decision: GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::BodyCompatibilityUnconfirmed),
             prefer_data_blocks_first_reader_setup: false,
         };
     }
-    if !is_get_codec_streaming_header_compat_confirmed() {
+    if !config.header_compat_confirmed {
         return GetCodecStreamingGate {
             object_class,
             decision: GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::HeaderCompatibilityUnconfirmed),
@@ -2006,7 +2090,7 @@ fn get_codec_streaming_reader_gate(
             };
         }
     }
-    let Ok(min_size) = i64::try_from(get_codec_streaming_min_size()) else {
+    let Ok(min_size) = i64::try_from(config.min_size) else {
         return GetCodecStreamingGate {
             object_class,
             decision: GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::InvalidMinSize),
@@ -2376,6 +2460,7 @@ pub struct SetDisks {
     get_object_metadata_cache_hash_builder: std::collections::hash_map::RandomState,
     get_object_metadata_cache_generations: Arc<[AtomicU64]>,
     pub lockers: Vec<Arc<dyn LockClient>>,
+    shared_lockers: Arc<[Arc<dyn LockClient>]>,
     local_lock_manager: Arc<rustfs_lock::GlobalLockManager>,
     /// Per-instance runtime context (Phase 5, backlog#939).
     ///
@@ -2503,9 +2588,9 @@ impl Hash for GetObjectMetadataCacheKey {
 struct GetObjectMetadataCacheEntry {
     #[allow(dead_code)] // Kept for debugging; moka handles TTL internally
     created_at: Instant,
-    fi: Arc<FileInfo>,
+    fi: FileInfo,
-    parts_metadata: Arc<Vec<FileInfo>>,
+    parts_metadata: Vec<FileInfo>,
-    online_disks: Arc<Vec<Option<DiskStore>>>,
+    online_disks: Vec<Option<DiskStore>>,
     read_quorum: usize,
 }
 
@@ -2772,6 +2857,7 @@ impl SetDisks {
     ) -> Arc<Self> {
         let ctx = instance_ctx;
         let set_lock_namespace: Arc<str> = format!("set-{pool_index}-{set_index}").into();
+        let shared_lockers = Arc::from(lockers.to_vec());
         Arc::new(SetDisks {
             locker_owner,
             disks,
@@ -2794,6 +2880,7 @@ impl SetDisks {
                     .collect::<Vec<_>>(),
             ),
             lockers,
+            shared_lockers,
             // Sourced from the instance context so each instance owns its lock
             // namespace (Phase 5 Slice 3). Single-instance: ctx aliases the
             // process lock-manager singleton, so this is unchanged.
@@ -4999,6 +5086,89 @@ mod tests {
         assert_eq!(Arc::strong_count(&set.set_lock_namespace), before);
     }
 
+    #[tokio::test]
+    async fn new_ns_lock_shares_clients_without_changing_quorum() {
+        let healthy: Arc<dyn LockClient> = Arc::new(LocalClient::with_manager(Arc::new(rustfs_lock::GlobalLockManager::new())));
+        let failing: Arc<dyn LockClient> = Arc::new(FailingClient);
+        let ctx = Arc::new(InstanceContext::new());
+        ctx.update_erasure_type(SetupType::DistErasure).await;
+        let set = make_test_set_disks_with_ctx(vec![healthy.clone(), failing.clone()], ctx).await;
+
+        assert!(Arc::ptr_eq(&set.lockers[0], &healthy));
+        assert!(Arc::ptr_eq(&set.lockers[1], &failing));
+        let clients_before = Arc::strong_count(&set.shared_lockers);
+        let healthy_before = Arc::strong_count(&healthy);
+        let failing_before = Arc::strong_count(&failing);
+        let write_lock = set
+            .new_ns_lock("bucket", "write-object")
+            .await
+            .expect("namespace lock should be created");
+
+        assert_eq!(
+            Arc::strong_count(&set.shared_lockers),
+            clients_before + 1,
+            "each object lock should share one client slice allocation"
+        );
+        assert_eq!(
+            Arc::strong_count(&healthy),
+            healthy_before,
+            "constructing an object lock must not clone each client Arc"
+        );
+        assert_eq!(
+            Arc::strong_count(&failing),
+            failing_before,
+            "constructing an object lock must not clone each client Arc"
+        );
+
+        let write_error = write_lock
+            .get_write_lock(Duration::from_millis(500))
+            .await
+            .expect_err("one healthy client must not satisfy the two-client write quorum");
+        assert!(
+            matches!(
+                write_error,
+                LockError::QuorumNotReached {
+                    required: 2,
+                    achieved: 1
+                }
+            ),
+            "the shared client representation must preserve the exact write quorum result: {write_error}"
+        );
+        let read_lock = set
+            .new_ns_lock("bucket", "read-object")
+            .await
+            .expect("second namespace lock should be created");
+        assert_eq!(Arc::strong_count(&set.shared_lockers), clients_before + 2);
+        let read_guard = read_lock
+            .get_read_lock(Duration::from_millis(500))
+            .await
+            .expect("one healthy client should satisfy the two-client read quorum");
+        assert!(matches!(read_guard, NamespaceLockGuard::Standard(_)));
+    }
+
+    #[tokio::test]
+    async fn new_ns_lock_uses_the_current_public_client_domain() {
+        let stale_a: Arc<dyn LockClient> = Arc::new(FailingClient);
+        let stale_b: Arc<dyn LockClient> = Arc::new(FailingClient);
+        let healthy_a: Arc<dyn LockClient> = Arc::new(LocalClient::with_manager(Arc::new(rustfs_lock::GlobalLockManager::new())));
+        let healthy_b: Arc<dyn LockClient> = Arc::new(LocalClient::with_manager(Arc::new(rustfs_lock::GlobalLockManager::new())));
+        let ctx = Arc::new(InstanceContext::new());
+        ctx.update_erasure_type(SetupType::DistErasure).await;
+        let set = make_test_set_disks_with_ctx(vec![stale_a, stale_b], ctx).await;
+        let mut set = (*set).clone();
+        set.lockers = vec![healthy_a, healthy_b];
+
+        let lock = set
+            .new_ns_lock("bucket", "object")
+            .await
+            .expect("namespace lock should use the current public clients");
+        let guard = lock
+            .get_write_lock(Duration::from_millis(500))
+            .await
+            .expect("the current healthy clients should satisfy the two-client quorum");
+        assert!(matches!(guard, NamespaceLockGuard::Standard(_)));
+    }
+
     struct SetupTypeGuard {
         previous: SetupType,
     }

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

@@ -3171,10 +3171,11 @@ mod heal_result_report_tests {
             .await
             .expect("object should be written");
 
-        let (fi, _, _) = set
+        let snapshot = set
             .get_object_fileinfo(bucket, object, &opts, true, false)
             .await
             .expect("object metadata should resolve");
+        let fi = snapshot.fi();
         assert_eq!(fi.erasure.parity_blocks, 0);
         let data_dir = fi.data_dir.expect("non-inline object should have a data directory");
         let part_path = dir.path().join(bucket).join(object).join(data_dir.to_string()).join("part.1");

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

@@ -36,10 +36,21 @@ impl crate::storage_api_contracts::namespace::NamespaceLocking for SetDisks {
         // test's transient DistErasure window) would push this set's namespace
         // locking onto its own — possibly empty — dist locker list.
         let set_lock = if self.ctx.is_dist_erasure().await {
-            // Calculate quorum based on lockers count (majority)
+            let lockers = if self.lockers.len() == self.shared_lockers.len()
-            let lockers_count = self.lockers.len();
+                && self
+                    .lockers
+                    .iter()
+                    .zip(self.shared_lockers.iter())
+                    .all(|(current, shared)| Arc::ptr_eq(current, shared))
+            {
+                self.shared_lockers.clone()
+            } else {
+                Arc::from(self.lockers.clone())
+            };
+            // Calculate quorum from the exact client domain used by this lock.
+            let lockers_count = lockers.len();
             let write_quorum = if lockers_count > 1 { (lockers_count / 2) + 1 } else { 1 };
-            NamespaceLock::with_clients_and_quorum_shared(self.set_lock_namespace.clone(), self.lockers.clone(), write_quorum)
+            NamespaceLock::with_clients_and_quorum_shared(self.set_lock_namespace.clone(), lockers, write_quorum)
         } else {
             NamespaceLock::with_local_manager_shared(self.set_lock_namespace.clone(), self.local_lock_manager.clone())
         };

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

@@ -380,7 +380,7 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks {
     type GetObjectReader = GetObjectReader;
     type PutObjectReader = PutObjReader;
 
-    #[tracing::instrument(level = "debug", skip(self))]
+    #[tracing::instrument(level = "debug", skip(self, h))]
     async fn get_object_reader(
         &self,
         bucket: &str,
@@ -429,11 +429,11 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks {
         };
 
         let metadata_stage_start = Instant::now();
-        let (fi, files, disks, prepared_object_info) = if let Some(prepared) = take_prepared_get_object_metadata() {
+        let (snapshot, prepared_object_info) = if let Some(prepared) = take_prepared_get_object_metadata() {
-            (prepared.fi, prepared.files, prepared.disks, prepared.object_info)
+            (prepared.snapshot, prepared.object_info)
         } else {
             match self.get_object_fileinfo(bucket, object, opts, true, true).await {
-                Ok((fi, files, disks)) => (fi, files, disks, None),
+                Ok(snapshot) => (snapshot, None),
                 Err(err) => {
                     rustfs_io_metrics::record_get_object_metadata_phase_duration(metadata_stage_start.elapsed().as_secs_f64());
                     let failure_path = if is_meta_bucketname(bucket) {
@@ -446,10 +446,13 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks {
                 }
             }
         };
+        let fi = snapshot.fi();
+        let files = snapshot.parts_metadata();
+        let disks = snapshot.online_disks();
         let object_info_stage_start = get_stage_timer_if_enabled(stage_metrics_enabled);
         let object_info = prepared_object_info
-            .unwrap_or_else(|| build_get_object_info(&fi, bucket, object, opts.versioned || opts.version_suspended));
+            .unwrap_or_else(|| build_get_object_info(fi, bucket, object, opts.versioned || opts.version_suspended));
-        let object_class = classify_get_codec_streaming_object_class(&range, &object_info, &fi);
+        let object_class = classify_get_codec_streaming_object_class(&range, &object_info, fi);
         let size_bucket = rustfs_io_metrics::get_object_size_bucket(object_info.size);
         record_get_stage_duration_if_enabled(GET_OBJECT_PATH_SET_DISK, GET_STAGE_OBJECT_INFO, object_info_stage_start);
         let metadata_elapsed = metadata_stage_start.elapsed().as_secs_f64();
@@ -497,7 +500,7 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks {
         // Uses the shared predicate from ObjectInfo; additionally checks that
         // inline data is actually present and neither range nor partNumber is
         // in flight.
-        if should_use_inline_fast_path(&range, &object_info, &fi, opts) {
+        if should_use_inline_fast_path(&range, &object_info, fi, opts) {
             let mut inline_prepare_stage_start = get_stage_timer_if_enabled(stage_metrics_enabled);
             let data_shards = fi.erasure.data_blocks;
 
@@ -513,7 +516,7 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks {
                 };
 
             if can_try_inline_data_shards_direct(object_size, fi.erasure.block_size)
-                && let Some(data_files) = collect_inline_data_shard_fileinfos_by_index(&files, &fi, data_shards, |index| {
+                && let Some(data_files) = collect_inline_data_shard_fileinfos_by_index(files, fi, data_shards, |index| {
                     disks.get(index).is_some_and(Option::is_some)
                 })
             {
@@ -581,10 +584,10 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks {
                 }
             }
 
-            let erasure = erasure_from_file_info(&fi, fi.uses_legacy_checksum)?;
+            let erasure = erasure_from_file_info(fi, fi.uses_legacy_checksum)?;
             let read_length = erasure.shard_file_offset(0, object_size, object_size);
             let total_shards = data_shards + fi.erasure.parity_blocks;
-            let (_disks, files) = Self::shuffle_disks_and_parts_metadata_by_index(&disks, &files, &fi);
+            let (_disks, files) = Self::shuffle_disks_and_parts_metadata_by_index(disks, files, fi);
 
             // Check if we have enough inline data shards
             let inline_count = files
@@ -663,10 +666,10 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks {
         let codec_streaming_gate = get_codec_streaming_reader_gate(
             bucket,
             object,
-            &range,
             opts.part_number,
+            object_class,
             &object_info,
-            &fi,
+            fi,
             lock_optimization_enabled,
         );
         record_get_stage_duration_if_enabled(GET_OBJECT_PATH_SET_DISK, GET_STAGE_PATH_DECISION, path_decision_stage_start);
@@ -744,15 +747,15 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks {
             }
         }
 
-        let direct_memory_decision = get_small_object_direct_memory_decision(&range, &object_info, &fi, opts);
+        let direct_memory_decision = get_small_object_direct_memory_decision(&range, &object_info, fi, opts);
         record_get_direct_memory_decision(object_class, direct_memory_decision, size_bucket);
         if let GetDirectMemoryDecision::Use { object_size } = direct_memory_decision {
             if let Some(body) = Self::try_get_object_direct_data_shards_with_fileinfo(
                 bucket,
                 object,
-                &fi,
+                fi,
-                &files,
+                files,
-                &disks,
+                disks,
                 opts.skip_verify_bitrot,
                 object_class.as_str(),
                 size_bucket,
@@ -781,14 +784,15 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks {
             }
 
             let mut output = Vec::with_capacity(object_size);
+            let (fi, files, disks) = snapshot.into_owned();
             Self::get_object_with_fileinfo(
                 bucket,
                 object,
                 0,
                 object_info.size,
                 &mut output,
-                fi.into_owned(),
+                fi,
-                files.into_owned(),
+                files,
                 &disks,
                 self.set_index,
                 self.pool_index,
@@ -827,9 +831,9 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks {
                 match Self::get_object_decode_reader_with_fileinfo(
                     bucket,
                     object,
-                    &fi,
+                    fi,
-                    &files,
+                    files,
-                    &disks,
+                    disks,
                     self.set_index,
                     self.pool_index,
                     opts.skip_verify_bitrot,
@@ -891,6 +895,7 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks {
         let set_index = self.set_index;
         let pool_index = self.pool_index;
         let skip_verify = opts.skip_verify_bitrot;
+        let (fi, files, disks) = snapshot.into_owned();
         tokio::spawn(async move {
             let _guard = read_lock_guard;
             let mut writer = GetObjectDownstreamWriter::new(wd);
@@ -904,8 +909,8 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks {
                 offset,
                 length,
                 &mut writer,
-                fi.into_owned(),
+                fi,
-                files.into_owned(),
+                files,
                 &disks,
                 set_index,
                 pool_index,
@@ -3526,10 +3531,10 @@ impl SetDisks {
         // quorum, failing write quorum on update_object_meta (backlog#872).
         let mut read_opts = opts.clone();
         read_opts.include_part_checksums = true;
-        let (fi, _, disks) = self
+        let (mut fi, _, disks) = self
             .get_object_fileinfo_gated(bucket, object, &read_opts, false, false)
-            .await?;
+            .await?
-        let mut fi = fi.into_owned();
+            .into_owned();
 
         fi.metadata.insert(AMZ_OBJECT_TAGGING.to_owned(), tags.to_owned());
         if let Some(eval_metadata) = &opts.eval_metadata {
@@ -4807,12 +4812,12 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
         // Use the same full xl.meta read path as GetObject metadata resolution.
         // This avoids HEAD/GetObject metadata visibility skew immediately after
         // PutObject/CompleteMultipartUpload.
-        let (fi, _, _) = self
+        let snapshot = self
             .get_object_fileinfo(bucket, object, opts, true, false)
             .await
             .map_err(|e| to_object_err(e, vec![bucket, object]))?;
 
-        let oi = ObjectInfo::from_file_info(&fi, bucket, object, opts.versioned || opts.version_suspended);
+        let oi = ObjectInfo::from_file_info(snapshot.fi(), bucket, object, opts.versioned || opts.version_suspended);
 
         Ok(oi)
     }
@@ -4980,10 +4985,10 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
 
         let mut transition_read_opts = opts.clone();
         transition_read_opts.include_part_checksums = true;
-        let (fi, meta_arr, online_disks) = self
+        let (mut fi, meta_arr, online_disks) = self
             .get_object_fileinfo(bucket, object, &transition_read_opts, true, false)
-            .await?;
+            .await?
-        let mut fi = fi.into_owned();
+            .into_owned();
         /*if err != nil {
             return Err(to_object_err(err, vec![bucket, object]));
         }*/
@@ -5098,7 +5103,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
                 cloned_fi.size,
                 &mut writer,
                 cloned_fi,
-                meta_arr.into_owned(),
+                meta_arr,
                 &online_disks,
                 set_index,
                 pool_index,
@@ -5223,7 +5228,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
         };
         self.invalidate_get_object_metadata_cache(bucket, object).await;
         let current = self.get_object_fileinfo(bucket, object, &commit_opts, true, false).await;
-        let (current_fi, _, _) = match current {
+        let current = match current {
             Ok(current) => current,
             Err(err) => {
                 drop(transition_lock_guard);
@@ -5234,7 +5239,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
                 return Err(err);
             }
         };
-        let mut current_fi = current_fi.into_owned();
+        let (mut current_fi, _, _) = current.into_owned();
         let source_matches = current_fi.version_id == fi.version_id
             && current_fi.data_dir == fi.data_dir
             && current_fi.mod_time == fi.mod_time
@@ -5465,9 +5470,10 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
         if let Err(err) = fi {
             return set_restore_header_fn(&mut oi, Some(to_object_err(err, vec![bucket, object]))).await;
         }
-        let (actual_fi, _, _) = fi?;
+        let actual = fi?;
+        let actual_fi = actual.fi();
 
-        oi = ObjectInfo::from_file_info(&actual_fi, bucket, object, opts.versioned || opts.version_suspended);
+        oi = ObjectInfo::from_file_info(actual_fi, bucket, object, opts.versioned || opts.version_suspended);
         let expected_operation_id = restore_operation_id_from_metadata(&opts.user_defined)?;
         if let Some(expected_operation_id) = expected_operation_id {
             require_restore_operation_id(oi.user_defined.as_ref(), expected_operation_id)?;
@@ -5745,8 +5751,40 @@ mod object_encryption_resolver_wiring_tests {
     use super::*;
     use crate::object_api::{EncryptionResolutionError, ObjectEncryptionResolver, ReadEncryptionMaterial, ReadEncryptionRequest};
     use std::io::Cursor;
+    use std::sync::Mutex;
     use std::sync::atomic::{AtomicUsize, Ordering};
 
+    #[derive(Clone, Default)]
+    struct CapturedLogs(Arc<Mutex<Vec<u8>>>);
+
+    struct CapturedLogWriter(Arc<Mutex<Vec<u8>>>);
+
+    impl CapturedLogs {
+        fn contents(&self) -> String {
+            String::from_utf8(self.0.lock().expect("captured logs mutex should not poison").clone())
+                .expect("captured logs should be valid UTF-8")
+        }
+    }
+
+    impl std::io::Write for CapturedLogWriter {
+        fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
+            let mut captured = self.0.lock().expect("captured logs mutex should not poison");
+            std::io::Write::write(&mut *captured, buf)
+        }
+
+        fn flush(&mut self) -> std::io::Result<()> {
+            Ok(())
+        }
+    }
+
+    impl<'writer> tracing_subscriber::fmt::MakeWriter<'writer> for CapturedLogs {
+        type Writer = CapturedLogWriter;
+
+        fn make_writer(&'writer self) -> Self::Writer {
+            CapturedLogWriter(Arc::clone(&self.0))
+        }
+    }
+
     struct CountingResolver {
         calls: AtomicUsize,
     }
@@ -5793,6 +5831,34 @@ mod object_encryption_resolver_wiring_tests {
         assert!(result.is_err(), "resolver returning no material must fail closed");
         assert_eq!(resolver.calls.load(Ordering::Relaxed), 1);
     }
+
+    #[tokio::test(flavor = "current_thread")]
+    async fn get_object_reader_span_never_records_transport_headers() {
+        use super::hermetic_set_disks_support::hermetic_set_disks_isolated;
+        use crate::storage_api_contracts::object::ObjectIO as _;
+        use rustfs_utils::http::headers::SSEC_KEY_HEADER;
+
+        let logs = CapturedLogs::default();
+        let subscriber = tracing_subscriber::fmt()
+            .with_max_level(tracing::Level::DEBUG)
+            .with_writer(logs.clone())
+            .with_ansi(false)
+            .without_time()
+            .finish();
+        let _guard = tracing::subscriber::set_default(subscriber);
+        let (_temp_dirs, _disks, set_disks) = hermetic_set_disks_isolated(4).await;
+        let mut headers = HeaderMap::new();
+        headers.insert(http::header::AUTHORIZATION, HeaderValue::from_static("credential-must-not-be-logged"));
+        headers.insert(SSEC_KEY_HEADER, HeaderValue::from_static("customer-key-must-not-be-logged"));
+
+        let _ = set_disks
+            .get_object_reader("missing-bucket", "missing-object", None, headers, &ObjectOptions::default())
+            .await;
+
+        let captured = logs.contents();
+        assert!(!captured.contains("credential-must-not-be-logged"));
+        assert!(!captured.contains("customer-key-must-not-be-logged"));
+    }
 }
 
 #[cfg(test)]
@@ -6550,9 +6616,9 @@ mod metadata_mutation_generation_tests {
                 false,
             )
             .await
-            .expect("object metadata should be readable before adding the checksum sidecar");
+            .expect("object metadata should be readable before adding the checksum sidecar")
-        let mut fi = fi.into_owned();
+            .into_owned();
-        let disks = disks.into_owned();
+        let mut fi = fi;
         rustfs_utils::http::insert_str(&mut fi.metadata, rustfs_utils::http::SUFFIX_PART_CHECKSUMS, value.to_string());
         set_disks
             .update_object_meta(bucket, object, fi, &disks)
@@ -6709,9 +6775,9 @@ mod metadata_mutation_generation_tests {
                 false,
             )
             .await
-            .expect("conflicting object metadata should be readable before corruption is injected");
+            .expect("conflicting object metadata should be readable before corruption is injected")
-        let mut fi = fi.into_owned();
+            .into_owned();
-        let disks = disks.into_owned();
+        let mut fi = fi;
         let rustfs_key = format!(
             "{}{}",
             rustfs_utils::http::RUSTFS_INTERNAL_PREFIX,
@@ -6874,9 +6940,9 @@ mod transition_commit_failure_tests {
                 false,
             )
             .await
-            .expect("source metadata should be readable before adding the checksum sidecar");
+            .expect("source metadata should be readable before adding the checksum sidecar")
-        let mut source_fi = source_fi.into_owned();
+            .into_owned();
-        let online_disks = online_disks.into_owned();
+        let mut source_fi = source_fi;
         rustfs_utils::http::insert_str(
             &mut source_fi.metadata,
             rustfs_utils::http::SUFFIX_PART_CHECKSUMS,
@@ -7150,10 +7216,11 @@ mod transition_commit_failure_tests {
             .put_object(bucket, object, &mut reader, &ObjectOptions::default())
             .await
             .expect("source object should be written");
-        let (fi, parts_metadata, online_disks) = set_disks
+        let snapshot = set_disks
             .get_object_fileinfo(bucket, object, &ObjectOptions::default(), true, false)
             .await
             .expect("source metadata should resolve");
+        let (fi, parts_metadata, online_disks) = snapshot.into_owned();
         let generation = set_disks
             .get_object_metadata_cache_generation(bucket, object)
             .expect("metadata cache generation should be active");
@@ -7164,9 +7231,9 @@ mod transition_commit_failure_tests {
                 cache_key.clone(),
                 Arc::new(GetObjectMetadataCacheEntry {
                     created_at: Instant::now(),
-                    fi: Arc::new((*fi).clone()),
+                    fi,
-                    parts_metadata: Arc::new(parts_metadata.into_owned()),
+                    parts_metadata,
-                    online_disks: Arc::new(online_disks.into_owned()),
+                    online_disks,
                     read_quorum: 2,
                 }),
             )
@@ -7293,7 +7360,7 @@ mod transition_commit_failure_tests {
         let result = transition.await.expect("transition task should not panic");
 
         result.expect_err("partial local commit must fail when only two of four disks are writable");
-        let fi = set_disks
+        let snapshot = set_disks
             .get_object_fileinfo(
                 bucket,
                 object,
@@ -7305,10 +7372,10 @@ mod transition_commit_failure_tests {
                 false,
             )
             .await
-            .expect("rollback should keep the source metadata readable on applied disks")
+            .expect("rollback should keep the source metadata readable on applied disks");
-            .0;
         assert_ne!(
-            fi.transition_status, TRANSITION_COMPLETE,
+            snapshot.fi().transition_status,
+            TRANSITION_COMPLETE,
             "rollback must not leave the applied disks marked as transitioned"
         );
         let mut restored = Vec::new();
@@ -8443,7 +8510,7 @@ mod transition_upload_integrity_tests {
             .await
             .expect("local source should drain after failed transition");
         assert_eq!(restored, payload);
-        let (fi, _, _) = set_disks
+        let snapshot = set_disks
             .get_object_fileinfo(
                 bucket,
                 object,
@@ -8457,7 +8524,7 @@ mod transition_upload_integrity_tests {
             )
             .await
             .expect("local source metadata should remain available");
-        assert_ne!(fi.transition_status, TRANSITION_COMPLETE);
+        assert_ne!(snapshot.fi().transition_status, TRANSITION_COMPLETE);
     }
 
     async fn write_source(
@@ -8554,7 +8621,8 @@ mod transition_upload_integrity_tests {
                 false,
             )
             .await
-            .expect("the existing target should remain readable");
+            .expect("the existing target should remain readable")
+            .into_owned();
         assert_ne!(stored.transition_status, TRANSITION_COMPLETE);
         assert!(stored.transition_version.is_none());
     }
@@ -8626,9 +8694,9 @@ mod transition_upload_integrity_tests {
                 false,
             )
             .await
-            .expect("source metadata should be readable");
+            .expect("source metadata should be readable")
-        let mut source_fi = source_fi.into_owned();
+            .into_owned();
-        let online_disks = online_disks.into_owned();
+        let mut source_fi = source_fi;
         rustfs_utils::http::insert_str(
             &mut source_fi.metadata,
             rustfs_utils::http::SUFFIX_PART_CHECKSUMS,
@@ -8655,7 +8723,7 @@ mod transition_upload_integrity_tests {
             remote_object.starts_with(crate::bucket::lifecycle::transition_transaction::TRANSITION_TRANSACTION_PREFIX),
             "remote object should be transaction-scoped: {remote_object}"
         );
-        let (fi, _, _) = set_disks
+        let snapshot = set_disks
             .get_object_fileinfo(
                 bucket,
                 object,
@@ -8670,6 +8738,7 @@ mod transition_upload_integrity_tests {
             )
             .await
             .expect("committed transition metadata should be readable");
+        let fi = snapshot.fi();
         assert_eq!(fi.transition_status, TRANSITION_COMPLETE);
         assert_eq!(fi.transitioned_objname, *remote_object);
         assert_eq!(
@@ -9104,7 +9173,7 @@ mod transition_upload_integrity_tests {
             let object = format!("{}-corrupt.bin", position.label());
             let payload = vec![0x41; 2 * 1024 * 1024];
             let original = write_source(&set_disks, &disk_stores, &bucket, &object, &payload).await;
-            let (source, _, _) = set_disks
+            let source = set_disks
                 .get_object_fileinfo(
                     &bucket,
                     &object,
@@ -9118,6 +9187,7 @@ mod transition_upload_integrity_tests {
                 )
                 .await
                 .expect("source metadata should be available before shard corruption");
+            let source = source.fi();
             let data_dir = source.data_dir.expect("source object should have a data directory");
 
             corrupt_beyond_read_quorum(&temp_dirs, &bucket, &object, data_dir, source.erasure.parity_blocks, position).await;
@@ -9137,7 +9207,7 @@ mod transition_upload_integrity_tests {
                 ),
                 "{position:?}: unexpected transition producer error: {error:?}"
             );
-            let (after, _, _) = set_disks
+            let after = set_disks
                 .get_object_fileinfo(
                     &bucket,
                     &object,
@@ -9151,6 +9221,7 @@ mod transition_upload_integrity_tests {
                 )
                 .await
                 .expect("failed transition must leave metadata readable");
+            let after = after.fi();
             assert_eq!(
                 after.data_dir,
                 Some(data_dir),
@@ -9197,7 +9268,7 @@ mod transition_upload_integrity_tests {
             .transition_object(bucket, object, &transition_options(&original, tier_name))
             .await
             .expect("an unversioned remote version must commit");
-        let (fi, _, _) = set_disks
+        let snapshot = set_disks
             .get_object_fileinfo(
                 bucket,
                 object,
@@ -9211,6 +9282,7 @@ mod transition_upload_integrity_tests {
             )
             .await
             .expect("committed unversioned transition metadata should be readable");
+        let fi = snapshot.fi();
         assert_eq!(fi.transition_version_id, None);
         assert_eq!(fi.transition_version, None);
         assert_eq!(fi.transition_version_state, rustfs_filemeta::TransitionVersionState::KnownDisabled);
@@ -9418,7 +9490,7 @@ mod transition_upload_integrity_tests {
             matches!(error, StorageError::NamespaceLockQuorumUnavailable { .. }),
             "unexpected tagging lock-lost error: {error:?}"
         );
-        let (fi, _, _) = set_disks
+        let snapshot = set_disks
             .get_object_fileinfo(
                 bucket,
                 object,
@@ -9433,7 +9505,7 @@ mod transition_upload_integrity_tests {
             .await
             .expect("source metadata should remain readable");
         assert!(
-            !fi.metadata.contains_key(AMZ_OBJECT_TAGGING),
+            !snapshot.fi().metadata.contains_key(AMZ_OBJECT_TAGGING),
             "a stale tagging writer must not write metadata after refresh-quorum loss"
         );
     }
@@ -9629,13 +9701,14 @@ mod transition_source_identity_matrix_tests {
                 .put_object(bucket, &object, &mut reader, &source_opts)
                 .await
                 .expect("source object should be written");
-            let (source, _, _) = set_disks
+            let source = set_disks
                 .get_object_fileinfo(bucket, &object, &source_opts, true, false)
                 .await
                 .expect("source metadata should resolve");
+            let source = source.fi();
             assert_eq!(source.version_id, Some(source_version_id));
             assert_eq!(
-                transition_source_identity(bucket, &object, &source, &source_opts, &get_raw_etag(&source.metadata))
+                transition_source_identity(bucket, &object, source, &source_opts, &get_raw_etag(&source.metadata))
                     .expect("persisted versioned source identity should build")
                     .version_mode,
                 TransitionSourceVersionMode::Versioned
@@ -9683,10 +9756,11 @@ mod transition_source_identity_matrix_tests {
                 versioned: true,
                 ..Default::default()
             };
-            let (persisted, _, _) = set_disks
+            let persisted = set_disks
                 .get_object_fileinfo(bucket, &object, &persisted_opts, true, false)
                 .await
                 .expect("drifted source metadata should resolve");
+            let persisted = persisted.fi();
             put_barrier.release();
 
             let result = transition.await.expect("transition task should not panic");
@@ -10706,9 +10780,9 @@ mod put_object_tags_early_stop_regression_tests {
                         false,
                     )
                     .await
-                    .expect("object metadata should be readable before adding the checksum sidecar");
+                    .expect("object metadata should be readable before adding the checksum sidecar")
-                let mut fi = fi.into_owned();
+                    .into_owned();
-                let disks = disks.into_owned();
+                let mut fi = fi;
                 rustfs_utils::http::insert_str(
                     &mut fi.metadata,
                     rustfs_utils::http::SUFFIX_PART_CHECKSUMS,

crates/ecstore/src/set_disk/read.rs

@@ -180,9 +180,9 @@ impl SetDisks {
         let key = GetObjectMetadataCacheKey::new(bucket, object, generation);
         let entry = Arc::new(GetObjectMetadataCacheEntry {
             created_at: Instant::now(),
-            fi: Arc::new(fi.clone()),
+            fi: fi.clone(),
-            parts_metadata: Arc::new(parts_metadata.to_vec()),
+            parts_metadata: parts_metadata.to_vec(),
-            online_disks: Arc::new(online_disks.to_vec()),
+            online_disks: online_disks.to_vec(),
             read_quorum,
         });
         self.insert_get_object_metadata_cache_entry_after_insert(key, generation, entry, || {})
@@ -300,11 +300,7 @@ impl SetDisks {
                         GET_STAGE_METADATA_CACHE_LOOKUP,
                         metadata_cache_lookup_start,
                     );
-                    return Ok((
+                    return Ok(GetObjectFileInfo::shared(cached));
-                        GetObjectMetadata::Shared(Arc::clone(&cached.fi)),
-                        GetObjectMetadata::Shared(Arc::clone(&cached.parts_metadata)),
-                        GetObjectMetadata::Shared(Arc::clone(&cached.online_disks)),
-                    ));
                 }
                 MetadataCacheLookup::Miss => {
                     rustfs_io_metrics::record_get_object_metadata_cache_decision(
@@ -436,11 +432,7 @@ impl SetDisks {
 
         // let online_disks: Vec<Option<DiskStore>> = op_online_disks.iter().filter(|v| v.is_some()).cloned().collect();
 
-        Ok((
+        Ok(GetObjectFileInfo::owned(fi, parts_metadata, op_online_disks))
-            GetObjectMetadata::Owned(fi),
-            GetObjectMetadata::Owned(parts_metadata),
-            GetObjectMetadata::Owned(op_online_disks),
-        ))
     }
 
     #[hotpath::measure(impl_type = "SetDisks")]
@@ -450,14 +442,15 @@ impl SetDisks {
         object: &str,
         opts: &ObjectOptions,
     ) -> (ObjectInfo, usize, Option<StorageError>) {
-        let fi = match self.get_object_fileinfo(bucket, object, opts, false, false).await {
+        let snapshot = match self.get_object_fileinfo(bucket, object, opts, false, false).await {
-            Ok((fi, _, _)) => fi,
+            Ok(snapshot) => snapshot,
             Err(e) => return (ObjectInfo::default(), 0, Some(e)),
         };
+        let fi = snapshot.fi();
 
         let write_quorum = fi.write_quorum(self.default_write_quorum());
 
-        let oi = ObjectInfo::from_file_info(&fi, bucket, object, opts.versioned || opts.version_suspended);
+        let oi = ObjectInfo::from_file_info(fi, bucket, object, opts.versioned || opts.version_suspended);
 
         if !fi.version_purge_status().is_empty() && opts.version_id.is_some() {
             return (
@@ -2723,22 +2716,14 @@ mod metadata_cache_tests {
             .await
             .expect("fresh cache entry should be returned");
 
-        let (returned_fi, returned_parts_metadata, returned_online_disks) = set
+        let returned = set
             .get_object_fileinfo("bucket", "object", &ObjectOptions::default(), true, false)
             .await
             .expect("cache-backed metadata lookup should succeed");
 
         assert!(
-            matches!(returned_fi, GetObjectMetadata::Shared(ref value) if Arc::ptr_eq(value, &cached.fi)),
+            returned.shared_entry().is_some_and(|value| Arc::ptr_eq(value, &cached)),
-            "cache hits must share FileInfo ownership"
+            "cache hits must share the complete metadata snapshot"
-        );
-        assert!(
-            matches!(returned_parts_metadata, GetObjectMetadata::Shared(ref value) if Arc::ptr_eq(value, &cached.parts_metadata)),
-            "cache hits must share the metadata vector"
-        );
-        assert!(
-            matches!(returned_online_disks, GetObjectMetadata::Shared(ref value) if Arc::ptr_eq(value, &cached.online_disks)),
-            "cache hits must share the online-disk vector"
         );
     }
 
@@ -2781,9 +2766,9 @@ mod metadata_cache_tests {
                 ),
                 Arc::new(GetObjectMetadataCacheEntry {
                     created_at: Instant::now(),
-                    fi: Arc::new(fi.clone()),
+                    fi: fi.clone(),
-                    parts_metadata: Arc::new(vec![fi]),
+                    parts_metadata: vec![fi],
-                    online_disks: Arc::new(vec![None]),
+                    online_disks: vec![None],
                     read_quorum: 1,
                 }),
             )
@@ -2877,13 +2862,12 @@ mod metadata_cache_tests {
         barrier.wait_until_paused().await;
         set.invalidate_get_object_metadata_cache(bucket, object).await;
         barrier.release();
-        let (fi, parts_metadata, online_disks) = read
+        let snapshot = read
             .await
             .expect("metadata read task should not panic")
             .expect("metadata fanout should still return its selected FileInfo");
-        assert!(matches!(fi, GetObjectMetadata::Owned(_)));
+        assert!(snapshot.owned.is_some());
-        assert!(matches!(parts_metadata, GetObjectMetadata::Owned(_)));
+        assert!(snapshot.has_valid_representation());
-        assert!(matches!(online_disks, GetObjectMetadata::Owned(_)));
 
         assert!(
             set.get_object_metadata_cache
@@ -2930,9 +2914,9 @@ mod metadata_cache_tests {
         let key = GetObjectMetadataCacheKey::new("bucket", "object", generation);
         let entry = Arc::new(GetObjectMetadataCacheEntry {
             created_at: Instant::now(),
-            fi: Arc::new(fi.clone()),
+            fi: fi.clone(),
-            parts_metadata: Arc::new(vec![fi]),
+            parts_metadata: vec![fi],
-            online_disks: Arc::new(Vec::new()),
+            online_disks: Vec::new(),
             read_quorum: 0,
         });
 
@@ -3040,9 +3024,9 @@ mod metadata_cache_tests {
         let entry = |fi: FileInfo| {
             Arc::new(GetObjectMetadataCacheEntry {
                 created_at: Instant::now(),
-                parts_metadata: Arc::new(vec![fi.clone()]),
+                parts_metadata: vec![fi.clone()],
-                fi: Arc::new(fi),
+                fi,
-                online_disks: Arc::new(Vec::new()),
+                online_disks: Vec::new(),
                 read_quorum: 0,
             })
         };
@@ -4105,8 +4089,8 @@ mod tests {
         get_codec_streaming_reader_gate(
             CODEC_STREAMING_TEST_BUCKET,
             CODEC_STREAMING_TEST_OBJECT,
-            range,
             None,
+            classify_get_codec_streaming_object_class(range, object_info, fi),
             object_info,
             fi,
             lock_optimization_enabled,
@@ -4123,8 +4107,8 @@ mod tests {
         get_codec_streaming_reader_gate(
             CODEC_STREAMING_TEST_BUCKET,
             CODEC_STREAMING_TEST_OBJECT,
-            range,
             part_number,
+            classify_get_codec_streaming_object_class(range, object_info, fi),
             object_info,
             fi,
             lock_optimization_enabled,
@@ -4850,6 +4834,114 @@ mod tests {
         .await
     }
 
+    async fn encoded_inline_blocks(blocks: &[&[u8]], shard_size: usize, hash_algo: HashAlgorithm) -> Bytes {
+        let mut writer = BitrotWriter::new(Cursor::new(Vec::new()), shard_size, hash_algo);
+        for block in blocks {
+            writer.write(block).await.expect("test block should be encoded");
+        }
+        Bytes::from(writer.into_inner().into_inner())
+    }
+
+    fn assert_reader_shares_inline_allocation(reader: &ObjectBitrotReader, source: &Bytes) {
+        let reader_bytes = reader
+            .inner_ref()
+            .inline_bytes()
+            .expect("inline scheduler should retain an in-memory Bytes source");
+        assert_eq!(
+            reader_bytes.as_ptr(),
+            source.as_ptr(),
+            "the scheduler must clone Bytes ownership instead of copying the inline shard payload"
+        );
+    }
+
+    #[tokio::test]
+    async fn inline_range_scheduler_shares_bytes_and_rejects_bitrot_mismatch() {
+        const SHARD_SIZE: usize = 16;
+        let hash_algo = HashAlgorithm::HighwayHash256S;
+        let first = [b'a'; SHARD_SIZE];
+        let second = [b'b'; SHARD_SIZE];
+        let mut source = encoded_inline_blocks(&[&first, &second], SHARD_SIZE, hash_algo.clone()).await;
+        let second_payload = hash_algo.size() * 2 + SHARD_SIZE;
+        source = {
+            let mut corrupt = source.to_vec();
+            corrupt[second_payload] ^= 0xff;
+            Bytes::from(corrupt)
+        };
+        let files = vec![encoded_reader_setup_fileinfo(Some(source.to_vec()))];
+        let source = files[0].data.clone().expect("inline shard should exist");
+        let disks = vec![None];
+
+        let mut setup = create_bitrot_readers_until_quorum_with_preference(
+            &files,
+            &disks,
+            "bucket",
+            "object",
+            1,
+            SHARD_SIZE,
+            SHARD_SIZE,
+            SHARD_SIZE,
+            hash_algo,
+            false,
+            false,
+            1,
+            0,
+            BitrotReaderSetupMode::ReadQuorum,
+            true,
+            None,
+            None,
+        )
+        .await;
+        let mut reader = setup.readers[0].take().expect("range reader should be ready");
+        assert_reader_shares_inline_allocation(&reader, &source);
+
+        let err = reader
+            .read(&mut [0; SHARD_SIZE])
+            .await
+            .expect_err("corrupt ranged inline block must fail bitrot verification");
+        assert_eq!(err.kind(), ErrorKind::InvalidData);
+    }
+
+    #[tokio::test]
+    async fn inline_part_scheduler_shares_bytes_and_rejects_bitrot_mismatch() {
+        const SHARD_SIZE: usize = 16;
+        let hash_algo = HashAlgorithm::HighwayHash256S;
+        let block = [b'p'; SHARD_SIZE];
+        let encoded = encoded_inline_blocks(&[&block], SHARD_SIZE, hash_algo.clone()).await;
+        let mut corrupt = encoded.to_vec();
+        corrupt[hash_algo.size()] ^= 0xff;
+        let files = vec![encoded_reader_setup_fileinfo(Some(corrupt))];
+        let source = files[0].data.clone().expect("inline shard should exist");
+        let disks = vec![None];
+
+        let mut setup = create_bitrot_readers_until_quorum_all_shards(
+            &files,
+            &disks,
+            "bucket",
+            "object",
+            7,
+            0,
+            SHARD_SIZE,
+            SHARD_SIZE,
+            hash_algo,
+            false,
+            false,
+            1,
+            0,
+            BitrotReaderSetupMode::VerifyReconstruction,
+            None,
+            None,
+        )
+        .await;
+        let mut reader = setup.readers[0].take().expect("part reader should be ready");
+        assert_reader_shares_inline_allocation(&reader, &source);
+
+        let err = reader
+            .read(&mut [0; SHARD_SIZE])
+            .await
+            .expect_err("corrupt inline part must fail bitrot verification");
+        assert_eq!(err.kind(), ErrorKind::InvalidData);
+    }
+
     async fn decode_codec_data_blocks_first_setup(
         erasure: coding::Erasure,
         data: &[u8],
@@ -5550,6 +5642,63 @@ mod tests {
         });
     }
 
+    #[test]
+    fn codec_streaming_config_cache_loads_once() {
+        use std::cell::Cell;
+
+        let loads = Cell::new(0);
+        let expected = GetCodecStreamingConfig {
+            enabled: true,
+            rollout: GetCodecStreamingRollout::Off,
+            rollout_pct: 100,
+            body_compat_confirmed: true,
+            header_compat_confirmed: true,
+            engine: GetCodecStreamingEngine::Legacy,
+            min_size: DEFAULT_RUSTFS_GET_CODEC_STREAMING_MIN_SIZE,
+        };
+
+        for _ in 0..3 {
+            assert_eq!(
+                get_codec_streaming_config_cached_core(|| {
+                    loads.set(loads.get() + 1);
+                    expected
+                }),
+                expected
+            );
+        }
+        assert_eq!(loads.get(), 1, "production config cache must not reload env per GET");
+    }
+
+    #[test]
+    fn codec_streaming_config_loader_preserves_all_gate_env_overrides() {
+        temp_env::with_vars(
+            [
+                (ENV_RUSTFS_GET_CODEC_STREAMING_ENABLE, Some("false")),
+                (ENV_RUSTFS_GET_CODEC_STREAMING_ENGINE, Some(GET_CODEC_STREAMING_ENGINE_RUSTFS)),
+                (ENV_RUSTFS_GET_CODEC_STREAMING_ROLLOUT, Some("production")),
+                (ENV_RUSTFS_GET_CODEC_STREAMING_ROLLOUT_PCT, Some("37")),
+                (ENV_RUSTFS_GET_CODEC_STREAMING_BODY_COMPAT_CONFIRMED, Some("false")),
+                (ENV_RUSTFS_GET_CODEC_STREAMING_HEADER_COMPAT_CONFIRMED, Some("false")),
+                (ENV_RUSTFS_GET_CODEC_STREAMING_MIN_SIZE, None::<&str>),
+                (ENV_RUSTFS_GET_CODEC_STREAMING_RUSTFS_MIN_SIZE, Some("262144")),
+            ],
+            || {
+                assert_eq!(
+                    load_get_codec_streaming_config(),
+                    GetCodecStreamingConfig {
+                        enabled: false,
+                        rollout: GetCodecStreamingRollout::On,
+                        rollout_pct: 37,
+                        body_compat_confirmed: false,
+                        header_compat_confirmed: false,
+                        engine: GetCodecStreamingEngine::Rustfs,
+                        min_size: 262144,
+                    }
+                );
+            },
+        );
+    }
+
     #[test]
     fn codec_streaming_default_min_size_meets_direct_memory_ceiling() {
         for engine in [None, Some(GET_CODEC_STREAMING_ENGINE_RUSTFS)] {

crates/ecstore/src/set_disk/replication.rs

@@ -78,10 +78,10 @@ impl SetDisks {
             include_part_checksums: true,
             ..Default::default()
         };
-        let (fi, _, disks) = self
+        let (mut fi, _, disks) = self
             .get_object_fileinfo_gated(bucket, object, &read_opts, false, false)
-            .await?;
+            .await?
-        let mut fi = fi.into_owned();
+            .into_owned();
         if let Some(expected_operation_id) = expected_operation_id {
             require_restore_operation_id(&fi.metadata, expected_operation_id)?;
         }
@@ -146,10 +146,10 @@ impl SetDisks {
             include_part_checksums: true,
             ..Default::default()
         };
-        let (fi, _, disks) = self
+        let (mut fi, _, disks) = self
             .get_object_fileinfo_gated(bucket, object, &read_opts, false, false)
-            .await?;
+            .await?
-        let mut fi = fi.into_owned();
+            .into_owned();
         if let Some(expected_operation_id) = expected_operation_id {
             match restore_operation_id_from_metadata(&fi.metadata)? {
                 Some(actual_operation_id) if actual_operation_id == expected_operation_id => {}

crates/ecstore/src/set_disk/shard_source.rs

@@ -16,6 +16,14 @@ use crate::diagnostics::get::{
     GET_SHARD_READ_COST_LOCAL, GET_SHARD_READ_COST_REMOTE, GET_SHARD_READ_COST_SAME_NODE, GET_SHARD_READ_COST_UNKNOWN,
 };
 use crate::disk::error::Error;
+use crate::layout::disks_layout::MAX_ERASURE_SET_DRIVE_COUNT;
+use smallvec::SmallVec;
+
+/// Generic codec callers may exceed the production set limit; `SmallVec` then
+/// spills without changing slot semantics.
+pub(crate) const INLINE_SHARD_SLOTS: usize = MAX_ERASURE_SET_DRIVE_COUNT;
+pub(crate) type ShardBuffers = SmallVec<[Option<Vec<u8>>; INLINE_SHARD_SLOTS]>;
+pub(crate) type ShardErrors = SmallVec<[Option<Error>; INLINE_SHARD_SLOTS]>;
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub(crate) enum ShardReadCost {
@@ -43,202 +51,137 @@ impl ShardReadCost {
     }
 }
 
-#[derive(Debug, Clone, PartialEq, Eq)]
-pub(crate) struct ShardSlot {
-    index: usize,
-    read_cost: ShardReadCost,
-    data: Option<Vec<u8>>,
-    error: Option<Error>,
-}
-
-impl ShardSlot {
-    pub(crate) fn new(index: usize, data: Option<Vec<u8>>, error: Option<Error>) -> Self {
-        Self::with_read_cost(index, ShardReadCost::Unknown, data, error)
-    }
-
-    pub(crate) fn with_read_cost(index: usize, read_cost: ShardReadCost, data: Option<Vec<u8>>, error: Option<Error>) -> Self {
-        Self {
-            index,
-            read_cost,
-            data,
-            error,
-        }
-    }
-
-    pub(crate) fn data(index: usize, data: Vec<u8>) -> Self {
-        Self::new(index, Some(data), None)
-    }
-
-    pub(crate) fn data_with_read_cost(index: usize, read_cost: ShardReadCost, data: Vec<u8>) -> Self {
-        Self::with_read_cost(index, read_cost, Some(data), None)
-    }
-
-    pub(crate) fn missing(index: usize, error: Error) -> Self {
-        Self::new(index, None, Some(error))
-    }
-
-    pub(crate) fn missing_with_read_cost(index: usize, read_cost: ShardReadCost, error: Error) -> Self {
-        Self::with_read_cost(index, read_cost, None, Some(error))
-    }
-
-    pub(crate) fn index(&self) -> usize {
-        self.index
-    }
-
-    pub(crate) fn read_cost(&self) -> ShardReadCost {
-        self.read_cost
-    }
-
-    pub(crate) fn has_data(&self) -> bool {
-        self.data.is_some()
-    }
-
-    pub(crate) fn data_bytes(&self) -> Option<&[u8]> {
-        self.data.as_deref()
-    }
-
-    pub(crate) fn error(&self) -> Option<&Error> {
-        self.error.as_ref()
-    }
-}
-
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub(crate) struct StripeReadState {
-    slots: Vec<ShardSlot>,
+    shards: ShardBuffers,
+    errors: ShardErrors,
     read_quorum: usize,
 }
 
 impl StripeReadState {
-    pub(crate) fn new(slots: Vec<ShardSlot>, read_quorum: usize) -> Self {
+    #[cfg(test)]
-        Self { slots, read_quorum }
-    }
-
     pub(crate) fn from_parts(shards: Vec<Option<Vec<u8>>>, errors: Vec<Option<Error>>, read_quorum: usize) -> Self {
-        Self::from_parts_with_read_costs(shards, errors, &[], read_quorum)
+        let mut shards = SmallVec::from_vec(shards);
+        let mut errors = SmallVec::from_vec(errors);
+        let slot_count = shards.len().max(errors.len());
+        shards.resize_with(slot_count, || None);
+        errors.resize_with(slot_count, || None);
+        Self {
+            shards,
+            errors,
+            read_quorum,
+        }
     }
 
-    pub(crate) fn from_parts_with_read_costs<S, E>(shards: S, errors: E, read_costs: &[ShardReadCost], read_quorum: usize) -> Self
+    pub(crate) fn with_slot_count(slot_count: usize, read_quorum: usize) -> Self {
-    where
+        let mut state = Self {
-        S: IntoIterator<Item = Option<Vec<u8>>>,
+            shards: SmallVec::new(),
-        S::IntoIter: ExactSizeIterator,
+            errors: SmallVec::new(),
-        E: IntoIterator<Item = Option<Error>>,
+            read_quorum,
-        E::IntoIter: ExactSizeIterator,
+        };
-    {
+        state.reset(slot_count, read_quorum);
-        let mut shards = shards.into_iter();
+        state
-        let mut errors = errors.into_iter();
+    }
-        let slot_count = shards.len().max(errors.len());
+
-        let mut slots = Vec::with_capacity(slot_count);
+    pub(crate) fn reset(&mut self, slot_count: usize, read_quorum: usize) {
-        for index in 0..slot_count {
+        self.shards.clear();
-            let read_cost = read_costs.get(index).copied().unwrap_or(ShardReadCost::Unknown);
+        self.shards.resize_with(slot_count, || None);
-            slots.push(ShardSlot::with_read_cost(
+        self.errors.clear();
-                index,
+        self.errors.resize_with(slot_count, || None);
-                read_cost,
+        self.read_quorum = read_quorum;
-                shards.next().flatten(),
-                errors.next().flatten(),
-            ));
-        }
-        Self::new(slots, read_quorum)
     }
 
     pub(crate) fn available_shards(&self) -> usize {
-        self.slots.iter().filter(|slot| slot.has_data()).count()
+        self.shards.iter().filter(|shard| shard.is_some()).count()
     }
 
     pub(crate) fn can_decode(&self) -> bool {
         self.available_shards() >= self.read_quorum
     }
 
-    pub(crate) fn slots(&self) -> &[ShardSlot] {
+    pub(crate) fn is_empty(&self) -> bool {
-        &self.slots
+        self.shards.is_empty()
     }
 
-    pub(crate) fn slot_by_index(&self, index: usize) -> Option<&ShardSlot> {
+    pub(crate) fn data_bytes(&self, index: usize) -> Option<&[u8]> {
-        if let Some(slot) = self.slots.get(index)
+        self.shards.get(index).and_then(Option::as_deref)
-            && slot.index == index
+    }
-        {
+
-            return Some(slot);
+    #[cfg(test)]
-        }
+    pub(crate) fn error(&self, index: usize) -> Option<&Error> {
-        self.slots.iter().find(|slot| slot.index == index)
+        self.errors.get(index).and_then(Option::as_ref)
     }
 
     pub(crate) fn data_shards_complete(&self, data_shards: usize) -> bool {
-        (0..data_shards).all(|index| self.slot_by_index(index).is_some_and(ShardSlot::has_data))
+        self.shards.len() >= data_shards && self.shards.iter().take(data_shards).all(Option::is_some)
     }
 
-    pub(crate) fn into_parts(self) -> (Vec<Option<Vec<u8>>>, Vec<Option<Error>>) {
+    pub(crate) fn parts_mut(&mut self) -> (&mut ShardBuffers, &mut ShardErrors) {
-        let part_count = self.slots.iter().map(|slot| slot.index).max().map_or(0, |index| index + 1);
+        (&mut self.shards, &mut self.errors)
-        let mut shards = Vec::with_capacity(part_count);
+    }
-        shards.resize_with(part_count, || None);
+
-        let mut errors = Vec::with_capacity(part_count);
+    pub(crate) fn shards_mut(&mut self) -> &mut ShardBuffers {
-        errors.resize_with(part_count, || None);
+        &mut self.shards
-        for slot in self.slots {
+    }
-            shards[slot.index] = slot.data;
+
-            errors[slot.index] = slot.error;
+    pub(crate) fn into_parts(self) -> (ShardBuffers, ShardErrors) {
-        }
+        (self.shards, self.errors)
-        (shards, errors)
+    }
+
+    #[cfg(test)]
+    pub(crate) fn scratch_storage(&self) -> (*const Option<Vec<u8>>, *const Option<Error>, bool, bool) {
+        (self.shards.as_ptr(), self.errors.as_ptr(), self.shards.spilled(), self.errors.spilled())
+    }
+
+    #[cfg(test)]
+    pub(crate) fn shard_allocation(&self, index: usize) -> Option<(*const u8, usize)> {
+        self.shards
+            .get(index)
+            .and_then(|shard| shard.as_ref().map(|shard| (shard.as_ptr(), shard.capacity())))
     }
 }
 
 #[async_trait::async_trait]
 pub(crate) trait ShardStripeSource: Send {
-    async fn read_next_stripe(&mut self) -> StripeReadState;
+    async fn read_next_stripe(&mut self) -> Box<StripeReadState>;
+
+    fn recycle_stripe(&mut self, _state: Box<StripeReadState>) {}
 }
 
 #[cfg(test)]
 mod tests {
     use super::*;
+    use std::mem::size_of;
 
     #[test]
-    fn stripe_read_state_tracks_decode_quorum() {
+    fn stripe_scratch_capacity_matches_the_production_set_limit() {
-        let state = StripeReadState::new(
+        type OversizedShardBuffers = SmallVec<[Option<Vec<u8>>; 32]>;
-            vec![
+        type OversizedShardErrors = SmallVec<[Option<Error>; 32]>;
-                ShardSlot::data_with_read_cost(0, ShardReadCost::Local, vec![1]),
+
-                ShardSlot::missing_with_read_cost(1, ShardReadCost::Remote, Error::FileNotFound),
+        assert_eq!(INLINE_SHARD_SLOTS, MAX_ERASURE_SET_DRIVE_COUNT);
-                ShardSlot::data_with_read_cost(2, ShardReadCost::SameNode, vec![2]),
+        assert!(size_of::<ShardBuffers>() < size_of::<OversizedShardBuffers>());
-            ],
+        assert!(size_of::<ShardErrors>() < size_of::<OversizedShardErrors>());
-            2,
+    }
-        );
+
+    #[test]
+    fn stripe_read_state_tracks_decode_quorum_and_slot_access() {
+        let state =
+            StripeReadState::from_parts(vec![Some(vec![1]), None, Some(vec![2])], vec![None, Some(Error::FileNotFound), None], 2);
 
         assert_eq!(state.available_shards(), 2);
         assert!(state.can_decode());
-        assert_eq!(state.slots()[1].index(), 1);
+        assert_eq!(state.data_bytes(0), Some(&[1][..]));
-        assert_eq!(state.slots()[0].read_cost(), ShardReadCost::Local);
+        assert_eq!(state.error(1), Some(&Error::FileNotFound));
-        assert!(state.slots()[2].read_cost().is_low_cost());
     }
 
     #[test]
     fn stripe_read_state_preserves_shards_and_errors() {
-        let state = StripeReadState::new(vec![ShardSlot::missing(1, Error::FileCorrupt), ShardSlot::data(0, vec![1, 2, 3])], 2);
+        let state = StripeReadState::from_parts(vec![Some(vec![1, 2, 3]), None], vec![None, Some(Error::FileCorrupt)], 2);
 
         assert!(!state.can_decode());
         let (shards, errors) = state.into_parts();
-        assert_eq!(shards, vec![Some(vec![1, 2, 3]), None]);
+        assert_eq!(shards.as_slice(), &[Some(vec![1, 2, 3]), None]);
-        assert_eq!(errors, vec![None, Some(Error::FileCorrupt)]);
+        assert_eq!(errors.as_slice(), &[None, Some(Error::FileCorrupt)]);
-    }
-
-    #[test]
-    fn stripe_read_state_builds_slots_from_parallel_reader_parts() {
-        let state =
-            StripeReadState::from_parts(vec![Some(vec![1]), None, Some(vec![3])], vec![None, Some(Error::FileNotFound)], 2);
-
-        assert!(state.can_decode());
-        assert_eq!(state.slots()[1].index(), 1);
-        assert_eq!(state.slots()[1].error(), Some(&Error::FileNotFound));
-    }
-
-    #[test]
-    fn stripe_read_state_preserves_read_cost_hints() {
-        let state = StripeReadState::from_parts_with_read_costs(
-            vec![Some(vec![1]), None, Some(vec![3])],
-            vec![None, Some(Error::FileNotFound)],
-            &[ShardReadCost::Local, ShardReadCost::Remote, ShardReadCost::Unknown],
-            2,
-        );
-
-        assert_eq!(state.slots()[0].read_cost(), ShardReadCost::Local);
-        assert_eq!(state.slots()[1].read_cost(), ShardReadCost::Remote);
-        assert_eq!(state.slots()[2].read_cost(), ShardReadCost::Unknown);
-        assert_eq!(ShardReadCost::SameNode.as_str(), GET_SHARD_READ_COST_SAME_NODE);
     }
 
     #[test]
@@ -258,8 +201,8 @@ mod tests {
         let state = StripeReadState::from_parts(vec![Some(vec![1]), Some(vec![2]), None], Vec::new(), 2);
 
         assert!(state.data_shards_complete(2));
-        assert_eq!(state.slots()[0].data_bytes(), Some(&[1][..]));
+        assert_eq!(state.data_bytes(0), Some(&[1][..]));
-        assert_eq!(state.slot_by_index(1).and_then(ShardSlot::data_bytes), Some(&[2][..]));
+        assert_eq!(state.data_bytes(1), Some(&[2][..]));
     }
 
     #[test]
@@ -268,12 +211,4 @@ mod tests {
 
         assert!(!state.data_shards_complete(2));
     }
-
-    #[test]
-    fn stripe_read_state_finds_out_of_order_slots_by_index() {
-        let state = StripeReadState::new(vec![ShardSlot::data(2, vec![3]), ShardSlot::data(0, vec![1])], 2);
-
-        assert_eq!(state.slot_by_index(0).and_then(ShardSlot::data_bytes), Some(&[1][..]));
-        assert!(state.slot_by_index(1).is_none());
-    }
 }

crates/ecstore/src/store/mod.rs

@@ -389,7 +389,7 @@ impl crate::storage_api_contracts::object::ObjectIO for ECStore {
     type GetObjectReader = GetObjectReader;
     type PutObjectReader = PutObjReader;
 
-    #[instrument(level = "debug", skip(self))]
+    #[instrument(level = "debug", skip(self, h))]
     async fn get_object_reader(
         &self,
         bucket: &str,

crates/ecstore/src/store/object.rs

@@ -667,18 +667,6 @@ impl SelectObjectSnapshotLockLossWake {
     }
 }
 
-fn select_object_ssec_headers(headers: &HeaderMap) -> HeaderMap {
-    use rustfs_utils::http::headers::{SSEC_ALGORITHM_HEADER, SSEC_KEY_HEADER, SSEC_KEY_MD5_HEADER};
-
-    let mut selected = HeaderMap::new();
-    for name in [SSEC_ALGORITHM_HEADER, SSEC_KEY_HEADER, SSEC_KEY_MD5_HEADER] {
-        if let Some(value) = headers.get(name) {
-            selected.insert(name, value.clone());
-        }
-    }
-    selected
-}
-
 // LockRegistry clones its canonical client Arc for each endpoint host, so an
 // exact Arc set identifies one distributed namespace-lock quorum domain.
 fn same_distributed_lock_domain(left: &[Arc<dyn rustfs_lock::LockClient>], right: &[Arc<dyn rustfs_lock::LockClient>]) -> bool {
@@ -1296,7 +1284,7 @@ impl ECStore {
             pool,
             bucket: bucket.to_owned(),
             object,
-            headers: select_object_ssec_headers(headers),
+            headers: rustfs_utils::http::project_ssec_transport_headers(headers),
             opts,
             object_info,
             logical_size,
@@ -1769,7 +1757,7 @@ impl ECStore {
         Self::resolve_decommission_tiered_object_result(result, bucket, &object)
     }
 
-    #[instrument(level = "debug", skip(self))]
+    #[instrument(level = "debug", skip(self, h))]
     #[hotpath::measure(impl_type = "ECStore")]
     pub(super) async fn handle_get_object_reader(
         &self,
@@ -3360,25 +3348,6 @@ mod tests {
         assert!(second_signal.is_lost());
     }
 
-    #[test]
-    fn select_snapshot_retains_only_ssec_headers() {
-        use rustfs_utils::http::headers::{SSEC_ALGORITHM_HEADER, SSEC_KEY_HEADER, SSEC_KEY_MD5_HEADER};
-
-        let mut headers = HeaderMap::new();
-        headers.insert(SSEC_ALGORITHM_HEADER, "AES256".parse().expect("valid SSE-C algorithm header"));
-        headers.insert(SSEC_KEY_HEADER, "secret-key".parse().expect("valid SSE-C key header"));
-        headers.insert(SSEC_KEY_MD5_HEADER, "key-md5".parse().expect("valid SSE-C key digest header"));
-        headers.insert("authorization", "credential".parse().expect("valid authorization header"));
-
-        let selected = select_object_ssec_headers(&headers);
-
-        assert_eq!(selected.len(), 3);
-        assert_eq!(selected.get(SSEC_ALGORITHM_HEADER), headers.get(SSEC_ALGORITHM_HEADER));
-        assert_eq!(selected.get(SSEC_KEY_HEADER), headers.get(SSEC_KEY_HEADER));
-        assert_eq!(selected.get(SSEC_KEY_MD5_HEADER), headers.get(SSEC_KEY_MD5_HEADER));
-        assert!(selected.get("authorization").is_none());
-    }
-
     #[test]
     fn tier_delete_entry_is_prepared_and_bound_to_source_generation() {
         let identity = [9_u8; 32];
@@ -4608,6 +4577,8 @@ mod tests {
             assert_eq!(snapshot.headers.get(SSEC_KEY_HEADER), request_headers.get(SSEC_KEY_HEADER));
             assert_eq!(snapshot.headers.get(SSEC_KEY_MD5_HEADER), request_headers.get(SSEC_KEY_MD5_HEADER));
             assert!(snapshot.headers.get("authorization").is_none());
+            assert!(snapshot.headers.values().all(http::HeaderValue::is_sensitive));
+            assert!(!format!("{:?}", snapshot.headers).contains("secret-key"));
             assert_eq!(
                 snapshot.logical_size(),
                 u64::try_from(payload.len()).expect("test payload length should fit in u64")

crates/lock/src/distributed_lock.rs

@@ -477,7 +477,7 @@ impl Drop for DistributedLockGuard {
 #[derive(Debug)]
 pub struct DistributedLock {
     /// Lock clients for this namespace
-    clients: Vec<Arc<dyn LockClient>>,
+    clients: Arc<[Arc<dyn LockClient>]>,
     /// Namespace identifier
     namespace: Arc<str>,
     /// Quorum size for exclusive/write operations
@@ -496,11 +496,11 @@ struct LockAcquireQuorumResult {
 impl DistributedLock {
     /// Create new distributed lock
     pub fn new(namespace: String, clients: Vec<Arc<dyn LockClient>>, quorum: usize) -> Self {
-        Self::new_shared(namespace.into(), clients, quorum)
+        Self::new_shared(namespace.into(), clients.into(), quorum)
     }
 
-    /// Create a distributed lock that shares an existing namespace allocation.
+    /// Create a distributed lock that shares existing namespace and client allocations.
-    pub(crate) fn new_shared(namespace: Arc<str>, clients: Vec<Arc<dyn LockClient>>, quorum: usize) -> Self {
+    pub(crate) fn new_shared(namespace: Arc<str>, clients: Arc<[Arc<dyn LockClient>]>, quorum: usize) -> Self {
         let q = if clients.len() <= 1 {
             1
         } else {
@@ -777,7 +777,7 @@ impl DistributedLock {
 
     fn spawn_pending_cleanup(
         mut pending: JoinSet<LockAcquireTaskResult>,
-        clients: Vec<Arc<dyn LockClient>>,
+        clients: Arc<[Arc<dyn LockClient>]>,
         fallback_lock_id: LockId,
         context: &'static str,
     ) {

crates/lock/src/namespace/mod.rs

@@ -200,9 +200,13 @@ impl NamespaceLock {
         Self::Distributed(DistributedLock::new(namespace, clients, quorum))
     }
 
-    /// Create a namespace lock that shares an existing namespace allocation.
+    /// Create a namespace lock that shares existing namespace and client allocations.
-    pub fn with_clients_and_quorum_shared(namespace: Arc<str>, clients: Vec<Arc<dyn LockClient>>, quorum: usize) -> Self {
+    pub fn with_clients_and_quorum_shared(
-        Self::Distributed(DistributedLock::new_shared(namespace, clients, quorum))
+        namespace: Arc<str>,
+        clients: impl Into<Arc<[Arc<dyn LockClient>]>>,
+        quorum: usize,
+    ) -> Self {
+        Self::Distributed(DistributedLock::new_shared(namespace, clients.into(), quorum))
     }
 
     /// Get namespace identifier

crates/utils/src/http/object_encryption_keys.rs

@@ -25,7 +25,7 @@
 // The lowercase stored forms, matching exactly what encryption_material_to_metadata
 // persists. The read-path SSE-C check is case-sensitive, so restoring under any
 // other casing would classify the replica as managed-SSE and reject SSE-C GETs.
-use super::headers::{SSEC_ALGORITHM_HEADER, SSEC_KEY_MD5_HEADER};
+use super::headers::{SSEC_ALGORITHM_HEADER, SSEC_KEY_HEADER, SSEC_KEY_MD5_HEADER};
 
 pub const INTERNAL_ENCRYPTION_KEY_ID_HEADER: &str = "x-rustfs-encryption-key-id";
 pub const INTERNAL_ENCRYPTION_KEY_HEADER: &str = "x-rustfs-encryption-key";
@@ -85,6 +85,20 @@ pub const SSEC_REPLICATION_TRANSPORT_HEADERS: &[(&str, &str)] = &[
     (MINIO_INTERNAL_ENCRYPTION_MULTIPART_HEADER, REPLICATION_ENCRYPTED_MULTIPART_HEADER),
 ];
 
+/// Retains only the SSE-C headers consumed by object readers and marks their
+/// values sensitive so instrumented storage calls cannot expose key material.
+pub fn project_ssec_transport_headers(headers: &http::HeaderMap) -> http::HeaderMap {
+    let mut projected = http::HeaderMap::new();
+    for name in [SSEC_ALGORITHM_HEADER, SSEC_KEY_HEADER, SSEC_KEY_MD5_HEADER] {
+        if let Some(value) = headers.get(name) {
+            let mut value = value.clone();
+            value.set_sensitive(true);
+            projected.insert(name, value);
+        }
+    }
+    projected
+}
+
 /// Prefixes of replication SSE transport keys whose values carry encryption
 /// material and must never reach logs. Consumed by `rustfs_filemeta` redaction.
 pub const REPLICATION_SSE_TRANSPORT_PREFIXES: &[&str] = &[
@@ -155,6 +169,22 @@ pub fn is_replication_stripped_encryption_key(key: &str) -> bool {
 mod tests {
     use super::*;
 
+    #[test]
+    fn ssec_transport_projection_retains_only_redacted_reader_headers() {
+        let mut headers = http::HeaderMap::new();
+        headers.insert(SSEC_ALGORITHM_HEADER, http::HeaderValue::from_static("AES256"));
+        headers.insert(SSEC_KEY_HEADER, http::HeaderValue::from_static("secret-key"));
+        headers.insert(SSEC_KEY_MD5_HEADER, http::HeaderValue::from_static("key-md5"));
+        headers.insert(http::header::AUTHORIZATION, http::HeaderValue::from_static("credential"));
+
+        let projected = project_ssec_transport_headers(&headers);
+
+        assert_eq!(projected.len(), 3);
+        assert!(projected.values().all(http::HeaderValue::is_sensitive));
+        assert!(projected.get(http::header::AUTHORIZATION).is_none());
+        assert!(!format!("{projected:?}").contains("secret-key"));
+    }
+
     #[test]
     fn transport_metadata_roundtrip_restores_stored_keys() {
         let mut headers = http::HeaderMap::new();

rustfs/src/app/object_usecase.rs

@@ -138,6 +138,8 @@ use rustfs_s3select_api::object_store::bytes_stream;
 use rustfs_targets::{EventName, get_request_host, get_request_port, get_request_user_agent};
 use rustfs_utils::CompressionAlgorithm;
 #[cfg(test)]
+use rustfs_utils::http::headers::{SSEC_ALGORITHM_HEADER, SSEC_KEY_HEADER, SSEC_KEY_MD5_HEADER};
+#[cfg(test)]
 use rustfs_utils::http::insert_header;
 use rustfs_utils::http::{
     AMZ_BUCKET_REPLICATION_STATUS, AMZ_CHECKSUM_MODE, AMZ_CHECKSUM_TYPE, AMZ_WEBSITE_REDIRECT_LOCATION, CONTENT_TYPE,
@@ -152,9 +154,8 @@ use rustfs_utils::http::{
         AMZ_RUSTFS_SNOWBALL_IGNORE_ERRORS, AMZ_RUSTFS_SNOWBALL_PREFIX, AMZ_SERVER_SIDE_ENCRYPTION,
         AMZ_SERVER_SIDE_ENCRYPTION_CUSTOMER_ALGORITHM, AMZ_SERVER_SIDE_ENCRYPTION_KMS_ID, AMZ_SNOWBALL_EXTRACT,
         AMZ_SNOWBALL_IGNORE_DIRS, AMZ_SNOWBALL_IGNORE_ERRORS, AMZ_SNOWBALL_PREFIX, AMZ_STORAGE_CLASS, AMZ_TAG_COUNT,
-        SSEC_ALGORITHM_HEADER, SSEC_KEY_HEADER, SSEC_KEY_MD5_HEADER,
     },
-    insert_str, remove_str,
+    insert_str, project_ssec_transport_headers, remove_str,
 };
 use rustfs_utils::path::{encode_dir_object, is_dir_object, path_join_buf};
 use rustfs_utils::retry::{DEFAULT_RETRY_CAP, DEFAULT_RETRY_UNIT, MAX_JITTER, RetryTimer};
@@ -2109,18 +2110,6 @@ struct GetObjectResumeContext {
     identity: GetObjectResumeIdentity,
 }
 
-fn get_object_store_headers(request_headers: &HeaderMap) -> HeaderMap {
-    let mut headers = HeaderMap::new();
-    for name in [SSEC_ALGORITHM_HEADER, SSEC_KEY_HEADER, SSEC_KEY_MD5_HEADER] {
-        if let Some(value) = request_headers.get(name) {
-            let mut value = value.clone();
-            value.set_sensitive(true);
-            headers.insert(name, value);
-        }
-    }
-    headers
-}
-
 impl GetObjectResumeContext {
     #[allow(clippy::too_many_arguments)]
     fn new(
@@ -2140,7 +2129,7 @@ impl GetObjectResumeContext {
         }
         // Store spans record their header argument at debug level. Retain only
         // the SSE-C inputs needed to reopen the reader and keep them redacted.
-        let ssec_headers = get_object_store_headers(request_headers);
+        let ssec_headers = project_ssec_transport_headers(request_headers);
         Self {
             store,
             bucket: bucket.to_string(),
@@ -4526,7 +4515,7 @@ impl DefaultObjectUsecase {
     ) -> S3Result<GetObjectPreparedRead> {
         let read_start = std::time::Instant::now();
         let read_stage_start = rustfs_io_metrics::get_stage_metrics_enabled().then_some(read_start);
-        let store_headers = get_object_store_headers(&req.headers);
+        let store_headers = project_ssec_transport_headers(&req.headers);
         let cache_adapter = self.object_data_cache();
         if cache_adapter.is_disabled() || !cache_adapter.materialize_fill_enabled() {
             let io_planning = Self::acquire_get_object_io_planning(
@@ -4541,7 +4530,7 @@ impl DefaultObjectUsecase {
             .await?;
             let reader = track_object_read_setup(
                 object_traffic_health.as_deref(),
-                store.get_object_reader(bucket, key, rs.clone(), store_headers.clone(), opts),
+                store.get_object_reader(bucket, key, rs.clone(), store_headers, opts),
             )
             .await
             .map_err(map_get_object_reader_error)?;
@@ -4830,12 +4819,10 @@ impl DefaultObjectUsecase {
             let io_planning = metadata_admission
                 .take()
                 .ok_or_else(|| s3_error!(InternalError, "prepared metadata admission is unavailable"))?;
-            let reader = track_object_read_setup(
+            let reader =
-                object_traffic_health.as_deref(),
+                track_object_read_setup(object_traffic_health.as_deref(), prepared.with_headers(store_headers).into_reader())
-                prepared.with_headers(store_headers.clone()).into_reader(),
+                    .await
-            )
+                    .map_err(map_get_object_reader_error)?;
-            .await
-            .map_err(map_get_object_reader_error)?;
             (io_planning, reader)
         } else {
             let io_planning = Self::acquire_get_object_io_planning(
@@ -4855,12 +4842,9 @@ impl DefaultObjectUsecase {
                 )
                 .await
                 .map_err(map_get_object_reader_error)?;
-                track_object_read_setup(
+                track_object_read_setup(object_traffic_health.as_deref(), prepared.with_headers(store_headers).into_reader())
-                    object_traffic_health.as_deref(),
+                    .await
-                    prepared.with_headers(store_headers.clone()).into_reader(),
+                    .map_err(map_get_object_reader_error)?
-                )
-                .await
-                .map_err(map_get_object_reader_error)?
             } else {
                 track_object_read_setup(
                     object_traffic_health.as_deref(),
@@ -13952,11 +13936,12 @@ mod tests {
         request_headers.insert(SSEC_KEY_MD5_HEADER, HeaderValue::from_static("bWQ1"));
         request_headers.insert(http::header::AUTHORIZATION, HeaderValue::from_static("AWS4-HMAC-SHA256 Credential=test"));
         request_headers.insert("x-amz-security-token", HeaderValue::from_static("session-token"));
-        let store_headers = get_object_store_headers(&request_headers);
+        let store_headers = project_ssec_transport_headers(&request_headers);
         assert_eq!(store_headers.len(), 3, "only store-consumed SSE-C headers are forwarded");
         assert!(store_headers.values().all(HeaderValue::is_sensitive));
         assert!(store_headers.get(http::header::AUTHORIZATION).is_none());
         assert!(store_headers.get("x-amz-security-token").is_none());
+        assert!(!format!("{store_headers:?}").contains("dGVzdC1rZXk="));
         let plain_info = ObjectInfo {
             size: 11,
             ..Default::default()

rustfs/src/storage/minio_generated_read_test.rs

@@ -103,6 +103,7 @@ fn load_file_info(case_dir: &Path, manifest: &ManifestRecord) -> FileInfo {
         FileInfoOpts {
             data: true,
             include_free_versions: true,
+            include_part_checksums: false,
         },
     )
     .unwrap_or_else(|err| panic!("decode {}: {err}", xl_meta_path.display()))

rustfs/src/storage/sse.rs

@@ -1718,20 +1718,23 @@ pub(crate) fn build_ssec_read_headers(
     let mut headers = HeaderMap::new();
 
     if let Some(algorithm) = algorithm
-        && let Ok(value) = HeaderValue::from_str(algorithm.as_str())
+        && let Ok(mut value) = HeaderValue::from_str(algorithm.as_str())
     {
+        value.set_sensitive(true);
         headers.insert(AMZ_SERVER_SIDE_ENCRYPTION_CUSTOMER_ALGORITHM, value);
     }
 
     if let Some(key) = key
-        && let Ok(value) = HeaderValue::from_str(key.as_str())
+        && let Ok(mut value) = HeaderValue::from_str(key.as_str())
     {
+        value.set_sensitive(true);
         headers.insert(AMZ_SERVER_SIDE_ENCRYPTION_CUSTOMER_KEY, value);
     }
 
     if let Some(key_md5) = key_md5
-        && let Ok(value) = HeaderValue::from_str(key_md5.as_str())
+        && let Ok(mut value) = HeaderValue::from_str(key_md5.as_str())
     {
+        value.set_sensitive(true);
         headers.insert(AMZ_SERVER_SIDE_ENCRYPTION_CUSTOMER_KEY_MD5, value);
     }
 
@@ -3481,6 +3484,21 @@ mod tests {
     };
     use rustfs_utils::http::headers::SSEC_ALGORITHM_HEADER;
 
+    #[test]
+    fn ssec_read_headers_are_sensitive() {
+        let headers = super::build_ssec_read_headers(
+            Some(&SSECustomerAlgorithm::from("AES256".to_string())),
+            Some(&SSECustomerKey::from("dHJhbnNwb3J0LXNlY3JldA==".to_string())),
+            Some(&SSECustomerKeyMD5::from("bWQ1LXNlY3JldA==".to_string())),
+        );
+
+        assert_eq!(headers.len(), 3);
+        assert!(headers.values().all(HeaderValue::is_sensitive));
+        let debug = format!("{headers:?}");
+        assert!(!debug.contains("dHJhbnNwb3J0LXNlY3JldA=="));
+        assert!(!debug.contains("bWQ1LXNlY3JldA=="));
+    }
+
     #[test]
     fn anonymous_s3_request_builds_kms_principal() {
         let mut request = s3s::S3Request {