diff --git a/crates/ecstore/src/erasure/coding/decode.rs b/crates/ecstore/src/erasure/coding/decode.rs index aa34a7a45..403b5df33 100644 --- a/crates/ecstore/src/erasure/coding/decode.rs +++ b/crates/ecstore/src/erasure/coding/decode.rs @@ -2299,6 +2299,7 @@ impl Erasure { written: &mut usize, ret_err: &mut Option, stage_metrics_enabled: bool, + require_surplus_source: bool, ) -> StripeFlow where W: AsyncWrite + Send + Sync + Unpin, @@ -2336,7 +2337,12 @@ impl Erasure { // 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) { + let decode_result = if require_surplus_source { + self.decode_data_with_reconstruction_verification_for_lockstep(shards) + } else { + self.decode_data_with_reconstruction_verification(shards) + }; + if let Err(e) = decode_result { record_get_stage_duration_if_enabled(GET_OBJECT_PATH_LEGACY_DUPLEX, GET_STAGE_RECONSTRUCT, reconstruct_stage_start); let reason = GetObjectFailureReason::DecodeError; error!( @@ -2547,6 +2553,7 @@ impl Erasure { // `shards` are borrowed again below. In the `Stop` case that // drop is what cancels the still-in-flight prefetch read. let (flow, next): (Option, Option) = { + let require_surplus_source = reader.demand_bound_lockstep; let read_fut = read_stripe_timed(&mut reader, stage_metrics_enabled); let emit_fut = self.emit_decoded_stripe( writer, @@ -2557,6 +2564,7 @@ impl Erasure { &mut written, &mut ret_err, stage_metrics_enabled, + require_surplus_source, ); tokio::pin!(read_fut); tokio::pin!(emit_fut); @@ -2604,6 +2612,7 @@ impl Erasure { &mut written, &mut ret_err, stage_metrics_enabled, + reader.demand_bound_lockstep, ) .await { @@ -2643,6 +2652,7 @@ impl Erasure { &mut written, &mut ret_err, stage_metrics_enabled, + reader.demand_bound_lockstep, ) .await { @@ -5306,6 +5316,58 @@ mod tests { assert!(error.is_none(), "a failed disposable hedge must not fail a recovered stripe: {error:?}"); } + /// Rollout guard for backlog#1308: when a data shard and the first parity + /// hedge both fail, the gate-on path must not settle at decode quorum and + /// emit an unverified body. The second parity can restore decode quorum but + /// cannot provide the extra source required for reconstruction verification, + /// so the stripe must fail before exposing bytes. + #[tokio::test] + #[serial_test::serial] + async fn test_data_shards_only_gate_data_and_parity_failure_fails_before_output() { + const BLOCK_SIZE: usize = 64; + const DATA_SHARDS: usize = 2; + const PARITY_SHARDS: usize = 2; + + temp_env::async_with_vars([(ENV_RUSTFS_GET_LOCKSTEP_DATA_SHARDS_ONLY_ENABLE, Some("true"))], async { + let erasure = Erasure::new(DATA_SHARDS, PARITY_SHARDS, BLOCK_SIZE); + let payload = (0..BLOCK_SIZE).map(|value| value as u8).collect::>(); + let shards = erasure.encode_data(&payload).expect("test payload should encode"); + let shard_size = erasure.shard_size(); + + let readers = vec![ + Some(BitrotReader::new(TestShardReader::TimedOut, shard_size, HashAlgorithm::None, false)), + Some(BitrotReader::new( + TestShardReader::Ready(Cursor::new(shards[1].to_vec())), + shard_size, + HashAlgorithm::None, + false, + )), + Some(BitrotReader::new( + TestShardReader::TerminalFileNotFound, + shard_size, + HashAlgorithm::None, + false, + )), + Some(BitrotReader::new( + TestShardReader::Ready(Cursor::new(shards[3].to_vec())), + shard_size, + HashAlgorithm::None, + false, + )), + ]; + + let mut output = Vec::new(); + let (written, error) = erasure.decode(&mut output, readers, 0, payload.len(), payload.len()).await; + + assert_eq!(written, 0, "an unverified stripe must not report body bytes"); + assert!(output.is_empty(), "an unverified stripe must not expose a clean short body"); + let error = error.expect("data plus parity loss must fail closed"); + assert_eq!(error.kind(), ErrorKind::InvalidData); + assert!(error.to_string().contains("insufficient source shards")); + }) + .await; + } + /// Lockstep verification-quorum regression (backlog#1156). When a data shard is /// missing, the hedge must settle only at `data_shards + 1` (decode quorum plus /// a reconstruction-verification source), never at exactly `data_shards` — that diff --git a/crates/ecstore/src/erasure/coding/erasure.rs b/crates/ecstore/src/erasure/coding/erasure.rs index a9e4c8bc3..602b24789 100644 --- a/crates/ecstore/src/erasure/coding/erasure.rs +++ b/crates/ecstore/src/erasure/coding/erasure.rs @@ -933,8 +933,29 @@ impl Erasure { } pub(crate) fn decode_data_with_reconstruction_verification(&self, shards: &mut [Option>]) -> io::Result<()> { + self.decode_data_with_reconstruction_verification_policy(shards, false) + } + + pub(crate) fn decode_data_with_reconstruction_verification_for_lockstep( + &self, + shards: &mut [Option>], + ) -> io::Result<()> { + self.decode_data_with_reconstruction_verification_policy(shards, true) + } + + fn decode_data_with_reconstruction_verification_policy( + &self, + shards: &mut [Option>], + require_surplus_source: bool, + ) -> io::Result<()> { let missing_data_source = shards.iter().take(self.data_shards).any(|shard| shard.is_none()); let available_shards = shards.iter().filter(|shard| shard.is_some()).count(); + if require_surplus_source && missing_data_source && available_shards == self.data_shards { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "insufficient source shards to verify reconstructed data", + )); + } let source_parity = if missing_data_source && available_shards > self.data_shards { shards .iter() @@ -1868,6 +1889,31 @@ mod tests { assert_eq!(err.kind(), io::ErrorKind::InvalidData); } + #[test] + fn decode_data_with_verification_scopes_exact_quorum_to_lockstep() { + for uses_legacy in [false, true] { + let erasure = Erasure::new_with_options(3, 2, 128, uses_legacy); + let data = b"verified reads must not accept reconstruction without a surplus source"; + let encoded = erasure.encode_data(data).expect("encode should succeed"); + let mut exact_quorum = optional_shards(&encoded); + exact_quorum[0] = None; + exact_quorum[erasure.total_shard_count() - 1] = None; + + let mut default_shards = exact_quorum.clone(); + erasure + .decode_data_with_reconstruction_verification(&mut default_shards) + .expect("default decode must preserve exact-quorum reconstruction"); + assert_eq!(default_shards[0].as_deref(), Some(encoded[0].as_ref())); + + let err = erasure + .decode_data_with_reconstruction_verification_for_lockstep(&mut exact_quorum) + .expect_err("data-shards-only lockstep must reject an exact decode quorum"); + + assert_eq!(err.kind(), io::ErrorKind::InvalidData); + assert!(err.to_string().contains("insufficient source shards")); + } + } + #[test] fn verify_data_and_parity_rejects_missing_and_mismatched_shards() { let erasure = Erasure::new(4, 2, 128); diff --git a/crates/ecstore/src/set_disk/ops/heal.rs b/crates/ecstore/src/set_disk/ops/heal.rs index c59fc16e2..60e189fd5 100644 --- a/crates/ecstore/src/set_disk/ops/heal.rs +++ b/crates/ecstore/src/set_disk/ops/heal.rs @@ -2751,6 +2751,26 @@ mod heal_result_report_tests { } } + async fn remove_current_object_part(temp_dir: &TempDir, bucket: &str, object: &str) -> std::io::Result<()> { + let object_dir = temp_dir.path().join(bucket).join(object); + let mut entries = tokio::fs::read_dir(&object_dir).await?; + while let Some(entry) = entries.next_entry().await? { + if !entry.file_type().await?.is_dir() { + continue; + } + let part = entry.path().join("part.1"); + match tokio::fs::remove_file(&part).await { + Ok(()) => return Ok(()), + Err(err) if err.kind() == std::io::ErrorKind::NotFound => continue, + Err(err) => return Err(err), + } + } + Err(std::io::Error::new( + std::io::ErrorKind::NotFound, + format!("no current part.1 found under {}", object_dir.display()), + )) + } + #[test] fn heal_writer_error_summary_redacts_io_message() { let error = DiskError::Io(std::io::Error::new(std::io::ErrorKind::PermissionDenied, "/sensitive/storage/path")); @@ -2783,21 +2803,13 @@ mod heal_result_report_tests { .read_version("", &bucket, object, "", &ReadOptions::default()) .await .expect("source metadata should be readable"); - let data_dir = source.data_dir.expect("non-inline source should have a data directory"); let mut target_slots = [source.erasure.distribution[0] - 1, source.erasure.distribution[1] - 1]; target_slots.sort_unstable(); for index in [0, 1] { - tokio::fs::remove_file( - temp_dirs[index] - .path() - .join(&bucket) - .join(object) - .join(data_dir.to_string()) - .join("part.1"), - ) - .await - .expect("target shard should be removed before heal"); + remove_current_object_part(&temp_dirs[index], &bucket, object) + .await + .expect("target shard should be removed before heal"); } let failed_slots = &target_slots[..failed_target_count];