gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-07-26 16:28:15 +00:00
Code Issues Packages Projects Releases Wiki Activity

perf(ecstore): wire legacy decode stripe prefetch behind default-off gate (HP-9 step 2) (#4542)

Browse Source 
perf(ecstore): wire legacy decode stripe prefetch / bitrot-decode overlap (backlog#930 HP-9 step 2)

The legacy GET decode duplex loop (default GET path + every Range request)
was strictly serial: read a stripe, reconstruct it, emit it, and only then
begin reading the next stripe. Two switches introduced by PR#3972 to overlap
this work — RUSTFS_GET_DECODE_STRIPE_PREFETCH_COUNT and
RUSTFS_GET_BITROT_DECODE_OVERLAP_ENABLE — were config-only with zero call
sites since introduction.

Wire both into the loop as a depth-1 stripe prefetch: while the current
stripe is reconstructed and emitted, the next stripe's shard reads (including
bitrot verification) run concurrently, hiding read latency under the emit /
duplex-backpressure stage. ParallelReader::read is inherently serial (it takes
&mut self and advances a shared stripe cursor), so at most one read can be in
flight; a prefetch count above 1 therefore collapses to the same
single-stripe-ahead pipeline rather than reading several stripes ahead. Both
switches feed one gate, legacy_stripe_prefetch_enabled().

Default (count == 1, overlap == false) keeps the loop on the pre-existing
strictly-serial path, byte for byte: the prefetch pipeline is a separate branch
and the serial branch is unchanged. The reconstruct/emit body is factored into
a shared emit_decoded_stripe helper used by both branches so error attribution,
read-quorum handling, reconstruction verification and stage metrics cannot
drift between them.

Correctness guarantees preserved under prefetch:
- A speculatively prefetched read for stripe N+1 is only consumed when the loop
  reaches N+1; if stripe N stops the loop, the in-flight read is awaited and
  dropped, so its errors never surface against stripe N.
- Bitrot (HighwayHash) verification runs inside each stripe read and is not
  bypassed or reordered; a corrupt shard is still rejected and, when
  unrecoverable, the read fails without emitting garbage.
- Shard buffers are recycled only after the overlapping next read has claimed
  its own — one extra stripe of memory (double buffering) with buffer reuse
  preserved at a one-stripe lag.
- Per-stripe exact length advance (backlog#799 B2), lockstep reconstruction
  verification (backlog#832) and the hash_size == 0 pass-through are unchanged.

Adds regression tests exercising serial-default, count>1 and overlap configs:
full/range reads byte-exact, degraded (missing-shard) reconstruction, corrupt
shard rejected-but-recovered, unrecoverable corruption erroring with no output,
late-stripe failure attributed correctly with stripe 0 still emitted,
hash_size == 0 pass-through, and the gate defaulting off.

Co-authored-by: heihutu <heihutu@gmail.com>
This commit is contained in:
houseme
2026-07-09 02:35:11 +08:00
committed by GitHub
parent e3e5693e60
commit d232a46b4d
1 changed files with 615 additions and 85 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/ecstore/src/erasure/coding/decode.rs
+615 -85
View File
@@ -141,6 +141,38 @@ fn is_bitrot_decode_overlap_enabled() -> bool {
) )
} }
/// Whether the legacy decode duplex loop should prefetch the next stripe while
/// the current stripe is reconstructed and emitted (backlog#930 HP-9 step 2).
///
/// Two independent switches feed this decision:
/// * `RUSTFS_GET_DECODE_STRIPE_PREFETCH_COUNT` (> 1) — the operator asks for
/// stripe reads to run ahead of decode.
/// * `RUSTFS_GET_BITROT_DECODE_OVERLAP_ENABLE` — overlap the read+bitrot-verify
/// of the next stripe with the decode of the current one.
///
/// Both describe the same realisable mechanism. `ParallelReader::read` is
/// inherently serial: it takes `&mut self` and advances a shared stripe cursor,
/// so at most one stripe read can be in flight. The only overlap available is
/// issuing the *next* stripe read while the *current* stripe is reconstructed
/// and emitted (depth 1). A prefetch count above 1 therefore collapses to the
/// same single-stripe-ahead pipeline rather than reading several stripes ahead.
///
/// Default (`count == 1`, `overlap == false`) => disabled => the loop keeps its
/// pre-existing strictly-serial read → reconstruct → emit behaviour, byte for
/// byte.
fn legacy_stripe_prefetch_enabled() -> bool {
get_decode_stripe_prefetch_count() > 1 || is_bitrot_decode_overlap_enabled()
}
/// Outcome of reconstructing and emitting a single already-read stripe in the
/// legacy decode loop.
enum StripeFlow {
/// The stripe was emitted; the loop should continue with the next stripe.
Continue,
/// A terminal condition was recorded in `ret_err`; the loop must stop.
Stop,
}
#[derive(Default)] #[derive(Default)]
struct ShardReadCostCounts { struct ShardReadCostCounts {
local: usize, local: usize,
@@ -1305,6 +1337,23 @@ fn get_data_block_len(shards: &[Option<Vec<u8>>], data_blocks: usize) -> usize {
size size
} }
/// Read one stripe from the parallel reader, wrapped in the legacy-duplex
/// stripe-read stage timer. Factored out so the depth-1 prefetch loop and the
/// serial loop time reads identically. A free `async fn` (rather than a closure)
/// so the returned future's borrow of `reader` is correctly tied to the call.
async fn read_stripe_timed<R>(
reader: &mut ParallelReader<R>,
stage_metrics_enabled: bool,
) -> (Vec<Option<Vec<u8>>>, Vec<Option<Error>>)
where
R: AsyncRead + Unpin + Send + Sync,
{
let stripe_read_stage_start = get_stage_timer_if_enabled(stage_metrics_enabled);
let out = reader.read().await;
record_get_stage_duration_if_enabled(GET_OBJECT_PATH_LEGACY_DUPLEX, GET_STAGE_STRIPE_READ, stripe_read_stage_start);
out
}
/// Write data blocks from encoded blocks to target, supporting offset and length /// Write data blocks from encoded blocks to target, supporting offset and length
async fn write_data_blocks<W>( async fn write_data_blocks<W>(
writer: &mut W, writer: &mut W,
@@ -1489,6 +1538,108 @@ impl Erasure {
.await .await
} }
/// Reconstruct and emit one already-read stripe.
///
/// This is the shared per-stripe body used by both the serial legacy loop
/// and the depth-1 prefetch loop, so the two paths cannot drift: the quorum
/// check, reconstruction-verification, emit, error attribution, and stage
/// metrics are identical regardless of whether the *next* stripe read is
/// already running concurrently. It never recycles shard buffers — the
/// caller owns that, because under prefetch the buffers must not return to
/// the pool until the overlapping next-stripe read has claimed its own.
#[allow(clippy::too_many_arguments)]
async fn emit_decoded_stripe<W>(
&self,
writer: &mut W,
shards: &mut [Option<Vec<u8>>],
errs: &[Option<Error>],
block_offset: usize,
block_length: usize,
written: &mut usize,
ret_err: &mut Option<std::io::Error>,
stage_metrics_enabled: bool,
) -> StripeFlow
where
W: AsyncWrite + Send + Sync + Unpin,
{
if ret_err.is_none()
&& let (_, Some(err)) = reduce_errs(errs, &[])
&& (err == Error::FileNotFound || err == Error::FileCorrupt)
{
*ret_err = Some(err.into());
}
// Equivalent to `ParallelReader::can_decode`; inlined so this helper does
// not need to borrow the reader, leaving the reader free for the
// concurrent next-stripe read under prefetch.
let available_shards = shards.iter().filter(|shard| shard.is_some()).count();
if available_shards < self.data_shards {
let reason = GetObjectFailureReason::ReadQuorum;
error!(
data_shards = self.data_shards,
total_shards = self.data_shards + self.parity_shards,
available_shards,
block_offset,
block_length,
stage = GET_STAGE_STRIPE_READ,
reason = reason.as_str(),
errors = ?errs,
"Erasure decode could not gather enough shards"
);
record_get_object_pipeline_failure(GET_STAGE_STRIPE_READ, reason);
*ret_err = Some(Error::ErasureReadQuorum.into());
return StripeFlow::Stop;
}
// Decode the shards. If this stripe needed parity to reconstruct a
// missing data shard and an extra source shard was available, verify
// the reconstructed data against that source before streaming bytes.
let reconstruct_stage_start = get_stage_timer_if_enabled(stage_metrics_enabled);
if let Err(e) = self.decode_data_with_reconstruction_verification(shards) {
record_get_stage_duration_if_enabled(GET_OBJECT_PATH_LEGACY_DUPLEX, GET_STAGE_RECONSTRUCT, reconstruct_stage_start);
let reason = GetObjectFailureReason::DecodeError;
error!(
data_shards = self.data_shards,
total_shards = self.data_shards + self.parity_shards,
block_offset,
block_length,
stage = GET_STAGE_RECONSTRUCT,
reason = reason.as_str(),
error = ?e,
"Erasure shard reconstruction failed"
);
record_get_object_pipeline_failure(GET_STAGE_RECONSTRUCT, reason);
*ret_err = Some(e);
return StripeFlow::Stop;
}
record_get_stage_duration_if_enabled(GET_OBJECT_PATH_LEGACY_DUPLEX, GET_STAGE_RECONSTRUCT, reconstruct_stage_start);
let emit_stage_start = get_stage_timer_if_enabled(stage_metrics_enabled);
let n = match write_data_blocks(writer, shards, self.data_shards, block_offset, block_length).await {
Ok(n) => {
record_get_stage_duration_if_enabled(GET_OBJECT_PATH_LEGACY_DUPLEX, GET_STAGE_EMIT, emit_stage_start);
n
}
Err(e) => {
record_get_stage_duration_if_enabled(GET_OBJECT_PATH_LEGACY_DUPLEX, GET_STAGE_EMIT, emit_stage_start);
error!(
block_offset,
block_length,
bytes_written = *written,
stage = GET_STAGE_EMIT,
reason = classify_io_error(&e).as_str(),
error = ?e,
"Erasure decode failed to emit reconstructed data"
);
*ret_err = Some(e);
return StripeFlow::Stop;
}
};
*written += n;
StripeFlow::Continue
}
#[allow(clippy::too_many_arguments)] #[allow(clippy::too_many_arguments)]
async fn decode_inner<W, R>( async fn decode_inner<W, R>(
&self, &self,
@@ -1552,8 +1703,10 @@ impl Erasure {
let start = offset / self.block_size; let start = offset / self.block_size;
let end = end_offset.saturating_sub(1) / self.block_size; let end = end_offset.saturating_sub(1) / self.block_size;
for i in start..=end { // Emit geometry (block offset, block length) for block `i`. Pure function
let (block_offset, block_length) = if start == end { // of `i`; identical to the original inline computation.
let block_geometry = |i: usize| -> (usize, usize) {
if start == end {
(offset % self.block_size, length) (offset % self.block_size, length)
} else if i == start { } else if i == start {
(offset % self.block_size, self.block_size - (offset % self.block_size)) (offset % self.block_size, self.block_size - (offset % self.block_size))
@@ -1562,97 +1715,138 @@ impl Erasure {
(0, if end_remainder == 0 { self.block_size } else { end_remainder }) (0, if end_remainder == 0 { self.block_size } else { end_remainder })
} else { } else {
(0, self.block_size) (0, self.block_size)
};
if block_length == 0 {
// error!("erasure decode decode block_length == 0");
break;
} }
};
if legacy_stripe_prefetch_enabled() {
// Depth-1 stripe prefetch (backlog#930 HP-9 step 2): while the current
// stripe is reconstructed and emitted, the next stripe's shard reads
// (including bitrot verification) run concurrently, hiding read
// latency under the emit / duplex-backpressure stage.
//
// Correctness notes:
// * The reconstruct/emit body is the *same* `emit_decoded_stripe`
// used by the serial path, so error attribution, quorum handling,
// reconstruction verification and metrics are unchanged.
// * A speculatively prefetched read for stripe N+1 is only ever
// consumed when the loop actually reaches stripe N+1. If stripe N
// stops the loop, the in-flight read is awaited (join) and then
// dropped, so its errors never surface against stripe N.
// * Shard buffers are recycled only after the overlapping next read
// has already claimed its own buffers — costing one extra stripe of
// memory (double buffering) while preserving buffer reuse with a
// one-stripe lag.
let stage_metrics_enabled = rustfs_io_metrics::get_stage_metrics_enabled(); let stage_metrics_enabled = rustfs_io_metrics::get_stage_metrics_enabled();
let stripe_read_stage_start = get_stage_timer_if_enabled(stage_metrics_enabled);
let (mut shards, errs) = reader.read().await;
record_get_stage_duration_if_enabled(GET_OBJECT_PATH_LEGACY_DUPLEX, GET_STAGE_STRIPE_READ, stripe_read_stage_start);
if ret_err.is_none() // Block geometries to emit, stopping at the first zero-length block
&& let (_, Some(err)) = reduce_errs(&errs, &[]) // exactly as the serial loop's `break` did.
&& (err == Error::FileNotFound || err == Error::FileCorrupt) let mut blocks = Vec::with_capacity(end - start + 1);
{ for i in start..=end {
ret_err = Some(err.into()); let geometry = block_geometry(i);
} if geometry.1 == 0 {
if !reader.can_decode(&shards) {
let reason = GetObjectFailureReason::ReadQuorum;
error!(
data_shards = self.data_shards,
total_shards = self.data_shards + self.parity_shards,
available_shards = shards.iter().filter(|shard| shard.is_some()).count(),
block_offset,
block_length,
stage = GET_STAGE_STRIPE_READ,
reason = reason.as_str(),
errors = ?errs,
"Erasure decode could not gather enough shards"
);
record_get_object_pipeline_failure(GET_STAGE_STRIPE_READ, reason);
ret_err = Some(Error::ErasureReadQuorum.into());
break;
}
// Decode the shards. If this stripe needed parity to reconstruct a
// missing data shard and an extra source shard was available, verify
// the reconstructed data against that source before streaming bytes.
let reconstruct_stage_start = get_stage_timer_if_enabled(stage_metrics_enabled);
if let Err(e) = self.decode_data_with_reconstruction_verification(&mut shards) {
record_get_stage_duration_if_enabled(
GET_OBJECT_PATH_LEGACY_DUPLEX,
GET_STAGE_RECONSTRUCT,
reconstruct_stage_start,
);
let reason = GetObjectFailureReason::DecodeError;
error!(
data_shards = self.data_shards,
total_shards = self.data_shards + self.parity_shards,
block_offset,
block_length,
stage = GET_STAGE_RECONSTRUCT,
reason = reason.as_str(),
error = ?e,
"Erasure shard reconstruction failed"
);
record_get_object_pipeline_failure(GET_STAGE_RECONSTRUCT, reason);
ret_err = Some(e);
break;
}
record_get_stage_duration_if_enabled(GET_OBJECT_PATH_LEGACY_DUPLEX, GET_STAGE_RECONSTRUCT, reconstruct_stage_start);
let emit_stage_start = get_stage_timer_if_enabled(stage_metrics_enabled);
let n = match write_data_blocks(writer, &shards, self.data_shards, block_offset, block_length).await {
Ok(n) => {
record_get_stage_duration_if_enabled(GET_OBJECT_PATH_LEGACY_DUPLEX, GET_STAGE_EMIT, emit_stage_start);
n
}
Err(e) => {
record_get_stage_duration_if_enabled(GET_OBJECT_PATH_LEGACY_DUPLEX, GET_STAGE_EMIT, emit_stage_start);
error!(
block_offset,
block_length,
bytes_written = written,
stage = GET_STAGE_EMIT,
reason = classify_io_error(&e).as_str(),
error = ?e,
"Erasure decode failed to emit reconstructed data"
);
ret_err = Some(e);
break; break;
} }
}; blocks.push(geometry);
}
written += n; if !blocks.is_empty() {
// Prime the first stripe read. `current` holds the stripe to
// process next; `take()` moves it out cleanly per iteration and
// the prefetch branch refills it with the overlapping read.
let mut current = Some(read_stripe_timed(&mut reader, stage_metrics_enabled).await);
// Hand active-reader buffers back so the next stripe can reuse them for idx in 0..blocks.len() {
// without retaining offline shard data that cannot be read again. let (block_offset, block_length) = blocks[idx];
reader.recycle_shards(&mut shards); let Some((mut shards, errs)) = current.take() else {
break;
};
if idx + 1 < blocks.len() {
// Overlap: read stripe idx+1 while reconstructing/emitting idx.
let read_fut = read_stripe_timed(&mut reader, stage_metrics_enabled);
let emit_fut = self.emit_decoded_stripe(
writer,
&mut shards,
&errs,
block_offset,
block_length,
&mut written,
&mut ret_err,
stage_metrics_enabled,
);
let (next, flow) = tokio::join!(read_fut, emit_fut);
match flow {
StripeFlow::Continue => {
reader.recycle_shards(&mut shards);
current = Some(next);
}
StripeFlow::Stop => {
// `next` is speculative; drop it without surfacing
// its errors against the stripe that stopped.
drop(next);
break;
}
}
} else {
// Final stripe: nothing left to prefetch.
match self
.emit_decoded_stripe(
writer,
&mut shards,
&errs,
block_offset,
block_length,
&mut written,
&mut ret_err,
stage_metrics_enabled,
)
.await
{
StripeFlow::Continue => reader.recycle_shards(&mut shards),
StripeFlow::Stop => break,
}
}
}
}
} else {
// Default strictly-serial path: read → reconstruct → emit, one stripe
// at a time. Byte-for-byte identical to the pre-HP-9 behaviour.
for i in start..=end {
let (block_offset, block_length) = block_geometry(i);
if block_length == 0 {
// error!("erasure decode decode block_length == 0");
break;
}
let stage_metrics_enabled = rustfs_io_metrics::get_stage_metrics_enabled();
let stripe_read_stage_start = get_stage_timer_if_enabled(stage_metrics_enabled);
let (mut shards, errs) = reader.read().await;
record_get_stage_duration_if_enabled(
GET_OBJECT_PATH_LEGACY_DUPLEX,
GET_STAGE_STRIPE_READ,
stripe_read_stage_start,
);
match self
.emit_decoded_stripe(
writer,
&mut shards,
&errs,
block_offset,
block_length,
&mut written,
&mut ret_err,
stage_metrics_enabled,
)
.await
{
// Hand active-reader buffers back so the next stripe can reuse
// them without retaining offline shard data.
StripeFlow::Continue => reader.recycle_shards(&mut shards),
StripeFlow::Stop => break,
}
}
} }
if ret_err.is_some() { if ret_err.is_some() {
@@ -2077,6 +2271,342 @@ mod tests {
} }
} }
// ---- backlog#930 HP-9 step 2: legacy decode-loop stripe prefetch ----
/// Encode `total_data` into per-shard bitrot streams for the given layout,
/// returning the raw shard buffers.
async fn encode_prefetch_object(erasure: &Erasure, total_data: &[u8], hash_algo: &HashAlgorithm) -> Vec<Vec<u8>> {
let total_shards = erasure.data_shards + erasure.parity_shards;
let shard_size = erasure.shard_size();
let block_size = erasure.block_size;
let mut shard_writers: Vec<BitrotWriter<Cursor<Vec<u8>>>> = (0..total_shards)
.map(|_| BitrotWriter::new(Cursor::new(Vec::new()), shard_size, hash_algo.clone()))
.collect();
let mut offset = 0;
while offset < total_data.len() {
let end = (offset + block_size).min(total_data.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;
}
shard_writers.into_iter().map(|w| w.into_inner().into_inner()).collect()
}
/// Build a reader set positioned at the block covering `offset`, optionally
/// dropping (`missing`) or byte-flipping (`corrupt`) individual shards. All
/// readers keep bitrot verification ON so corruption is actually detected.
fn build_prefetch_readers(
shard_bufs: &[Vec<u8>],
shard_size: usize,
hash_algo: &HashAlgorithm,
block_size: usize,
offset: usize,
missing: &[usize],
corrupt: &[usize],
) -> Vec<Option<BitrotReader<Cursor<Vec<u8>>>>> {
let hash_size = hash_algo.size();
let start_block = offset / block_size;
let cursor_pos = start_block * (shard_size + hash_size);
shard_bufs
.iter()
.enumerate()
.map(|(i, buf)| {
if missing.contains(&i) {
return None;
}
let mut bytes = buf.clone();
if corrupt.contains(&i) {
// Flip the first data byte of the starting block, past its
// per-block checksum, so bitrot verification must reject it.
let flip = cursor_pos + hash_size;
if flip < bytes.len() {
bytes[flip] ^= 0xFF;
}
}
let mut cursor = Cursor::new(bytes);
cursor.set_position(cursor_pos as u64);
Some(BitrotReader::new(cursor, shard_size, hash_algo.clone(), false))
})
.collect()
}
/// One prefetch env configuration: a label plus the env vars it sets.
type PrefetchEnvConfig = (&'static str, Vec<(&'static str, Option<&'static str>)>);
/// Prefetch on/off env configurations exercised by the HP-9 tests.
/// `serial-default` pins both switches off (the byte-identical serial path);
/// the others turn the depth-1 prefetch pipeline on through each switch and
/// through a count above 1 (which collapses to depth 1 by design).
fn prefetch_env_configs() -> Vec<PrefetchEnvConfig> {
vec![
(
"serial-default",
vec![
(ENV_RUSTFS_GET_DECODE_STRIPE_PREFETCH_COUNT, None),
(ENV_RUSTFS_GET_BITROT_DECODE_OVERLAP_ENABLE, None),
],
),
(
"prefetch-count-2",
vec![
(ENV_RUSTFS_GET_DECODE_STRIPE_PREFETCH_COUNT, Some("2")),
(ENV_RUSTFS_GET_BITROT_DECODE_OVERLAP_ENABLE, None),
],
),
(
"bitrot-overlap",
vec![
(ENV_RUSTFS_GET_DECODE_STRIPE_PREFETCH_COUNT, None),
(ENV_RUSTFS_GET_BITROT_DECODE_OVERLAP_ENABLE, Some("true")),
],
),
(
"prefetch-count-8",
vec![
(ENV_RUSTFS_GET_DECODE_STRIPE_PREFETCH_COUNT, Some("8")),
(ENV_RUSTFS_GET_BITROT_DECODE_OVERLAP_ENABLE, Some("true")),
],
),
]
}
/// Full and ranged healthy reads must be byte-exact under every prefetch
/// configuration, proving the depth-1 pipeline preserves range offset/length
/// precision, the short final stripe, and equivalence with the serial path.
#[tokio::test]
#[serial_test::serial]
async fn test_legacy_prefetch_full_and_range_reads_match_serial() {
const DATA_SHARDS: usize = 4;
const PARITY_SHARDS: usize = 2;
const BLOCK_SIZE: usize = 64;
let total_data: Vec<u8> = (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 shard_size = erasure.shard_size();
let hash_algo = HashAlgorithm::HighwayHash256;
let shard_bufs = encode_prefetch_object(&erasure, &total_data, &hash_algo).await;
let cases: &[(usize, usize, &str)] = &[
(0, total_len, "full read"),
(0, 50, "head partial block"),
(10, 30, "within first block"),
(60, 80, "crossing two block boundaries"),
(128, 50, "starting at block boundary"),
(130, 10, "small range deep in middle"),
(192, 8, "tail partial block"),
// Multi-stripe ranges ending exactly on a block boundary exercise the
// `end_remainder == 0 => block_size` geometry branch under the prefetch
// pipeline (the final covered stripe must emit a full block, not a
// short tail). See backlog#930 HP-9 step 2 adversarial review.
(0, 128, "two full stripes ending on block boundary"),
(64, 64, "single full interior stripe on boundaries"),
];
for (label, vars) in prefetch_env_configs() {
temp_env::async_with_vars(vars, async {
for &(off, len, desc) in cases {
let readers = build_prefetch_readers(&shard_bufs, shard_size, &hash_algo, BLOCK_SIZE, off, &[], &[]);
let mut output = Vec::new();
let (written, err) = erasure.decode(&mut output, readers, off, len, total_len).await;
assert!(err.is_none(), "{label}/{desc}: unexpected error {err:?}");
assert_eq!(written, len, "{label}/{desc}: written mismatch");
assert_eq!(output, total_data[off..off + len], "{label}/{desc}: bytes mismatch");
}
})
.await;
}
}
/// Degraded read: two data shards offline, so every stripe reconstructs from
/// the survivors. Must stay byte-exact under prefetch, exercising buffer
/// recycle and reconstruction while the next stripe read runs concurrently.
#[tokio::test]
#[serial_test::serial]
async fn test_legacy_prefetch_degraded_read_reconstructs() {
const DATA_SHARDS: usize = 4;
const PARITY_SHARDS: usize = 2;
const BLOCK_SIZE: usize = 64;
let total_data: Vec<u8> = (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 shard_size = erasure.shard_size();
let hash_algo = HashAlgorithm::HighwayHash256;
let shard_bufs = encode_prefetch_object(&erasure, &total_data, &hash_algo).await;
for (label, vars) in prefetch_env_configs() {
temp_env::async_with_vars(vars, async {
let readers = build_prefetch_readers(&shard_bufs, shard_size, &hash_algo, BLOCK_SIZE, 0, &[0, 2], &[]);
let mut output = Vec::new();
let (written, err) = erasure.decode(&mut output, readers, 0, total_len, total_len).await;
assert!(err.is_none(), "{label}: degraded read errored {err:?}");
assert_eq!(written, total_len, "{label}: degraded short write");
assert_eq!(output, total_data, "{label}: degraded bytes mismatch");
})
.await;
}
}
/// A single bit-flipped data shard must be rejected by HighwayHash on every
/// prefetch configuration, and the object must still reconstruct byte-exact
/// from the survivors — proving prefetch neither bypasses bitrot nor lets a
/// corrupt shard leak into the output.
#[tokio::test]
#[serial_test::serial]
async fn test_legacy_prefetch_rejects_corrupt_shard_but_recovers() {
const DATA_SHARDS: usize = 4;
const PARITY_SHARDS: usize = 2;
const BLOCK_SIZE: usize = 64;
let total_data: Vec<u8> = (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 shard_size = erasure.shard_size();
let hash_algo = HashAlgorithm::HighwayHash256;
let shard_bufs = encode_prefetch_object(&erasure, &total_data, &hash_algo).await;
for (label, vars) in prefetch_env_configs() {
temp_env::async_with_vars(vars, async {
let readers = build_prefetch_readers(&shard_bufs, shard_size, &hash_algo, BLOCK_SIZE, 0, &[], &[1]);
let mut output = Vec::new();
let (written, err) = erasure.decode(&mut output, readers, 0, total_len, total_len).await;
assert!(err.is_none(), "{label}: corrupt-shard read errored {err:?}");
assert_eq!(written, total_len, "{label}: corrupt-shard short write");
assert_eq!(output, total_data, "{label}: corrupt bytes leaked into output");
})
.await;
}
}
/// Unrecoverable stripe 0 (two shards offline + one corrupt leaves < quorum)
/// must fail the read on every prefetch configuration, and must not emit any
/// bytes reconstructed from absent/corrupt data — so a speculatively
/// prefetched read can never turn a failure into silent garbage output.
#[tokio::test]
#[serial_test::serial]
async fn test_legacy_prefetch_unrecoverable_corruption_errors() {
const DATA_SHARDS: usize = 4;
const PARITY_SHARDS: usize = 2;
const BLOCK_SIZE: usize = 64;
let total_data: Vec<u8> = (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 shard_size = erasure.shard_size();
let hash_algo = HashAlgorithm::HighwayHash256;
let shard_bufs = encode_prefetch_object(&erasure, &total_data, &hash_algo).await;
for (label, vars) in prefetch_env_configs() {
temp_env::async_with_vars(vars, async {
let readers = build_prefetch_readers(&shard_bufs, shard_size, &hash_algo, BLOCK_SIZE, 0, &[0, 1], &[2]);
let mut output = Vec::new();
let (_written, err) = erasure.decode(&mut output, readers, 0, total_len, total_len).await;
assert!(err.is_some(), "{label}: unrecoverable corruption must fail the read");
assert!(output.is_empty(), "{label}: emitted bytes despite an unrecoverable stripe 0");
})
.await;
}
}
/// The first stripe is fully recoverable but the parity readers are
/// truncated to one block, so a *later* stripe loses quorum. Under prefetch
/// that later read is issued speculatively while stripe 0 emits, yet its
/// failure must be attributed to the later stripe: stripe 0's bytes are
/// emitted and the read then fails, rather than the prefetch error poisoning
/// the already-good stripe.
#[tokio::test]
#[serial_test::serial]
async fn test_legacy_prefetch_attributes_late_stripe_failure() {
const DATA_SHARDS: usize = 4;
const PARITY_SHARDS: usize = 2;
const BLOCK_SIZE: usize = 64;
let total_data: Vec<u8> = (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 shard_size = erasure.shard_size();
let hash_algo = HashAlgorithm::HighwayHash256;
let hash_size = hash_algo.size();
let shard_bufs = encode_prefetch_object(&erasure, &total_data, &hash_algo).await;
let first_block_len = hash_size + shard_size;
for (label, vars) in prefetch_env_configs() {
temp_env::async_with_vars(vars, async {
let readers: Vec<Option<BitrotReader<Cursor<Vec<u8>>>>> = shard_bufs
.iter()
.enumerate()
.map(|(i, buf)| {
if i == 0 || i == 1 {
return None; // two data shards offline
}
let bytes = if i >= DATA_SHARDS {
buf[..first_block_len.min(buf.len())].to_vec() // parity: one block only
} else {
buf.clone()
};
Some(BitrotReader::new(Cursor::new(bytes), shard_size, hash_algo.clone(), false))
})
.collect();
let mut output = Vec::new();
let (written, err) = erasure.decode(&mut output, readers, 0, total_len, total_len).await;
assert!(err.is_some(), "{label}: truncated-parity object must fail on the later stripe");
assert_eq!(written, BLOCK_SIZE, "{label}: first stripe must emit before the later failure");
assert_eq!(output, total_data[..BLOCK_SIZE], "{label}: first stripe bytes wrong");
})
.await;
}
}
/// `hash_size == 0` (HashAlgorithm::None) direct pass-through must stay
/// byte-exact under prefetch for both a full and a mid-object range read.
#[tokio::test]
#[serial_test::serial]
async fn test_legacy_prefetch_no_hash_passthrough() {
const DATA_SHARDS: usize = 4;
const PARITY_SHARDS: usize = 2;
const BLOCK_SIZE: usize = 64;
let total_data: Vec<u8> = (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 shard_size = erasure.shard_size();
let hash_algo = HashAlgorithm::None;
let shard_bufs = encode_prefetch_object(&erasure, &total_data, &hash_algo).await;
for (label, vars) in prefetch_env_configs() {
temp_env::async_with_vars(vars, async {
for &(off, len) in &[(0usize, total_len), (70usize, 90usize)] {
let readers = build_prefetch_readers(&shard_bufs, shard_size, &hash_algo, BLOCK_SIZE, off, &[], &[]);
let mut output = Vec::new();
let (written, err) = erasure.decode(&mut output, readers, off, len, total_len).await;
assert!(err.is_none(), "{label}: no-hash read errored {err:?}");
assert_eq!(written, len, "{label}: no-hash written mismatch");
assert_eq!(output, total_data[off..off + len], "{label}: no-hash bytes mismatch");
}
})
.await;
}
}
/// The prefetch gate is off by default and turns on through either switch.
#[test]
#[serial_test::serial]
fn test_legacy_stripe_prefetch_gate_defaults_off() {
temp_env::with_vars(
[
(ENV_RUSTFS_GET_DECODE_STRIPE_PREFETCH_COUNT, None::<&str>),
(ENV_RUSTFS_GET_BITROT_DECODE_OVERLAP_ENABLE, None::<&str>),
],
|| assert!(!legacy_stripe_prefetch_enabled(), "prefetch must default off"),
);
temp_env::with_vars([(ENV_RUSTFS_GET_DECODE_STRIPE_PREFETCH_COUNT, Some("2"))], || {
assert!(legacy_stripe_prefetch_enabled(), "count > 1 must enable prefetch");
});
temp_env::with_vars([(ENV_RUSTFS_GET_BITROT_DECODE_OVERLAP_ENABLE, Some("true"))], || {
assert!(legacy_stripe_prefetch_enabled(), "overlap switch must enable prefetch");
});
temp_env::with_vars([(ENV_RUSTFS_GET_DECODE_STRIPE_PREFETCH_COUNT, Some("1"))], || {
assert!(!legacy_stripe_prefetch_enabled(), "count == 1 must stay serial");
});
}
/// Decode a healthy 2+2 multi-stripe object through the lockstep GET path /// Decode a healthy 2+2 multi-stripe object through the lockstep GET path
/// and return the raw bytes pulled from each shard stream. /// and return the raw bytes pulled from each shard stream.
async fn healthy_lockstep_shard_bytes() -> (usize, Vec<usize>) { async fn healthy_lockstep_shard_bytes() -> (usize, Vec<usize>) {