perf(ecstore): stop zeroing pooled shard buffers on the GET path (backlog#1159) (#4681)

`ShardBufferPool::take` handed out a `resize(len, 0)`-ed buffer, and the
reader then overwrote every byte of it. CPU profiling of a cached 1 MiB
GET (device reads = 0, so all cost is CPU) attributed 4.81% of the whole
server to that memset — a buffer pool exists to reuse an allocation, and
memsetting it gives the saving straight back.

The zeroing was load-bearing only because `BitrotReader::read` takes
`&mut [u8]`, which must be initialized. But the reader never reads what
the caller put there, and never returns a partially filled buffer: both
the hashed and the no-hash path either fill the whole shard or fail with
UnexpectedEof, and a hash mismatch is an error rather than a short read.
So the initialization bought nothing observable.

Add `BitrotReader::read_appending(&mut Vec<u8>, want)`, which appends into
the buffer's spare capacity instead of demanding initialized bytes:

  * hashed path — unchanged single copy, `extend_from_slice(data)` in place
    of `copy_from_slice` into a pre-zeroed buffer, and only after the hash
    verifies, so corrupt bytes never reach the caller's buffer;
  * no-hash path — `read_buf` writes straight into the spare capacity and
    advances the length only over bytes the reader actually wrote, so an
    uninitialized tail can never be exposed.

`ShardBufferPool::take` now yields an empty buffer with capacity, and
`read_shard` no longer needs to `truncate`. `read` keeps its old signature
for the remaining callers.

Four tests gate the contract rather than the call:
  * `read_appending` is byte-for-byte identical to `read` on both paths;
  * a truncated shard is UnexpectedEof, never a partially filled buffer;
  * bytes that fail the bitrot hash never reach the caller's buffer;
  * `want > shard_size` is rejected;
plus the pool test now asserts the allocation is reused (same pointer) and
never zeroed.

Verified: `erasure::` 213 passed, 0 failed; on a real Linux host
`erasure::` 209 and `disk::local::` 143 pass serially, and the failures
seen in a parallel full-suite run reproduce identically on unmodified main
(they are ENOSPC from a full root filesystem plus pre-existing flakes).

Not claimed: an end-to-end throughput number. The A/B on the bench host was
too noisy to attribute (one rep pair was not fully cached, and its root
filesystem filled mid-run); what is measured is that the removed memset was
4.81% of GET CPU in the pre-change profile.

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

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

@@ -49,15 +49,24 @@ impl ShardBufferPool {
         }
     }
 
-    // Returned bytes may contain stale scratch data and must be overwritten before use.
+    /// An **empty** buffer with room for at least `len` bytes. The caller fills it
+    /// by appending (see `BitrotReader::read_appending`), so the pool never has to
+    /// initialize the bytes it hands out.
+    ///
+    /// Zeroing here was pure waste: `read_appending` overwrites every byte it
+    /// returns, and a short read is an error rather than a partial buffer, so no
+    /// caller ever observes a byte the reader did not write. At 1 MiB shards the
+    /// `resize(len, 0)` was ~4.8% of GET CPU (rustfs/backlog#1159) — a buffer pool
+    /// exists to reuse an allocation, and memsetting it gives that saving straight
+    /// back.
     #[inline]
     pub(crate) fn take(&mut self, index: usize, len: usize) -> Vec<u8> {
         self.ensure_slots(index + 1);
         let mut buf = self.buffers[index].take().unwrap_or_else(|| Vec::with_capacity(len));
+        buf.clear();
         if buf.capacity() < len {
-            buf.reserve_exact(len - buf.capacity());
+            buf.reserve_exact(len - buf.len());
         }
-        buf.resize(len, 0);
         buf
     }
 
