fix(ecstore): reject short-read shards and advance ParallelReader per stripe (backlog#799 B2) (#4327)

Two compounding defects let a truncated shard corrupt a GET (ranged GET could
return HTTP 200 with wrong bytes):

- `BitrotReader::read` returned `Ok(short_len)` when the shard stream hit EOF
  before filling the caller's buffer. With `skip_verify` /
  `HashAlgorithm::None` / parity=0 there is no hash to catch it, so the short
  shard was accepted and every downstream byte shifted.
- `ParallelReader.offset` was set once and never advanced per stripe, so every
  stripe after the first reused the first stripe's geometry and the last-stripe
  length clamp was wrong.

Fix both (they are mutually required):

- `BitrotReader::read` now errors (`UnexpectedEof`) on a short read, before and
  independent of the bitrot hash check, so it fires under skip-verify/no-hash
  too. The caller sizes the buffer to the expected per-stripe shard length, so
  "buffer not filled" == "shard truncated". A short read routes through the
  existing `errs[i]` path, dropping that reader from the stripe so parity
  reconstruction engages; with parity=0 the stripe fails read quorum and the GET
  errors loudly instead of streaming shifted bytes. Mirrors MinIO's
  `parallelReader.Read` (`n != shardSize` -> reader failed).
- `ParallelReader::read` / `read_lockstep` advance `self.offset += shard_size`
  per stripe so the per-stripe expected length (incl. the shorter final stripe)
  is exact, matching the correct pattern already used by heal.

Updates three bitrot tests that used an obsolete oversized-buffer pattern to
size per-stripe (as the real decode/heal paths do), and adds a regression test
that a truncated shard errors under None/HighwayHash + skip_verify.

Refs backlog#799 (B2), issue rustfs/backlog#851. Design converged by two
independent expert reviews referencing MinIO cmd/erasure-decode.go.

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

@@ -92,6 +92,23 @@ where
             data_len += n;
         }
 
