From bd5d3c5d92a0aa70a7d92da3e48761d6e61f0dc9 Mon Sep 17 00:00:00 2001 From: houseme Date: Wed, 8 Jul 2026 07:15:49 +0800 Subject: [PATCH] perf(ecstore): data-shards-only lockstep GET reads with stripe-aligned deferred parity engagement (opt-in) (#4392) * feat(ecstore): add stripe-advance handles for deferred bitrot readers Give DeferredObjectReader a shared pending state and expose a DeferredReaderStripeHandle that advances the still-unopened source by whole bitrot blocks using the same bitrot_encoded_range geometry the reader was created with (identity mapping when hash_size == 0). This lets the GET decode path open a parity shard aligned to the stripe where a data shard failed instead of reading every parity shard on every stripe (backlog#923). An already-opened (or failed) reader rejects the advance so callers retire it rather than engage it out of alignment; bitrot verification after an advance checks the advanced stripe's block against that stripe's stored hash. Co-Authored-By: heihutu * perf(ecstore): read only data shards on healthy lockstep GET behind opt-in gate PR #4289's lockstep fix made every reconstruction-verifying GET read all data+parity shards per stripe; the parity blocks are read, bitrot-hashed and then discarded, a deterministic 2x read-bytes/IOPS/hash-CPU amplification on healthy 2+2 objects (backlog#923). With the new opt-in gate RUSTFS_GET_LOCKSTEP_DATA_SHARDS_ONLY_ENABLE=true (default: false, behavior identical to main): - read_lockstep keeps only the data slots engaged while the object is healthy; parity slots stay unopened deferred readers. - When a data shard is missing or dies at stripe k, parity readers are engaged mid-object by advancing their deferred stripe handle to stripe k, preserving the lockstep alignment invariant from backlog#832. - Degraded stripes engage one parity beyond the decode quorum so reconstruction verification keeps an extra source to check against (erasure.rs only verifies when available > data shards); an engaged parity reader that errors is retired for the rest of the object like any other, and a parity reader that cannot be realigned is retired instead of being read out of position. - fill_deferred_bitrot_readers records stripe handles for deferred slots and, gate-on only, swaps eagerly opened parity readers for unopened deferred ones so they remain engageable mid-object; ready/error bookkeeping used by quorum decisions is untouched. - Both GET paths (legacy duplex via Erasure::decode_with_stripe_handles, codec streaming via ParallelReader::with_deferred_parity_handles) carry the handles from reader setup. Short-read -> UnexpectedEof -> whole-object retirement and the inconsistent-source rejection are unchanged in both gate modes; tests lock the healthy-path data-shards-only call counts, the default read-all-shards behavior, mid-object parity engagement for streaming and hash_size==0 formats, and mid-stream inconsistent-parity rejection. Co-Authored-By: heihutu --------- Co-authored-by: heihutu --- crates/ecstore/src/erasure/coding/decode.rs | 547 ++++++++++++++++-- crates/ecstore/src/io_support/bitrot.rs | 238 +++++++- .../src/set_disk/core/io_primitives.rs | 81 ++- crates/ecstore/src/set_disk/read.rs | 77 ++- 4 files changed, 871 insertions(+), 72 deletions(-) diff --git a/crates/ecstore/src/erasure/coding/decode.rs b/crates/ecstore/src/erasure/coding/decode.rs index 1ad448379..fb12649f4 100644 --- a/crates/ecstore/src/erasure/coding/decode.rs +++ b/crates/ecstore/src/erasure/coding/decode.rs @@ -24,6 +24,7 @@ use crate::disk::error::Error; 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 futures::FutureExt; use futures::stream::{FuturesUnordered, StreamExt}; @@ -112,6 +113,26 @@ fn get_decode_stripe_prefetch_count() -> usize { const ENV_RUSTFS_GET_BITROT_DECODE_OVERLAP_ENABLE: &str = "RUSTFS_GET_BITROT_DECODE_OVERLAP_ENABLE"; const DEFAULT_RUSTFS_GET_BITROT_DECODE_OVERLAP_ENABLE: bool = false; +/// Enable the data-shards-only lockstep GET read (backlog#923). +/// When enabled, the reconstruction-verifying (lockstep) GET path reads only +/// the data shards per stripe while the object is healthy; parity slots stay +/// unopened deferred readers that are engaged — realigned to the failing +/// stripe via their stripe handle — only when a data shard is missing or dies. +/// This halves per-GET read bytes, IOPS, and bitrot hashing on healthy +/// objects with 2+2 layouts. +/// Default: false (current behavior: every live shard reader is read on every +/// stripe). +const ENV_RUSTFS_GET_LOCKSTEP_DATA_SHARDS_ONLY_ENABLE: &str = "RUSTFS_GET_LOCKSTEP_DATA_SHARDS_ONLY_ENABLE"; +const DEFAULT_RUSTFS_GET_LOCKSTEP_DATA_SHARDS_ONLY_ENABLE: bool = false; + +/// Whether the data-shards-only lockstep GET read is enabled (backlog#923). +pub(crate) fn get_lockstep_data_shards_only_enabled() -> bool { + rustfs_utils::get_env_bool( + ENV_RUSTFS_GET_LOCKSTEP_DATA_SHARDS_ONLY_ENABLE, + DEFAULT_RUSTFS_GET_LOCKSTEP_DATA_SHARDS_ONLY_ENABLE, + ) +} + /// Get whether bitrot-decode overlap is enabled. fn is_bitrot_decode_overlap_enabled() -> bool { rustfs_utils::get_env_bool( @@ -327,6 +348,14 @@ pub(crate) struct ParallelReader { // Request-scoped shard buffers keyed by shard index. Keeping ownership in // `ParallelReader` avoids dropping unused parity/backup slot buffers between stripes. buffers: ShardBufferPool, + // 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 + // parity stays an unopened deferred reader; `deferred_handles[i]` realigns + // it to the current stripe when it is engaged mid-object (backlog#923). + engaged: Vec, + deferred_handles: Vec>, + stripe_index: usize, } } @@ -502,6 +531,14 @@ where // Ensure offset does not exceed shard_file_size + // Default (gate off): every slot is engaged, i.e. the lockstep path + // reads all live readers on every stripe — the pre-backlog#923 + // behavior. With the gate on, only data slots start engaged; parity is + // engaged on demand, stripe-aligned through its deferred handle. + let data_shards_only = get_lockstep_data_shards_only_enabled(); + let engaged = (0..readers.len()) + .map(|index| !data_shards_only || index < e.data_shards) + .collect(); ParallelReader { readers, offset, @@ -515,8 +552,23 @@ where verify_reconstruction, locality_preference_enabled: get_shard_locality_preference_enabled(), buffers: ShardBufferPool::new(e.data_shards + e.parity_shards), + engaged, + deferred_handles: Vec::new(), + stripe_index: 0, } } + + /// Attach the per-slot deferred stripe handles produced during bitrot + /// reader setup. Only parity slots are consulted: a handle lets the + /// lockstep path open a parity shard aligned to the stripe where a data + /// shard failed, instead of reading every parity shard on every stripe + /// (backlog#923). + pub(crate) fn with_deferred_parity_handles(mut self, mut handles: Vec>) -> Self { + handles.resize_with(self.readers.len(), || None); + handles.truncate(self.readers.len()); + self.deferred_handles = handles; + self + } } #[allow(clippy::too_many_arguments)] @@ -983,18 +1035,29 @@ where /// Lockstep stripe read for the reconstruction-verifying GET path. /// - /// Reads every still-live shard reader exactly once per stripe and waits for - /// all of them, so all readers advance by one block per stripe and remain - /// mutually block-aligned. This removes the desync that produced - /// "inconsistent read source shards" (backlog#832): with the adaptive - /// data-first path a parity reader pulled in as a substitute mid-object was - /// still at its stream start and returned an earlier stripe than the data - /// shards, so reconstruction verification correctly rejected the mismatch - /// and the large-object GET truncated mid-stream. + /// Reads every *engaged* shard reader exactly once per stripe and waits for + /// all of them, so all engaged readers advance by one block per stripe and + /// remain mutually block-aligned. This preserves the alignment invariant + /// that removed the "inconsistent read source shards" desync (backlog#832): + /// a reader must never contribute a block from a different stripe than the + /// rest of the set. /// - /// Any reader that errors is retired for the rest of the object: a streaming - /// shard read that failed mid-block can no longer be trusted to be aligned, - /// so re-reading it on a later stripe would reintroduce the desync. + /// While the object is healthy only the data shards are engaged, so a + /// healthy GET reads exactly `data_shards` shards per stripe instead of all + /// `data + parity` shards (backlog#923). Parity slots stay unopened + /// deferred readers; when a data shard is missing or dies at stripe `k`, + /// enough parity readers are engaged — realigned to stripe `k` through + /// their [`DeferredReaderStripeHandle`] — to restore the decode quorum + /// *plus one extra shard* so reconstruction verification still has a + /// source to check against (`erasure.rs` only verifies when + /// `available_shards > data_shards`). + /// + /// Any reader that errors is retired for the rest of the object — data and + /// newly engaged parity alike: a streaming shard read that failed mid-block + /// can no longer be trusted to be aligned, so re-reading it on a later + /// 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) -> (Vec>>, Vec>) { let num_readers = self.readers.len(); let shard_size = if self.offset + self.shard_size > self.shard_file_size { @@ -1012,14 +1075,42 @@ where // 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; + let stripe_index = self.stripe_index; + self.stripe_index += 1; self.buffers.ensure_slots(num_readers); + + // Engage parity up front when data slots are already known to be + // missing (offline at setup or retired on an earlier stripe), so the + // substitute reads run in parallel with the surviving data reads. + let missing_data_readers = self.readers.iter().take(self.data_shards).filter(|r| r.is_none()).count(); + if missing_data_readers > 0 { + // One extra engaged parity beyond the reconstruction quorum keeps + // reconstruction verification effective: with exactly + // `data_shards` available shards the verification would silently + // turn itself off. + let want = missing_data_readers + 1; + let mut have = (self.data_shards..num_readers) + .filter(|&i| self.engaged[i] && self.readers[i].is_some()) + .count(); + for idx in self.data_shards..num_readers { + if have >= want { + break; + } + if self.readers[idx].is_some() && !self.engaged[idx] && self.try_engage_parity(idx, stripe_index) { + have += 1; + } + } + } + // Pre-claim per-slot buffers so the `self.readers` borrow below stays // disjoint from `self.buffers`. - let has_reader: Vec = self.readers.iter().map(Option::is_some).collect(); + let participating: Vec = (0..num_readers) + .map(|i| self.engaged[i] && self.readers[i].is_some()) + .collect(); let mut bufs: Vec>> = Vec::with_capacity(num_readers); - for (i, present) in has_reader.iter().enumerate() { - bufs.push(if *present { + for (i, participates) in participating.iter().enumerate() { + bufs.push(if *participates { Some(self.buffers.take(i, shard_size)) } else { None @@ -1045,7 +1136,7 @@ where let mut sets = FuturesUnordered::new(); let reader_iter = ReaderLaunchIter::new(&mut self.readers, &read_costs, locality_preference_enabled); for (i, reader) in reader_iter { - if reader.is_none() { + if reader.is_none() || !participating[i] { continue; } let read_cost = read_costs.get(i).copied().unwrap_or(ShardReadCost::Unknown); @@ -1085,6 +1176,51 @@ where } } + // A data shard may have died during this stripe's reads. The unengaged + // parity readers are still unopened, so they can be aligned to *this* + // stripe and read now: engage them one at a time until the stripe has + // `data_shards + 1` successful shards (decode quorum plus one + // reconstruction-verification source) or parity runs out. + loop { + let success_now = shards.iter().filter(|shard| shard.is_some()).count(); + let data_shard_missing = shards.iter().take(data_shards).any(|shard| shard.is_none()); + if !data_shard_missing || success_now > data_shards { + break; + } + let Some(idx) = (data_shards..num_readers).find(|&i| self.readers[i].is_some() && !self.engaged[i]) else { + break; + }; + if !self.try_engage_parity(idx, stripe_index) { + continue; + } + let read_cost = read_costs.get(idx).copied().unwrap_or(ShardReadCost::Unknown); + let recycled_buf = Some(self.buffers.take(idx, shard_size)); + scheduled += 1; + let (i, _read_cost, result, _should_retire) = read_shard( + idx, + read_cost, + &mut self.readers[idx], + recycled_buf, + shard_size, + data_shards, + read_timeout, + metrics_path, + ) + .await; + completed += 1; + match result { + Ok(v) => { + shards[i] = Some(v); + success += 1; + } + Err(e) => { + errs[i] = Some(e); + retire_readers.push(i); + failed += 1; + } + } + } + if let Some(path) = metrics_path { record_get_stage_duration_if_enabled(path, GET_STAGE_STRIPE_READ_QUORUM, stripe_read_start); rustfs_io_metrics::record_get_object_shard_read_fanout(path, scheduled, completed, success, failed); @@ -1097,6 +1233,40 @@ where (shards, errs) } + /// Attempt to bring an as-yet-unread parity reader into the lockstep read + /// set at `stripe_index`. + /// + /// At stripe 0 every reader is still positioned at the stream start, so + /// engagement is trivially aligned. Past stripe 0 the parity reader must + /// still be an unopened deferred reader: its pending open offset is + /// advanced by `stripe_index` bitrot blocks (the `bitrot_encoded_range` + /// geometry) so its first read returns the current stripe. A parity reader + /// that cannot be realigned is retired for the rest of the object, + /// mirroring the retire-on-error rule: reading it would return an earlier + /// stripe and reintroduce the backlog#832 desync. + fn try_engage_parity(&mut self, idx: usize, stripe_index: usize) -> bool { + if stripe_index == 0 { + self.engaged[idx] = true; + return true; + } + let advanced = self + .deferred_handles + .get(idx) + .and_then(|handle| handle.as_ref()) + .is_some_and(|handle| handle.advance_stripes(stripe_index)); + if advanced { + self.engaged[idx] = true; + true + } else { + warn!( + shard_index = idx, + stripe_index, "retiring parity reader that cannot be aligned to the current stripe" + ); + self.readers[idx] = None; + false + } + } + pub fn recycle_shards(&mut self, shards: &mut [Option>]) { for (i, reader) in self.readers.iter().enumerate() { if reader.is_some() @@ -1274,7 +1444,8 @@ impl Erasure { W: AsyncWrite + Send + Sync + Unpin, R: AsyncRead + Unpin + Send + Sync, { - self.decode_inner(writer, readers, offset, length, total_length, None).await + self.decode_inner(writer, readers, offset, length, total_length, None, Vec::new()) + .await } pub(crate) async fn decode_with_read_costs( @@ -1290,10 +1461,33 @@ impl Erasure { W: AsyncWrite + Send + Sync + Unpin, R: AsyncRead + Unpin + Send + Sync, { - self.decode_inner(writer, readers, offset, length, total_length, Some(read_costs)) + self.decode_inner(writer, readers, offset, length, total_length, Some(read_costs), Vec::new()) .await } + /// GET decode entry point that also carries the deferred-parity stripe + /// handles from bitrot reader setup, so unengaged parity readers can be + /// opened aligned to the stripe where a data shard fails (backlog#923). + #[allow(clippy::too_many_arguments)] + pub(crate) async fn decode_with_stripe_handles( + &self, + writer: &mut W, + readers: Vec>>, + offset: usize, + length: usize, + total_length: usize, + read_costs: Option>, + deferred_handles: Vec>, + ) -> (usize, Option) + where + W: AsyncWrite + Send + Sync + Unpin, + R: AsyncRead + Unpin + Send + Sync, + { + self.decode_inner(writer, readers, offset, length, total_length, read_costs, deferred_handles) + .await + } + + #[allow(clippy::too_many_arguments)] async fn decode_inner( &self, writer: &mut W, @@ -1302,6 +1496,7 @@ impl Erasure { length: usize, total_length: usize, read_costs: Option>, + deferred_handles: Vec>, ) -> (usize, Option) where W: AsyncWrite + Send + Sync + Unpin, @@ -1349,7 +1544,8 @@ impl Erasure { ) } else { ParallelReader::new_for_decode(readers, self.clone(), offset, total_length, Some(GET_OBJECT_PATH_LEGACY_DUPLEX)) - }; + } + .with_deferred_parity_handles(deferred_handles); let start = offset / self.block_size; let end = end_offset.saturating_sub(1) / self.block_size; @@ -1479,9 +1675,82 @@ mod tests { use rustfs_utils::HashAlgorithm; use std::io::Cursor; use std::pin::Pin; + use std::sync::{ + Arc, + atomic::{AtomicUsize, Ordering}, + }; use std::task::{Context, Poll}; use tokio::io::ReadBuf; + type BoxedShardReader = Box; + + /// 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 { + inner: Cursor>, + bytes_read: Arc, + } + + impl AsyncRead for CountingShardReader { + fn poll_read(mut self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll> { + let before = buf.filled().len(); + let result = Pin::new(&mut self.inner).poll_read(cx, buf); + if let Poll::Ready(Ok(())) = result { + let delta = buf.filled().len() - before; + self.bytes_read.fetch_add(delta, Ordering::SeqCst); + } + result + } + } + + /// Build a production-shaped reader set: data shards as opened stream + /// readers, parity shards as *unopened* deferred readers carrying stripe + /// handles (mirroring `fill_deferred_bitrot_readers`). `truncate` shortens + /// the given data shard buffers to simulate a reader dying mid-object. + fn readers_with_deferred_parity( + shard_bufs: &[Vec], + data_shards: usize, + shard_size: usize, + hash_algo: &HashAlgorithm, + truncate: &[(usize, usize)], + ) -> (Vec>>, Vec>) { + use crate::io_support::bitrot::create_deferred_bitrot_reader_with_stripe_handle; + + let mut readers = Vec::with_capacity(shard_bufs.len()); + let mut handles: Vec> = vec![None; shard_bufs.len()]; + for (i, buf) in shard_bufs.iter().enumerate() { + if i < data_shards { + let bytes = truncate + .iter() + .find(|(index, _)| *index == i) + .map(|(_, len)| buf[..*len].to_vec()) + .unwrap_or_else(|| buf.clone()); + readers.push(Some(BitrotReader::new( + Box::new(Cursor::new(bytes)) as BoxedShardReader, + shard_size, + hash_algo.clone(), + false, + ))); + } else { + let (reader, handle) = create_deferred_bitrot_reader_with_stripe_handle( + Some(bytes::Bytes::from(buf.clone())), + None, + "test-bucket", + "test-object", + 0, + buf.len(), + shard_size, + hash_algo.clone(), + false, + false, + ); + readers.push(Some(reader)); + handles[i] = Some(handle); + } + } + (readers, handles) + } + enum TestShardReader { Ready(Cursor>), Pending, @@ -1722,17 +1991,9 @@ mod tests { assert!(output.is_empty()); } - /// Regression for backlog#832: a data shard reader that dies partway through - /// a multi-stripe object must still reconstruct byte-exact output. The parity - /// readers that fill in for the failed data shard must be aligned to the - /// current stripe. Before the lockstep read path, parity readers were only - /// pulled in on demand and were still positioned at their stream start - /// (block 0), so they returned an earlier stripe than the surviving data - /// shards — every shard passed its own bitrot hash yet the set was mutually - /// inconsistent, so reconstruction verification rejected it ("inconsistent - /// read source shards") and the large-object GET truncated mid-stream. - #[tokio::test] - async fn test_erasure_decode_recovers_when_data_shard_dies_midway() { + /// One mid-object data-shard death scenario, shared by the gate-on and + /// gate-off variants of `test_erasure_decode_recovers_when_data_shard_dies_midway`. + async fn run_decode_midway_death_case(hash_algo: HashAlgorithm, context: &str) { const DATA_SHARDS: usize = 4; const PARITY_SHARDS: usize = 2; const BLOCK_SIZE: usize = 64; @@ -1745,7 +2006,6 @@ mod tests { let erasure = Erasure::new(DATA_SHARDS, PARITY_SHARDS, BLOCK_SIZE); let total_shards = DATA_SHARDS + PARITY_SHARDS; let shard_size = erasure.shard_size(); - let hash_algo = HashAlgorithm::HighwayHash256; let hash_size = hash_algo.size(); let mut shard_writers: Vec>>> = (0..total_shards) @@ -1764,29 +2024,228 @@ mod tests { let shard_bufs: Vec> = shard_writers.into_iter().map(|w| w.into_inner().into_inner()).collect(); - // Truncate data shard 0 to only its first (hash+data) block: it reads the - // first stripe successfully, then errors (UnexpectedEof reading the next - // block's hash), forcing every later stripe to be reconstructed from the - // parity shards. With all readers present the reconstruction must succeed. + // Truncate data shard 0 to only its first (hash+data) block: it reads + // the first stripe successfully, then errors (UnexpectedEof), forcing + // every later stripe to engage the deferred parity readers aligned to + // the failing stripe. let first_block_len = (hash_size + shard_size).min(shard_bufs[0].len()); - let readers: Vec>>>> = shard_bufs + let (readers, handles) = + readers_with_deferred_parity(&shard_bufs, DATA_SHARDS, shard_size, &hash_algo, &[(0, first_block_len)]); + + let mut output = Vec::new(); + let (written, err) = erasure + .decode_with_stripe_handles(&mut output, readers, 0, total_len, total_len, None, handles) + .await; + assert!( + err.is_none(), + "{context}, algo={hash_algo:?}: mid-object data-shard failure must still reconstruct: {err:?}" + ); + assert_eq!( + written, total_len, + "{context}, algo={hash_algo:?}: short write after mid-object shard failure" + ); + assert_eq!( + output, total_data, + "{context}, algo={hash_algo:?}: reconstructed bytes mismatch (stripe desync?)" + ); + } + + /// Regression for backlog#832 (extended for backlog#923): a data shard + /// reader that dies partway through a multi-stripe object must still + /// reconstruct byte-exact output, and the parity readers that fill in for + /// it must be aligned to the current stripe. With the data-shards-only + /// gate on, parity slots are unopened deferred readers whose stripe handle + /// advances the pending open offset to the failing stripe — mirroring the + /// production GET reader setup. Covers both the streaming per-block + /// checksum format and the no-per-block-hash (`hash_size == 0`) format, + /// where the offset mapping degrades to the identity function (merge gate + /// for backlog#923). Also run with the gate off to lock the default + /// (read-all-shards) behavior. + #[tokio::test] + #[serial_test::serial] + async fn test_erasure_decode_recovers_when_data_shard_dies_midway() { + for enabled in [None, Some("true")] { + temp_env::async_with_vars([(ENV_RUSTFS_GET_LOCKSTEP_DATA_SHARDS_ONLY_ENABLE, enabled)], async { + let context = if enabled.is_some() { "data-shards-only" } else { "default" }; + for hash_algo in [HashAlgorithm::HighwayHash256, HashAlgorithm::None] { + run_decode_midway_death_case(hash_algo, context).await; + } + }) + .await; + } + } + + /// Decode a healthy 2+2 multi-stripe object through the lockstep GET path + /// and return the raw bytes pulled from each shard stream. + async fn healthy_lockstep_shard_bytes() -> (usize, Vec) { + const DATA_SHARDS: usize = 2; + const PARITY_SHARDS: usize = 2; + const BLOCK_SIZE: usize = 64; + + // 200 bytes => 3 full stripes + 1 partial. + let total_data: Vec = (0..200u32).map(|i| i as u8).collect(); + let total_len = total_data.len(); + + let erasure = Erasure::new(DATA_SHARDS, PARITY_SHARDS, BLOCK_SIZE); + let total_shards = DATA_SHARDS + PARITY_SHARDS; + let shard_size = erasure.shard_size(); + let hash_algo = HashAlgorithm::HighwayHash256; + + let mut shard_writers: Vec>>> = (0..total_shards) + .map(|_| BitrotWriter::new(Cursor::new(Vec::new()), shard_size, hash_algo.clone())) + .collect(); + + let mut offset = 0; + while offset < total_len { + let end = (offset + BLOCK_SIZE).min(total_len); + let shards = erasure.encode_data(&total_data[offset..end]).unwrap(); + for (i, shard) in shards.iter().enumerate() { + shard_writers[i].write(shard).await.unwrap(); + } + offset = end; + } + + let shard_bufs: Vec> = shard_writers.into_iter().map(|w| w.into_inner().into_inner()).collect(); + + let counters: Vec> = (0..total_shards).map(|_| Arc::new(AtomicUsize::new(0))).collect(); + let readers: Vec>> = shard_bufs .iter() .enumerate() .map(|(i, buf)| { - let bytes = if i == 0 { - buf[..first_block_len].to_vec() - } else { - buf.clone() - }; - Some(BitrotReader::new(Cursor::new(bytes), shard_size, hash_algo.clone(), false)) + Some(BitrotReader::new( + CountingShardReader { + inner: Cursor::new(buf.clone()), + bytes_read: Arc::clone(&counters[i]), + }, + shard_size, + hash_algo.clone(), + false, + )) }) .collect(); + // `Erasure::decode` is the reconstruction-verifying (lockstep) GET path. let mut output = Vec::new(); let (written, err) = erasure.decode(&mut output, readers, 0, total_len, total_len).await; - assert!(err.is_none(), "mid-object data-shard failure must still reconstruct: {err:?}"); - assert_eq!(written, total_len, "short write after mid-object shard failure"); - assert_eq!(output, total_data, "reconstructed bytes mismatch (stripe desync?)"); + assert!(err.is_none(), "healthy decode must succeed: {err:?}"); + assert_eq!(written, total_len); + assert_eq!(output, total_data); + + (DATA_SHARDS, counters.iter().map(|counter| counter.load(Ordering::SeqCst)).collect()) + } + + /// Merge gate for backlog#923: with the gate on, the healthy lockstep GET + /// path must read only the data shards — parity readers stay unopened and + /// contribute zero read bytes. This is the call-count proof of the 2x read + /// amplification fix (2+2 layout reads 2 shards per stripe, not 4). + #[tokio::test] + #[serial_test::serial] + async fn test_lockstep_healthy_get_reads_only_data_shards() { + temp_env::async_with_vars([(ENV_RUSTFS_GET_LOCKSTEP_DATA_SHARDS_ONLY_ENABLE, Some("true"))], async { + let (data_shards, shard_bytes) = healthy_lockstep_shard_bytes().await; + for (i, bytes) in shard_bytes.iter().enumerate() { + if i < data_shards { + assert!(*bytes > 0, "data shard {i} must be read on the healthy path"); + } else { + assert_eq!( + *bytes, 0, + "healthy GET must not read parity shard {i} (2x read amplification, backlog#923)" + ); + } + } + }) + .await; + } + + /// Compatibility lock: with the gate off (default), the lockstep path + /// keeps the pre-backlog#923 behavior and reads every live shard — + /// including parity — on every stripe. + #[tokio::test] + #[serial_test::serial] + async fn test_lockstep_default_reads_all_shards_per_stripe() { + temp_env::async_with_vars([(ENV_RUSTFS_GET_LOCKSTEP_DATA_SHARDS_ONLY_ENABLE, None::<&str>)], async { + let (_data_shards, shard_bytes) = healthy_lockstep_shard_bytes().await; + for (i, bytes) in shard_bytes.iter().enumerate() { + assert!(*bytes > 0, "default lockstep behavior must read shard {i} on every stripe"); + } + }) + .await; + } + + /// Merge gate for backlog#923: reconstruction verification must stay + /// active when parity is engaged mid-object. When a data shard dies at + /// stripe k, the lockstep path engages one parity reader beyond the decode + /// quorum (`erasure.rs` only verifies when `available > data_shards`), so + /// a parity shard whose content is inconsistent with the surviving data — + /// while still passing its own bitrot hash — must be detected and fail the + /// read instead of silently corrupting the reconstructed output. Run with + /// the gate on and off: the rejection must hold in both modes. + #[tokio::test] + #[serial_test::serial] + async fn test_erasure_decode_rejects_inconsistent_parity_engaged_midstream() { + for enabled in [None, Some("true")] { + temp_env::async_with_vars([(ENV_RUSTFS_GET_LOCKSTEP_DATA_SHARDS_ONLY_ENABLE, enabled)], async { + run_inconsistent_parity_midstream_case(if enabled.is_some() { "data-shards-only" } else { "default" }).await; + }) + .await; + } + } + + async fn run_inconsistent_parity_midstream_case(context: &str) { + const DATA_SHARDS: usize = 2; + const PARITY_SHARDS: usize = 2; + const BLOCK_SIZE: usize = 64; + + let total_data: Vec = (0..200u32).map(|i| i as u8).collect(); + let total_len = total_data.len(); + + let erasure = Erasure::new(DATA_SHARDS, PARITY_SHARDS, BLOCK_SIZE); + let total_shards = DATA_SHARDS + PARITY_SHARDS; + let shard_size = erasure.shard_size(); + let hash_algo = HashAlgorithm::HighwayHash256; + let hash_size = hash_algo.size(); + + let mut shard_writers: Vec>>> = (0..total_shards) + .map(|_| BitrotWriter::new(Cursor::new(Vec::new()), shard_size, hash_algo.clone())) + .collect(); + + let mut offset = 0; + let mut stripe = 0usize; + while offset < total_len { + let end = (offset + BLOCK_SIZE).min(total_len); + let shards = erasure.encode_data(&total_data[offset..end]).unwrap(); + for (i, shard) in shards.iter().enumerate() { + // Corrupt the first parity shard's payload for every stripe + // after the first, *before* it is bitrot-hashed: the shard + // passes its own hash check but is inconsistent with the + // erasure-coded set. + if i == DATA_SHARDS && stripe >= 1 { + let mut corrupted = shard.to_vec(); + corrupted[0] ^= 0x80; + shard_writers[i].write(&corrupted).await.unwrap(); + } else { + shard_writers[i].write(shard).await.unwrap(); + } + } + offset = end; + stripe += 1; + } + + let shard_bufs: Vec> = shard_writers.into_iter().map(|w| w.into_inner().into_inner()).collect(); + + // Data shard 0 dies after stripe 0, forcing parity engagement at stripe 1. + let first_block_len = (hash_size + shard_size).min(shard_bufs[0].len()); + let (readers, handles) = + readers_with_deferred_parity(&shard_bufs, DATA_SHARDS, shard_size, &hash_algo, &[(0, first_block_len)]); + + let mut output = Vec::new(); + let (_written, err) = erasure + .decode_with_stripe_handles(&mut output, readers, 0, total_len, total_len, None, handles) + .await; + + let err = err.unwrap_or_else(|| panic!("{context}: inconsistent parity engaged mid-stream must fail the read")); + assert_eq!(err.kind(), ErrorKind::InvalidData, "{context}"); + assert!(err.to_string().contains("inconsistent read source shards"), "{context}: {err}"); } #[cfg(feature = "rio-v2")] diff --git a/crates/ecstore/src/io_support/bitrot.rs b/crates/ecstore/src/io_support/bitrot.rs index b641dc204..bd1981750 100644 --- a/crates/ecstore/src/io_support/bitrot.rs +++ b/crates/ecstore/src/io_support/bitrot.rs @@ -27,7 +27,7 @@ use rustfs_utils::HashAlgorithm; use std::future::Future; use std::io::{self, Cursor}; use std::pin::Pin; -use std::sync::Mutex; +use std::sync::{Arc, Mutex}; use std::task::{Context, Poll}; use std::time::Instant; use tokio::io::{AsyncRead, ReadBuf}; @@ -134,7 +134,7 @@ impl AsyncRead for FirstReadMetricsReader { } struct DeferredObjectReader { - state: Mutex, + state: Arc>, } enum DeferredObjectReaderState { @@ -147,7 +147,64 @@ enum DeferredObjectReaderState { impl DeferredObjectReader { fn new(source: BitrotReaderSource) -> Self { Self { - state: Mutex::new(DeferredObjectReaderState::Pending(Some(source))), + state: Arc::new(Mutex::new(DeferredObjectReaderState::Pending(Some(source)))), + } + } + + fn stripe_handle(&self, stripe_stride: usize) -> DeferredReaderStripeHandle { + DeferredReaderStripeHandle { + state: Arc::clone(&self.state), + stripe_stride, + } + } +} + +/// Handle to a still-unopened [`DeferredObjectReader`] that can advance the +/// pending open offset by whole bitrot blocks (stripes) before the first read. +/// +/// The GET lockstep decode reads only the data shards while the object is +/// healthy and keeps every parity slot as an unopened deferred reader. When a +/// data shard dies at stripe `k`, the decoder uses this handle to shift the +/// parity reader's pending open offset by `k` encoded blocks — the same +/// `bitrot_encoded_range` geometry used when the reader was created — so its +/// first read returns stripe `k` and the lockstep alignment invariant holds +/// (backlog#923; alignment rule from backlog#832). +/// +/// `stripe_stride` is `shard_size + checksum_algo.size()` per full stripe. +/// For `hash_size == 0` (e.g. `HashAlgorithm::None`) this degrades to the +/// identity mapping `k * shard_size`, matching `bitrot_encoded_range`. +#[derive(Clone)] +pub(crate) struct DeferredReaderStripeHandle { + state: Arc>, + stripe_stride: usize, +} + +impl DeferredReaderStripeHandle { + /// Advance the pending source by `stripes` full stripes. + /// + /// Returns `false` when the reader has already been opened (or failed): + /// its stream position is then unknown to the caller and it must not be + /// engaged mid-object. + pub(crate) fn advance_stripes(&self, stripes: usize) -> bool { + if stripes == 0 { + return true; + } + let Ok(mut state) = self.state.lock() else { + return false; + }; + match &mut *state { + DeferredObjectReaderState::Pending(Some(source)) => { + let Some(delta) = self.stripe_stride.checked_mul(stripes) else { + return false; + }; + let Some(offset) = source.offset.checked_add(delta) else { + return false; + }; + source.offset = offset; + source.length = source.length.saturating_sub(delta); + true + } + _ => false, } } } @@ -471,6 +528,38 @@ pub fn create_deferred_bitrot_reader( skip_verify: bool, use_mmap_read: bool, ) -> BitrotReader> { + create_deferred_bitrot_reader_with_stripe_handle( + inline_data, + disk, + bucket, + path, + offset, + length, + shard_size, + checksum_algo, + skip_verify, + use_mmap_read, + ) + .0 +} + +/// Like [`create_deferred_bitrot_reader`], but also returns a +/// [`DeferredReaderStripeHandle`] that can realign the still-unopened reader +/// to a later stripe before its first read (backlog#923). +#[allow(clippy::too_many_arguments)] +pub(crate) fn create_deferred_bitrot_reader_with_stripe_handle( + inline_data: Option, + disk: Option, + bucket: &str, + path: &str, + offset: usize, + length: usize, + shard_size: usize, + checksum_algo: HashAlgorithm, + skip_verify: bool, + use_mmap_read: bool, +) -> (BitrotReader>, DeferredReaderStripeHandle) { + let stripe_stride = shard_size + checksum_algo.size(); let (offset, length) = bitrot_encoded_range(offset, length, shard_size, checksum_algo.clone()); let source = BitrotReaderSource { inline_data, @@ -483,7 +572,15 @@ pub fn create_deferred_bitrot_reader( stage_metrics: None, }; - BitrotReader::new(Box::new(DeferredObjectReader::new(source)), shard_size, checksum_algo, skip_verify) + let deferred = DeferredObjectReader::new(source); + let handle = deferred.stripe_handle(stripe_stride); + let reader = BitrotReader::new( + Box::new(deferred) as Box, + shard_size, + checksum_algo, + skip_verify, + ); + (reader, handle) } /// Create a new BitrotWriterWrapper based on the provided parameters @@ -741,6 +838,139 @@ mod tests { assert_eq!(&out[..n], b"efgh"); } + /// Merge gate for backlog#923: engaging a parity shard at stripe `k` + /// converts `k` stripes into an encoded byte offset via the + /// `bitrot_encoded_range` geometry. Cover both on-disk formats: the + /// streaming checksum layout (32-byte hash per block) and the + /// no-per-block-hash layout (`hash_size == 0`), where the mapping must + /// degrade to the identity `k * shard_size`. + #[tokio::test] + async fn test_deferred_stripe_handle_aligns_reader_to_requested_stripe() { + let shard_size = 4; + let payload = b"aaaabbbbccccdddd"; // 4 full bitrot blocks + + for algo in [HashAlgorithm::HighwayHash256S, HashAlgorithm::None] { + let mut writer = + create_bitrot_writer(true, None, "test-volume", "test-path", payload.len() as i64, shard_size, algo.clone()) + .await + .expect("inline bitrot writer"); + for chunk in payload.chunks(shard_size) { + writer.write(chunk).await.expect("write chunk"); + } + let inline_data = writer.into_inline_data().expect("inline buffer"); + + for stripe in 0..payload.len() / shard_size { + let (mut reader, handle) = create_deferred_bitrot_reader_with_stripe_handle( + Some(inline_data.clone().into()), + None, + "test-bucket", + "test-path", + 0, + payload.len(), + shard_size, + algo.clone(), + false, + false, + ); + + assert!( + handle.advance_stripes(stripe), + "pending deferred reader must accept a stripe advance (algo={algo:?}, stripe={stripe})" + ); + + let mut out = [0u8; 4]; + let n = reader.read(&mut out).await.expect("read shard block after stripe advance"); + assert_eq!(n, shard_size); + assert_eq!( + &out[..n], + &payload[stripe * shard_size..(stripe + 1) * shard_size], + "advanced reader must return exactly stripe {stripe} (algo={algo:?})" + ); + } + } + } + + /// Bitrot verification must stay active for a parity shard engaged + /// mid-object: after `advance_stripes(k)` the reader verifies stripe `k`'s + /// block against stripe `k`'s stored hash. + #[tokio::test] + async fn test_deferred_stripe_handle_preserves_bitrot_verification_after_advance() { + let shard_size = 4; + let algo = HashAlgorithm::HighwayHash256S; + let payload = b"aaaabbbbccccdddd"; + + let mut writer = + create_bitrot_writer(true, None, "test-volume", "test-path", payload.len() as i64, shard_size, algo.clone()) + .await + .expect("inline bitrot writer"); + for chunk in payload.chunks(shard_size) { + writer.write(chunk).await.expect("write chunk"); + } + let mut inline_data = writer.into_inline_data().expect("inline buffer"); + + // Corrupt one payload byte of block 2 (blocks are hash + data). + let block_len = algo.size() + shard_size; + inline_data[2 * block_len + algo.size()] ^= 0x01; + + let (mut reader, handle) = create_deferred_bitrot_reader_with_stripe_handle( + Some(inline_data.into()), + None, + "test-bucket", + "test-path", + 0, + payload.len(), + shard_size, + algo, + false, + false, + ); + assert!(handle.advance_stripes(2)); + + let mut out = [0u8; 4]; + let err = reader + .read(&mut out) + .await + .expect_err("bitrot mismatch at the advanced stripe must fail the read"); + assert_eq!(err.kind(), io::ErrorKind::InvalidData); + } + + /// A reader that has already been opened has an unknown stream position + /// from the handle's point of view; the advance must be refused so the + /// caller retires the reader instead of engaging it out of alignment. + #[tokio::test] + async fn test_deferred_stripe_handle_rejects_advance_after_open() { + let shard_size = 4; + let algo = HashAlgorithm::HighwayHash256S; + let payload = b"aaaabbbb"; + + let mut writer = + create_bitrot_writer(true, None, "test-volume", "test-path", payload.len() as i64, shard_size, algo.clone()) + .await + .expect("inline bitrot writer"); + for chunk in payload.chunks(shard_size) { + writer.write(chunk).await.expect("write chunk"); + } + let inline_data = writer.into_inline_data().expect("inline buffer"); + + let (mut reader, handle) = create_deferred_bitrot_reader_with_stripe_handle( + Some(inline_data.into()), + None, + "test-bucket", + "test-path", + 0, + payload.len(), + shard_size, + algo, + false, + false, + ); + + let mut out = [0u8; 4]; + reader.read(&mut out).await.expect("first read opens the deferred source"); + + assert!(!handle.advance_stripes(1), "an opened deferred reader must reject stripe advances"); + } + #[tokio::test] async fn test_create_bitrot_reader_without_data_or_disk() { let shard_size = 16; diff --git a/crates/ecstore/src/set_disk/core/io_primitives.rs b/crates/ecstore/src/set_disk/core/io_primitives.rs index b87b6473a..f9de75a9a 100644 --- a/crates/ecstore/src/set_disk/core/io_primitives.rs +++ b/crates/ecstore/src/set_disk/core/io_primitives.rs @@ -48,7 +48,8 @@ use crate::diagnostics::get::{ }; use crate::erasure::coding::BitrotReader; use crate::io_support::bitrot::{ - BitrotReaderStageMetrics, create_bitrot_reader_with_stage_metrics, create_deferred_bitrot_reader, object_mmap_read_enabled, + BitrotReaderStageMetrics, DeferredReaderStripeHandle, create_bitrot_reader_with_stage_metrics, + create_deferred_bitrot_reader_with_stripe_handle, object_mmap_read_enabled, }; use crate::set_disk::shard_source::ShardReadCost; use futures::stream::{FuturesUnordered, StreamExt}; @@ -911,6 +912,10 @@ pub(in crate::set_disk) const DIRECT_MEMORY_BITROT_READER_STAGE_METRICS: BitrotR pub(in crate::set_disk) struct BitrotReaderSetup { pub(in crate::set_disk) readers: Vec>, + /// Per-slot stripe handles for readers that are still unopened deferred + /// readers. The lockstep GET decode uses them to open a parity shard + /// aligned to the stripe where a data shard failed (backlog#923). + pub(in crate::set_disk) deferred_stripe_handles: Vec>, pub(in crate::set_disk) errors: Vec>, pub(in crate::set_disk) scheduled: Vec, pub(in crate::set_disk) attempted: Vec, @@ -985,6 +990,7 @@ impl BitrotReaderSetup { pub(in crate::set_disk) fn new(shards: usize) -> Self { Self { readers: (0..shards).map(|_| None).collect(), + deferred_stripe_handles: (0..shards).map(|_| None).collect(), errors: vec![Some(DiskError::DiskNotFound); shards], scheduled: vec![false; shards], attempted: vec![false; shards], @@ -1110,8 +1116,14 @@ impl BitrotReaderSetup { } } - pub(in crate::set_disk) fn retain_deferred_reader(&mut self, idx: usize, reader: ObjectBitrotReader) { + pub(in crate::set_disk) fn retain_deferred_reader( + &mut self, + idx: usize, + reader: ObjectBitrotReader, + stripe_handle: DeferredReaderStripeHandle, + ) { self.readers[idx] = Some(reader); + self.deferred_stripe_handles[idx] = Some(stripe_handle); self.errors[idx] = None; self.deferred_count = self.deferred_count.saturating_add(1); } @@ -1204,21 +1216,58 @@ pub(in crate::set_disk) fn fill_deferred_bitrot_readers( let disk = disks[idx].clone(); let data_dir = files[idx].data_dir.unwrap_or_default(); let path = format!("{object}/{data_dir}/part.{part_number}"); - setup.retain_deferred_reader( - idx, - create_deferred_bitrot_reader( - inline_data, - disk, - bucket, - &path, - read_offset, - read_length, - shard_size, - checksum_algo.clone(), - skip_verify_bitrot, - use_mmap_read, - ), + let (reader, stripe_handle) = create_deferred_bitrot_reader_with_stripe_handle( + inline_data, + disk, + bucket, + &path, + read_offset, + read_length, + shard_size, + checksum_algo.clone(), + skip_verify_bitrot, + use_mmap_read, ); + setup.retain_deferred_reader(idx, reader, stripe_handle); + } + + // With the data-shards-only lockstep gate on (backlog#923), the GET decode + // reads only the data shards while the object is healthy; a parity reader + // is engaged on demand and must therefore stay unopened so its start + // offset can be advanced to the failing stripe. A parity reader that was + // opened eagerly during setup is pinned at the stripe-0 stream position + // and could never be engaged mid-object, so swap it for an unopened + // deferred reader carrying a stripe handle. The eager open already proved + // the shard is reachable; no shard bytes were read from it, and the + // ready/error bookkeeping that quorum decisions rely on is left untouched. + // Gate off (default): keep the eagerly opened parity readers exactly as + // before — the lockstep path reads them on every stripe. + if !crate::erasure::coding::decode::get_lockstep_data_shards_only_enabled() { + return; + } + for idx in data_shards..disks.len() { + if setup.readers[idx].is_none() || setup.deferred_stripe_handles[idx].is_some() { + continue; + } + + let inline_data = files[idx].data.clone(); + let disk = disks[idx].clone(); + let data_dir = files[idx].data_dir.unwrap_or_default(); + let path = format!("{object}/{data_dir}/part.{part_number}"); + let (reader, stripe_handle) = create_deferred_bitrot_reader_with_stripe_handle( + inline_data, + disk, + bucket, + &path, + read_offset, + read_length, + shard_size, + checksum_algo.clone(), + skip_verify_bitrot, + use_mmap_read, + ); + setup.readers[idx] = Some(reader); + setup.deferred_stripe_handles[idx] = Some(stripe_handle); } } diff --git a/crates/ecstore/src/set_disk/read.rs b/crates/ecstore/src/set_disk/read.rs index af0a58538..0627218c2 100644 --- a/crates/ecstore/src/set_disk/read.rs +++ b/crates/ecstore/src/set_disk/read.rs @@ -912,13 +912,18 @@ impl SetDisks { let decode_stage_start = Instant::now(); let unattempted_data_shards = !reader_setup.data_shards_attempted(erasure.data_shards); let readers = reader_setup.readers; - let (written, err) = if let Some(read_costs) = read_costs { - erasure - .decode_with_read_costs(writer, readers, part_offset, part_length, part_size, read_costs) - .await - } else { - erasure.decode(writer, readers, part_offset, part_length, part_size).await - }; + let deferred_stripe_handles = reader_setup.deferred_stripe_handles; + let (written, err) = erasure + .decode_with_stripe_handles( + writer, + readers, + part_offset, + part_length, + part_size, + read_costs, + deferred_stripe_handles, + ) + .await; let decode_elapsed = decode_stage_start.elapsed(); rustfs_io_metrics::record_get_object_decode_duration(decode_elapsed.as_secs_f64()); rustfs_io_metrics::record_get_object_stage_duration_by_size( @@ -1271,6 +1276,7 @@ impl SetDisks { } let readers = reader_setup.readers; + let deferred_stripe_handles = reader_setup.deferred_stripe_handles; let source = if let Some(read_costs) = read_costs { coding::decode::ParallelReader::new_with_metrics_path_read_costs_and_reconstruction_verification( readers, @@ -1288,7 +1294,8 @@ impl SetDisks { part_size, Some(metrics_path), ) - }; + } + .with_deferred_parity_handles(deferred_stripe_handles); let engine = build_get_codec_streaming_decode_engine(erasure.clone())?; let reader = coding::decode_reader::ErasureDecodeReader::new_with_metrics_path(source, engine, part_length, metrics_path)?; @@ -3205,6 +3212,60 @@ mod tests { assert_eq!(&out[..n], [b"aaaa", b"bbbb", b"cccc", b"dddd"][fallback_index]); } + /// backlog#923: with the data-shards-only lockstep gate on, every retained + /// parity reader must be an unopened deferred reader carrying a stripe + /// handle, so the decode path can realign it to a mid-object stripe. With + /// the gate off (default), eagerly opened parity readers are kept exactly + /// as before and carry no handles. + #[tokio::test] + #[serial_test::serial] + async fn bitrot_reader_setup_gates_parity_stripe_handle_conversion() { + for enabled in [None, Some("true")] { + // A missing data shard forces the VerifyReconstruction quorum to 3, + // so both parity slots complete eagerly (attempted + ready) before + // the deferred fill runs. + let mut setup = temp_env::async_with_vars( + [("RUSTFS_GET_LOCKSTEP_DATA_SHARDS_ONLY_ENABLE", enabled)], + setup_inline_bitrot_readers_with_preference( + vec![None, Some(b"bbbb"), Some(b"cccc"), Some(b"dddd")], + 2, + 2, + BitrotReaderSetupMode::VerifyReconstruction, + false, + ), + ) + .await; + + assert_eq!(setup.available_shards(), 3); + for idx in 2..4 { + assert!(setup.attempted[idx] && setup.ready[idx], "parity slot {idx} should be eagerly ready"); + assert!(setup.readers[idx].is_some(), "parity slot {idx} must keep a reader (enabled={enabled:?})"); + assert_eq!( + setup.deferred_stripe_handles[idx].is_some(), + enabled.is_some(), + "parity slot {idx} stripe handle must match the gate (enabled={enabled:?})" + ); + } + + if enabled.is_some() { + // The converted parity reader is still unopened: its handle + // accepts a stripe advance, and reading it yields the shard + // bytes (block 0 here). + let handle = setup.deferred_stripe_handles[3].as_ref().expect("slot 3 handle"); + assert!(handle.advance_stripes(1), "unopened converted parity reader must accept a stripe advance"); + + let mut reader = setup.readers[2].take().expect("slot 2 reader"); + let mut out = [0u8; 4]; + let n = reader + .read(&mut out) + .await + .expect("converted parity reader should open on read"); + assert_eq!(n, 4); + assert_eq!(&out[..n], b"cccc"); + } + } + } + #[tokio::test] async fn bitrot_reader_setup_data_blocks_first_keeps_deferred_fallback_readers() { let mut setup = setup_inline_bitrot_readers_with_env(