perf(storage): converge Wave 2 hot-path optimizations (#6065)

* perf(get): share inline shards and lock clients Co-Authored-By: heihutu <heihutu@gmail.com> * perf(ecstore): converge PUT encoding on contiguous blocks Co-Authored-By: heihutu <heihutu@gmail.com> * perf(get): cache codec streaming gate config Co-Authored-By: heihutu <heihutu@gmail.com> * fix(sse): redact projected customer headers Co-Authored-By: heihutu <heihutu@gmail.com> * perf(ecstore): collapse GET metadata snapshots Co-Authored-By: heihutu <heihutu@gmail.com> * perf(ecstore): reuse decode stripe scratch Co-Authored-By: heihutu <heihutu@gmail.com> * refactor(ecstore): trim decode scratch adapters Co-Authored-By: heihutu <heihutu@gmail.com> * test(ecstore): adapt transition checks to metadata snapshots Co-Authored-By: heihutu <heihutu@gmail.com> * perf(get): release metadata snapshots at ownership boundary Co-Authored-By: heihutu <heihutu@gmail.com> * refactor(ecstore): close cumulative fast-path findings Co-Authored-By: heihutu <heihutu@gmail.com> * fix(storage): preserve lock and header invariants Co-Authored-By: heihutu <heihutu@gmail.com> * test(ecstore): adapt cumulative paths after rebase Co-Authored-By: heihutu <heihutu@gmail.com> * fix(rio-v2): adapt generated metadata fixture Co-Authored-By: heihutu <heihutu@gmail.com> --------- Co-authored-by: heihutu <heihutu@gmail.com>
2026-08-16 18:08:21 +00:00 · 2026-08-13 16:34:28 +08:00
parent 36deab8670
commit d2b1003612
24 changed files with 1233 additions and 697 deletions
@@ -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
@@ -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 mut errs: ShardErrors = smallvec![None; num_readers];
+        let (shards, errs) = state.parts_mut();
        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,
-                    &mut errs,
+                    shards,
+                    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,
-                        &mut errs,
+                        shards,
+                        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 mut errs: ShardErrors = smallvec![None; num_readers];
+        let (shards, errs) = state.parts_mut();
        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 {
-        let read_quorum = self.data_shards;
-        let (shards, errors) = ParallelReader::read(self).await;
-        StripeReadState::from_parts_with_read_costs(shards, errors, &self.read_costs, read_quorum)
+    async fn read_next_stripe(&mut self) -> Box<StripeReadState> {
+        let mut state = self
+            .stripe_state
+            .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() {
-        let inline: ShardBuffers = smallvec![None; INLINE_SHARD_SLOTS];
-        assert!(!inline.spilled(), "the common shard-count boundary must not allocate");
-
-        let spilled: ShardBuffers = smallvec![None; INLINE_SHARD_SLOTS + 1];
-        assert!(spilled.spilled(), "larger supported shard counts must fall back to the heap");
-        assert_eq!(spilled.len(), INLINE_SHARD_SLOTS + 1);
+    fn parallel_reader_keeps_stripe_scratch_out_of_line() {
+        eprintln!(
+            "parallel_reader={} stripe_state={} cached_state={}",
+            std::mem::size_of::<ParallelReader<Cursor<Vec<u8>>>>(),
+            std::mem::size_of::<StripeReadState>(),
+            std::mem::size_of::<Option<Box<StripeReadState>>>()
+        );
+        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 {
@@ -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 =
-        match decode_stripe_into(metrics_path, stage_metrics_enabled, engine, workspace, state, remaining, &mut output_buf) {
-            Ok(true) => Ok(Some(output_buf)),
-            Ok(false) => {
-                reusable_buffers.push(output_buf);
-                Ok(None)
-            }
-            Err(err) => {
-                reusable_buffers.push(output_buf);
-                Err(err)
-            }
-        };
+    let result = match decode_stripe_into(
+        metrics_path,
+        stage_metrics_enabled,
+        engine,
+        workspace,
+        &mut state,
+        remaining,
+        &mut output_buf,
+    ) {
+        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(&mut shards, workspace) {
+    let reconstruct_outcome = match engine.reconstruct_into(state.shards_mut(), 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 {
-            return Err(io::Error::new(ErrorKind::UnexpectedEof, "decoded stripe is missing a data shard"));
-        };
-        let Some(shard) = slot.data_bytes() else {
+        let Some(shard) = state.data_bytes(index) 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
-                .pop_front()
-                .unwrap_or_else(|| StripeReadState::new(Vec::new(), self.read_quorum))
+            Box::new(
+                self.stripes
+                    .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() {
-        let state = StripeReadState::new(
-            vec![
-                ShardSlot::data(1, b"cd".to_vec()),
-                ShardSlot::data(0, b"ab".to_vec()),
-                ShardSlot::data(2, b"ef".to_vec()),
-            ],
-            2,
-        );
+    fn emit_data_shards_preserves_output_order() {
+        let state =
+            StripeReadState::from_parts(vec![Some(b"ab".to_vec()), Some(b"cd".to_vec()), Some(b"ef".to_vec())], Vec::new(), 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);
@@ -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 mut inline_shards = Vec::with_capacity(shards.len());
-        for shard in shards {
-            let hash = HashAlgorithm::HighwayHash256S.hash_encode(&shard);
+        let block = self.encode_data_owned_block(buf)?;
+        let mut inline_shards = Vec::with_capacity(block.shards().len());
+        for shard in block.shards() {
+            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));
-        let reader = tokio::io::BufReader::new(Cursor::new(payload.clone()));
+        for uses_legacy in [false, true] {
+            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
-            .clone()
-            .encode_inline_shards_with_size_hint(reader, payload.len())
-            .await
-            .expect("inline shards should encode");
-        let raw_shards = erasure
-            .encode_data_owned(payload.clone())
-            .expect("reference shards should encode");
+                assert_eq!(total, payload.len());
+                if payload.is_empty() {
+                    assert!(inline_shards.is_empty());
+                    continue;
+                }

-        assert_eq!(total, payload.len());
-        assert_eq!(inline_shards.len(), DATA_SHARDS + PARITY_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());
-            writer.write(&raw).await.expect("reference writer should accept shard");
-            writer.shutdown().await.expect("reference writer should shutdown");
-            assert_eq!(inline.as_ref(), writer.into_inline_data().expect("reference writer should retain bytes"));
+                let raw_shards = erasure.encode_data(&payload).expect("reference shards should encode");
+                assert_eq!(inline_shards.len(), DATA_SHARDS + PARITY_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());
+                    writer.write(&raw).await.expect("reference writer should accept shard");
+                    writer.shutdown().await.expect("reference writer should shutdown");
+                    assert_eq!(inline.as_ref(), writer.into_inline_data().expect("reference writer should retain bytes"));
+                }
+            }
        }
    }

@@ -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 {
-            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);
+        let mut data_buffer = BytesMut::with_capacity(self.encoded_capacity_for_data_len(data.len()));
        data_buffer.extend_from_slice(data);
-        data_buffer.resize(need_total_size, 0u8);
-
-        {
-            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,
-        })
+        self.encode_buffer(data_buffer, data.len())
    }

    /// 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 {
-            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(vec![Bytes::new(); self.total_shard_count()]);
-        }
-        let need_total_size = per_shard_size * self.total_shard_count();
+        self.encode_data_owned_block_inner(data)
+            .map(|block| block.into_shards(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() {
-            Ok(mut bm) => {
-                bm.resize(need_total_size, 0u8);
-                bm
-            }
+        let data_buffer = match Bytes::from(data).try_into_mut() {
+            Ok(data_buffer) => data_buffer,
            Err(b) => {
                // Rare path: refcount != 1, fall back to copy
-                let mut bm = BytesMut::with_capacity(need_total_size);
-                bm.extend_from_slice(&b);
-                bm.resize(need_total_size, 0u8);
-                bm
+                let mut data_buffer = BytesMut::with_capacity(self.encoded_capacity_for_data_len(data_len));
+                data_buffer.extend_from_slice(&b);
+                data_buffer
            }
        };
-
-        {
-            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)
+        self.encode_buffer(data_buffer, data_len)
    }

    /// 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);
-
-            assert_owned_encode_matches_borrowed(&erasure, Vec::new());
-            assert_owned_encode_matches_borrowed(&erasure, b"small payload".to_vec());
-            assert_owned_encode_matches_borrowed(&erasure, (0_u8..37).collect());
+            for data in [
+                Vec::new(),
+                vec![0xA5; 1],
+                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] {
-            let data = (0..data_len).map(|i| i as u8).collect::<Vec<_>>();
-            let expected = erasure.encode_data(&data).expect("public encode should succeed");
-            let borrowed = erasure
-                .encode_data_block(&data)
-                .expect("borrowed streaming encode should succeed");
-            let owned = erasure
-                .encode_data_bytes_mut_block(BytesMut::from(&data[..]), data.len())
-                .expect("BytesMut streaming encode should succeed");
+            for data_len in [0, 1, 63, 64] {
+                let data = (0..data_len).map(|i| i as u8).collect::<Vec<_>>();
+                let expected = erasure.encode_data(&data).expect("public encode should succeed");
+                let borrowed = erasure
+                    .encode_data_block(&data)
+                    .expect("borrowed streaming encode should succeed");
+                let owned = erasure
+                    .encode_data_owned_block(data.clone())
+                    .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!(owned.shards().eq(expected.iter().map(Bytes::as_ref)));
-            assert_eq!(borrowed.shards().len(), 16);
-            assert_eq!(borrowed.queued_bytes(), owned.queued_bytes());
+                assert_eq!(borrowed.queued_bytes(), owned.queued_bytes());
+                assert_eq!(borrowed.queued_bytes(), bytes_mut.queued_bytes());

-            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()));
+                if data_len == 0 {
+                    assert!(expected.iter().all(Bytes::is_empty));
+                    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!(