@@ -77,21 +86,26 @@ impl ShardBufferPool {
 mod tests {
     use super::*;
 
+    /// `take` hands out capacity, never length: the caller appends every byte it
+    /// will read back. Reusing a slot must keep the allocation and must not memset
+    /// it (rustfs/backlog#1159).
     #[test]
-    fn shard_buffer_pool_reuses_slot_without_clearing() {
+    fn shard_buffer_pool_reuses_the_allocation_and_never_zeroes_it() {
         let mut pool = ShardBufferPool::new(2);
         let mut buf = pool.take(1, 16);
-        assert_eq!(buf.len(), 16);
-        buf[0] = 42;
+        assert_eq!(buf.len(), 0, "take yields an empty buffer; the caller appends");
+        assert!(buf.capacity() >= 16);
+        buf.extend_from_slice(&[42u8; 16]);
         let capacity = buf.capacity();
+        let ptr = buf.as_ptr();
 
         pool.put(1, buf);
         assert_eq!(pool.stored_capacity(1), Some(capacity));
 
         let reused = pool.take(1, 8);
-        assert_eq!(reused.len(), 8);
-        assert!(reused.capacity() >= capacity);
-        assert_eq!(reused[0], 42);
+        assert_eq!(reused.len(), 0);
+        assert!(reused.capacity() >= capacity, "the allocation must be reused, not reallocated");
+        assert_eq!(reused.as_ptr(), ptr, "same allocation");
     }
 
     #[test]
@@ -99,7 +113,8 @@ mod tests {
         let mut pool = ShardBufferPool::new(0);
         let buf = pool.take(3, 4);
 
-        assert_eq!(buf.len(), 4);
+        assert_eq!(buf.len(), 0);
+        assert!(buf.capacity() >= 4);
         assert_eq!(pool.buffers.len(), 4);
     }
 
@@ -112,7 +127,7 @@ mod tests {
         pool.put(0, Vec::with_capacity(2));
         let grown = pool.take(0, 8);
 
-        assert_eq!(grown.len(), 8);
-        assert!(grown.capacity() >= 8);
+        assert_eq!(grown.len(), 0);
+        assert!(grown.capacity() >= 8, "a too-small reused slot must grow to the requested capacity");
     }
 }

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

@@ -132,6 +132,84 @@ where
         Ok(out.len())
     }
 
+    /// Same contract as [`Self::read`], but **appends** `want` bytes into `out`'s
+    /// spare capacity instead of demanding an initialized `&mut [u8]`
+    /// (rustfs/backlog#1159).
+    ///
+    /// `read` forced its caller to hand over a zeroed buffer purely to satisfy
+    /// `&mut [u8]`, and every one of those bytes was then overwritten. Because a
+    /// short read is an error here (never a partially filled buffer), `out` ends
+    /// up holding exactly the bytes this reader produced, so nothing the reader
+    /// did not write is ever observable — the zeroing bought nothing.
+    ///
+    /// On return `out.len()` has grown by exactly the returned count.
+    pub async fn read_appending(&mut self, out: &mut Vec<u8>, want: usize) -> std::io::Result<usize> {
+        use bytes::BufMut as _;
+        use tokio::io::AsyncReadExt as _;
+
+        self.last_verify_duration = Duration::ZERO;
+        if want > self.shard_size {
+            return Err(std::io::Error::new(
+                std::io::ErrorKind::InvalidInput,
+                format!("data size {want} exceeds shard size {}", self.shard_size),
+            ));
+        }
+        out.reserve(want);
+        let hash_size = self.hash_algo.size();
+
+        // No-hash path: read straight into `out`'s spare capacity. `read_buf`
+        // advances the length only over bytes the reader actually wrote, so an
+        // uninitialized tail can never be exposed.
+        if hash_size == 0 {
+            let start = out.len();
+            while out.len() - start < want {
+                let remaining = want - (out.len() - start);
+                let n = self
+                    .inner
+                    .read_buf(&mut (&mut *out).limit(remaining))
+                    .await
+                    .inspect_err(|e| {
+                        error!("bitrot reader read error: {}", e);
+                    })?;
+                if n == 0 {
+                    break;
+                }
+            }
+            return self.finish_len(out.len() - start, want);
+        }
+
+        // Hashed path: identical to `read` — one pass pulls `[hash][data]` into
+        // the scratch buffer — except the shard lands in `out` by `extend_from_slice`
+        // rather than `copy_from_slice` into a pre-zeroed buffer. Same single copy.
+        let need = hash_size + want;
+        if self.buf.len() < need {
+            self.buf.resize(need, 0);
+        }
+        let filled = fill(&mut self.inner, &mut self.buf[..need]).await?;
+        if filled < need {
+            let got_data = filled.saturating_sub(hash_size);
+            error!("bitrot reader short shard read: id={} got {} of {} bytes", self.id, got_data, want);
+            return Err(std::io::Error::new(
+                std::io::ErrorKind::UnexpectedEof,
+                format!("short shard read: got {got_data} of {want} bytes"),
+            ));
+        }
+
+        let (hash, data) = self.buf[..need].split_at(hash_size);
+        if !self.skip_verify {
+            let verify_start = std::time::Instant::now();
+            let actual_hash = self.hash_algo.hash_encode(data);
+            self.last_verify_duration = verify_start.elapsed();
+            if actual_hash.as_ref() != hash {
+                error!("bitrot reader hash mismatch, id={} data_len={}, out_len={}", self.id, data.len(), want);
+                return Err(std::io::Error::new(std::io::ErrorKind::InvalidData, "bitrot hash mismatch"));
+            }
+        }
+        // Only after verification: a corrupt shard must not reach the caller.
+        out.extend_from_slice(data);
+        Ok(want)
+    }
+
     /// Map a completed no-hash read to the shared short-shard contract: a full
     /// buffer returns its length, a short read is UnexpectedEof (backlog#799 B2).
     fn finish_len(&self, data_len: usize, want: usize) -> std::io::Result<usize> {
@@ -1152,4 +1230,96 @@ mod tests {
         let err = r.read(&mut out).await.expect_err("truncated hash must error");
         assert_eq!(err.kind(), std::io::ErrorKind::UnexpectedEof);
     }
+
+    /// `read_appending` must be byte-for-byte identical to `read`, for both the
+    /// hashed and the no-hash path (rustfs/backlog#1159). It exists so callers can
+    /// hand over an *uninitialized* buffer; if it ever diverged from `read`, the
+    /// GET path would silently return different bytes.
+    #[tokio::test]
+    async fn read_appending_matches_read_for_both_paths() {
+        for algo in [HashAlgorithm::HighwayHash256, HashAlgorithm::None] {
+            const SHARD: usize = 4096;
+            let data: Vec<u8> = (0..SHARD).map(|i| (i * 31 + 7) as u8).collect();
+
+            let mut encoded = Vec::new();
+            let mut w = BitrotWriter::new(&mut encoded, SHARD, algo.clone());
+            w.write(&data).await.expect("write shard");
+
+            let mut via_read = vec![0u8; SHARD];
+            let n1 = BitrotReader::new(Cursor::new(encoded.clone()), SHARD, algo.clone(), false)
+                .read(&mut via_read)
+                .await
+                .expect("read");
+
+            // A buffer with only capacity — no initialized bytes at all.
+            let mut via_append: Vec<u8> = Vec::with_capacity(SHARD);
+            let n2 = BitrotReader::new(Cursor::new(encoded), SHARD, algo.clone(), false)
+                .read_appending(&mut via_append, SHARD)
+                .await
+                .expect("read_appending");
+
+            assert_eq!(n1, n2, "{algo:?}: both must report the same length");
+            assert_eq!(via_append.len(), n2, "{algo:?}: the buffer grows by exactly n");
+            assert_eq!(via_read, via_append, "{algo:?}: bytes must be identical");
+            assert_eq!(via_append, data, "{algo:?}: and equal to what was written");
+        }
+    }
+
+    /// A truncated shard must be an error, never a partially filled buffer — that
+    /// contract is what lets the pool hand out uninitialized capacity.
+    #[tokio::test]
+    async fn read_appending_rejects_a_short_shard_instead_of_returning_partial_bytes() {
+        for algo in [HashAlgorithm::HighwayHash256, HashAlgorithm::None] {
+            const SHARD: usize = 4096;
+            let data = vec![9u8; SHARD];
+            let mut encoded = Vec::new();
+            let mut w = BitrotWriter::new(&mut encoded, SHARD, algo.clone());
+            w.write(&data).await.expect("write shard");
+            encoded.truncate(encoded.len() - 1);
+
+            let mut out: Vec<u8> = Vec::with_capacity(SHARD);
+            let err = BitrotReader::new(Cursor::new(encoded), SHARD, algo.clone(), false)
+                .read_appending(&mut out, SHARD)
+                .await
+                .expect_err("a truncated shard must not succeed");
+            assert_eq!(err.kind(), std::io::ErrorKind::UnexpectedEof, "{algo:?}");
+            assert!(
+                out.len() < SHARD,
+                "{algo:?}: a failed read must not claim a full shard; the caller drops the buffer"
+            );
+        }
+    }
+
+    /// A corrupt shard must fail verification, and the corrupt bytes must NOT be
+    /// appended: `read_appending` writes into a buffer the caller may recycle.
+    #[tokio::test]
+    async fn read_appending_does_not_expose_bytes_that_fail_the_hash() {
+        const SHARD: usize = 4096;
+        let algo = HashAlgorithm::HighwayHash256;
+        let data = vec![3u8; SHARD];
+        let mut encoded = Vec::new();
+        let mut w = BitrotWriter::new(&mut encoded, SHARD, algo.clone());
+        w.write(&data).await.expect("write shard");
+        let last = encoded.len() - 1;
+        encoded[last] ^= 0xff;
+
+        let mut out: Vec<u8> = Vec::with_capacity(SHARD);
+        let err = BitrotReader::new(Cursor::new(encoded), SHARD, algo, false)
+            .read_appending(&mut out, SHARD)
+            .await
+            .expect_err("a corrupt shard must not verify");
+        assert_eq!(err.kind(), std::io::ErrorKind::InvalidData);
+        assert!(out.is_empty(), "corrupt bytes must never reach the caller's buffer");
+    }
+
+    #[tokio::test]
+    async fn read_appending_rejects_a_want_larger_than_the_shard() {
+        let algo = HashAlgorithm::HighwayHash256;
+        let mut out: Vec<u8> = Vec::new();
+        let err = BitrotReader::new(Cursor::new(Vec::new()), 16, algo, false)
+            .read_appending(&mut out, 17)
+            .await
+            .expect_err("want > shard_size must be rejected");
+        assert_eq!(err.kind(), std::io::ErrorKind::InvalidInput);
+    }
 }

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

@@ -275,13 +275,15 @@ where
     let role = shard_role(index, data_shards);
     if let Some(reader) = reader {
         Box::pin(async move {
-            let mut buf = recycled_buf.unwrap_or_else(|| vec![0; shard_size]);
-            debug_assert_eq!(buf.len(), shard_size);
+            // Capacity, not length: `read_appending` writes every byte it returns, so
+            // the buffer never needs zeroing first (rustfs/backlog#1159).
+            let mut buf = recycled_buf.unwrap_or_else(|| Vec::with_capacity(shard_size));
+            buf.clear();
             let read_start = metrics_path.map(|_| Instant::now());
             let read_result = if read_timeout.is_zero() {
-                reader.read(&mut buf).await
+                reader.read_appending(&mut buf, shard_size).await
             } else {
-                match tokio::time::timeout(read_timeout, reader.read(&mut buf)).await {
+                match tokio::time::timeout(read_timeout, reader.read_appending(&mut buf, shard_size)).await {
                     Ok(result) => result,
                     Err(_) => {
                         let timeout_error = io::Error::new(ErrorKind::TimedOut, "shard read timed out");
@@ -306,7 +308,7 @@ where
 
             match read_result {
                 Ok(n) => {
-                    buf.truncate(n);
+                    debug_assert_eq!(buf.len(), n, "read_appending must grow the buffer by exactly n");
                     if let Some(path) = metrics_path {
                         rustfs_io_metrics::record_get_object_shard_read_observation(
                             path,