+        // A short read (EOF before the caller-sized buffer is filled) means a
+        // truncated/incomplete shard. Return an error so the caller drops this
+        // reader from the stripe and reconstruction from parity engages, instead
+        // of silently returning fewer bytes and shifting every downstream byte
+        // (backlog#799 B2). This is deliberately BEFORE and independent of the
+        // bitrot hash check so it also fires under skip_verify /
+        // HashAlgorithm::None / parity=0, where there is no hash to catch it. The
+        // caller sizes `out` to exactly the expected shard length for the current
+        // stripe, so "buffer full" == "shard complete".
+        if data_len < out.len() {
+            error!("bitrot reader short shard read: id={} got {} of {} bytes", self.id, data_len, out.len());
+            return Err(std::io::Error::new(
+                std::io::ErrorKind::UnexpectedEof,
+                format!("short shard read: got {data_len} of {} bytes", out.len()),
+            ));
+        }
+
         if hash_size > 0 && !self.skip_verify {
             let verify_start = std::time::Instant::now();
             let actual_hash = self.hash_algo.hash_encode(&out[..data_len]);
@@ -515,7 +532,12 @@ mod tests {
         let mut out = Vec::new();
         let mut n = 0;
         while n < data_size {
-            let mut buf = vec![0u8; shard_size];
+            // Size the buffer to the expected shard length for this stripe (the
+            // last stripe is legitimately shorter); BitrotReader now requires the
+            // buffer to be filled exactly, matching how the decode/heal paths size
+            // per-stripe shard buffers (backlog#799 B2).
+            let this_size = shard_size.min(data_size - n);
+            let mut buf = vec![0u8; this_size];
             let m = bitrot_reader.read(&mut buf).await.unwrap();
             assert_eq!(&buf[..m], &data[n..n + m]);
 
@@ -550,7 +572,8 @@ mod tests {
         let mut idx = 0;
         let mut n = 0;
         while n < data_size {
-            let mut buf = vec![0u8; shard_size];
+            let this_size = shard_size.min(data_size - n);
+            let mut buf = vec![0u8; this_size];
             let res = bitrot_reader.read(&mut buf).await;
 
             if idx == count - 1 {
@@ -591,7 +614,12 @@ mod tests {
         let mut out = Vec::new();
         let mut n = 0;
         while n < data_size {
-            let mut buf = vec![0u8; shard_size];
+            // Size the buffer to the expected shard length for this stripe (the
+            // last stripe is legitimately shorter); BitrotReader now requires the
+            // buffer to be filled exactly, matching how the decode/heal paths size
+            // per-stripe shard buffers (backlog#799 B2).
+            let this_size = shard_size.min(data_size - n);
+            let mut buf = vec![0u8; this_size];
             let m = bitrot_reader.read(&mut buf).await.unwrap();
             assert_eq!(&buf[..m], &data[n..n + m]);
             out.extend_from_slice(&buf[..m]);
@@ -601,6 +629,40 @@ mod tests {
         assert_eq!(data, &out[..]);
     }
 
+    #[tokio::test]
+    async fn bitrot_read_short_shard_errors_even_when_skip_verify() {
+        // A truncated shard (fewer bytes than the caller's expected-size buffer)
+        // must be an error, not a silent Ok(short) — including on the skip-verify
+        // / no-hash paths where there is no bitrot hash to catch it. This is the
+        // core B2 fix (backlog#799): a short read is a shard error so the decoder
+        // drops it and reconstructs from parity instead of shifting downstream
+        // bytes.
+        let shard_size = 16usize;
+        for (algo, skip_verify) in [
+            (HashAlgorithm::None, true),
+            (HashAlgorithm::None, false),
+            (HashAlgorithm::HighwayHash256, true),
+        ] {
+            let label = format!("{algo:?}");
+            let writer = Cursor::new(Vec::<u8>::new());
+            let mut w = BitrotWriter::new(writer, shard_size, algo.clone());
+            w.write(&[7u8; 16]).await.unwrap();
+            let written = w.into_inner().into_inner();
+            // Drop the last 4 data bytes so the shard is truncated.
+            let truncated = written[..written.len() - 4].to_vec();
+
+            let mut r = BitrotReader::new(Cursor::new(truncated), shard_size, algo, skip_verify);
+            let mut out = vec![0u8; shard_size];
+            let res = r.read(&mut out).await;
+            assert!(res.is_err(), "short shard must error (algo={label}, skip_verify={skip_verify})");
+            assert_eq!(
+                res.unwrap_err().kind(),
+                std::io::ErrorKind::UnexpectedEof,
+                "short shard must be UnexpectedEof (algo={label}, skip_verify={skip_verify})"
+            );
+        }
+    }
+
     #[tokio::test]
     async fn test_bitrot_writer_flushes_once_on_shutdown() {
         let flushes = Arc::new(AtomicUsize::new(0));

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

@@ -629,6 +629,14 @@ where
             return (vec![None; num_readers], vec![None; num_readers]);
         }
 
+        // Advance to the next stripe so the following read() computes the correct
+        // (possibly shorter, final) stripe length. `offset` previously never
+        // advanced, so every stripe after the first reused the first stripe's
+        // geometry; combined with BitrotReader now rejecting short reads, the
+        // per-stripe expected length must be exact (backlog#799 B2). `self.offset`
+        // is only read above to derive `shard_size`, so advancing here is safe.
+        self.offset += shard_size;
+
         let mut shards: Vec<Option<Vec<u8>>> = vec![None; num_readers];
         let mut errs = vec![None; num_readers];
         let read_costs = self.read_costs.as_slice();
@@ -1001,6 +1009,10 @@ where
             return (shards, errs);
         }
 
+        // Advance to the next stripe (see the matching note in `read`); the
+        // lockstep path must track stripe geometry identically (backlog#799 B2).
+        self.offset += shard_size;
+
         self.buffers.ensure_slots(num_readers);
         // Pre-claim per-slot buffers so the `self.readers` borrow below stays
         // disjoint from `self.buffers`.