From a5a73610b69de207af961ed29bd8ea0437a0c528 Mon Sep 17 00:00:00 2001 From: houseme Date: Fri, 24 Jul 2026 17:36:30 +0800 Subject: [PATCH] fix(ecstore): self-verify no-parity bitrot writes (#5179) Co-authored-by: heihutu --- crates/ecstore/src/erasure/coding/bitrot.rs | 29 +++ .../src/set_disk/ops/bitrot_self_verify.rs | 228 ++++++++++++++++++ crates/ecstore/src/set_disk/ops/mod.rs | 1 + crates/ecstore/src/set_disk/ops/multipart.rs | 47 +++- crates/ecstore/src/set_disk/ops/object.rs | 85 +++---- 5 files changed, 328 insertions(+), 62 deletions(-) create mode 100644 crates/ecstore/src/set_disk/ops/bitrot_self_verify.rs diff --git a/crates/ecstore/src/erasure/coding/bitrot.rs b/crates/ecstore/src/erasure/coding/bitrot.rs index 8b99b2df8..2c658e4e4 100644 --- a/crates/ecstore/src/erasure/coding/bitrot.rs +++ b/crates/ecstore/src/erasure/coding/bitrot.rs @@ -880,6 +880,35 @@ mod tests { assert!(err.to_string().contains("hash mismatch")); } + #[tokio::test] + async fn bitrot_verify_rejects_issue_5173_final_block_hash_mismatch() { + let logical_size = 8_250_370usize; + let shard_size = 1_048_576usize; + let algo = HashAlgorithm::HighwayHash256S; + let mut bitrot_writer = BitrotWriter::new(Cursor::new(Vec::new()), shard_size, algo.clone()); + let data = vec![0x5a; logical_size]; + for chunk in data.chunks(shard_size) { + bitrot_writer.write(chunk).await.expect("issue 5173 shard should encode"); + } + let mut written = bitrot_writer.into_inner().into_inner(); + + assert_eq!(written.len(), 8_250_626); + assert_eq!(written.len() - logical_size, 8 * algo.size()); + + let last = written.len() - 1; + written[last] ^= 0x01; + let err = bitrot_verify( + Cursor::new(written), + bitrot_shard_file_size(logical_size, shard_size, algo.clone()), + logical_size, + algo, + shard_size, + ) + .await + .expect_err("final block hash mismatch must be rejected"); + assert!(err.to_string().contains("hash mismatch")); + } + #[tokio::test] async fn write_all_vectored_retries_partial_hash_and_data_writes_and_rejects_zero_write() { let mut writer = LimitedVectoredWriter { diff --git a/crates/ecstore/src/set_disk/ops/bitrot_self_verify.rs b/crates/ecstore/src/set_disk/ops/bitrot_self_verify.rs new file mode 100644 index 000000000..694d78a7d --- /dev/null +++ b/crates/ecstore/src/set_disk/ops/bitrot_self_verify.rs @@ -0,0 +1,228 @@ +// Copyright 2024 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use super::super::*; + +/// Null out any disk whose shard writer failed (or was never created) so its +/// truncated/absent shard is not committed by the final rename, and return the +/// number of disks that still carry a fully-written shard. +pub(in crate::set_disk::ops) fn drop_failed_writer_disks(disks: &mut [Option], writers: &[Option]) -> usize { + let mut committed = 0usize; + for (slot, writer) in disks.iter_mut().zip(writers.iter()) { + if writer.is_none() { + *slot = None; + } else if slot.is_some() { + committed += 1; + } + } + committed +} + +#[derive(Clone, Copy)] +pub(in crate::set_disk::ops) struct BitrotSelfVerifyTarget<'a> { + pub(in crate::set_disk::ops) operation: &'static str, + pub(in crate::set_disk::ops) bucket: &'a str, + pub(in crate::set_disk::ops) object: &'a str, + pub(in crate::set_disk::ops) part_number: Option, + pub(in crate::set_disk::ops) volume: &'a str, + pub(in crate::set_disk::ops) path: &'a str, + pub(in crate::set_disk::ops) logical_shard_size: usize, + pub(in crate::set_disk::ops) shard_size: usize, + pub(in crate::set_disk::ops) write_quorum: usize, +} + +pub(in crate::set_disk::ops) async fn verify_written_bitrot_shards( + disks: &[Option], + inline_parts: Option<&[FileInfo]>, + target: BitrotSelfVerifyTarget<'_>, +) -> Result { + let algo = HashAlgorithm::HighwayHash256S; + let encoded_shard_size = coding::bitrot_shard_file_size(target.logical_shard_size, target.shard_size, algo.clone()); + + let results = if let Some(parts) = inline_parts { + let tasks = disks.iter().enumerate().filter_map(|(index, disk)| { + let disk = disk.as_ref()?.clone(); + let data = parts.get(index).and_then(|part| part.data.as_ref()).cloned(); + let algo = algo.clone(); + Some(async move { + let result = match data { + Some(data) => coding::bitrot_verify( + Cursor::new(data), + encoded_shard_size, + target.logical_shard_size, + algo, + target.shard_size, + ) + .await + .map_err(Error::from), + None => Err(Error::other(format!("{} inline shard {index} is missing bitrot data", target.operation))), + }; + (index, disk, result) + }) + }); + join_all(tasks).await + } else { + let tasks = disks.iter().enumerate().filter_map(|(index, disk)| { + let disk = disk.as_ref()?.clone(); + let algo = algo.clone(); + Some(async move { + let result = match disk.read_file(target.volume, target.path).await { + Ok(reader) => { + coding::bitrot_verify(reader, encoded_shard_size, target.logical_shard_size, algo, target.shard_size) + .await + .map_err(Error::from) + } + Err(err) => Err(Error::from(err)), + }; + (index, disk, result) + }) + }); + join_all(tasks).await + }; + + let mut verified_shards = 0usize; + for (index, disk, result) in results { + if let Err(err) = result { + warn!( + event = EVENT_SET_DISK_WRITE, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_SET_DISK, + operation = target.operation, + bucket = target.bucket, + object = target.object, + part_number = target.part_number, + shard_index = index, + disk = ?disk, + logical_shard_size = target.logical_shard_size, + encoded_shard_size, + error = ?err, + "Set disk no-parity bitrot self-verify failed" + ); + return Err(Error::other(format!( + "{} no-parity bitrot self-verify failed for shard {index}: {err}", + target.operation + ))); + } + + verified_shards += 1; + } + + if verified_shards < target.write_quorum { + return Err(Error::other(format!( + "{} no-parity bitrot self-verify quorum unavailable: {verified_shards} shard(s) verified, need {}", + target.operation, target.write_quorum + ))); + } + + Ok(verified_shards) +} + +#[cfg(test)] +mod tests { + use super::super::object::hermetic_set_disks_support::hermetic_set_disks_for_pool_with_default_parity; + use super::*; + use crate::disk::DiskAPI as _; + + async fn encode_streaming_shard(data: &[u8], shard_size: usize) -> Bytes { + let mut writer = coding::BitrotWriter::new(Cursor::new(Vec::new()), shard_size, HashAlgorithm::HighwayHash256S); + for chunk in data.chunks(shard_size) { + writer.write(chunk).await.expect("streaming shard should encode"); + } + Bytes::from(writer.into_inner().into_inner()) + } + + #[test] + fn excludes_failed_writers_and_counts_committed() { + let mut disks = vec![Some(0u8), Some(1), Some(2), Some(3)]; + let writers = vec![Some(()), Some(()), None, Some(())]; + assert_eq!(drop_failed_writer_disks(&mut disks, &writers), 3); + assert_eq!(disks, vec![Some(0), Some(1), None, Some(3)]); + } + + #[test] + fn offline_disk_stays_excluded_and_uncounted() { + let mut disks = vec![Some(0u8), None, Some(2)]; + let writers = vec![Some(()), None, Some(())]; + assert_eq!(drop_failed_writer_disks(&mut disks, &writers), 2); + assert_eq!(disks, vec![Some(0), None, Some(2)]); + } + + #[test] + fn all_writers_ok_keeps_every_disk() { + let mut disks = vec![Some(0u8), Some(1), Some(2)]; + let writers = vec![Some(()), Some(()), Some(())]; + assert_eq!(drop_failed_writer_disks(&mut disks, &writers), 3); + assert_eq!(disks, vec![Some(0), Some(1), Some(2)]); + } + + #[tokio::test] + async fn no_parity_self_verify_rejects_issue_5173_final_block_mismatch() { + let (_temp_dirs, disks, _set_disks) = hermetic_set_disks_for_pool_with_default_parity(1, 0, 0).await; + let disk = disks[0].clone(); + let shard_size = 1_048_576usize; + let logical_shard_size = 8_250_370usize; + let payload = vec![0x5a; logical_shard_size]; + let encoded = encode_streaming_shard(&payload, shard_size).await; + let path = "issue-5173/part.1"; + + assert_eq!(encoded.len(), 8_250_626); + disk.write_all(RUSTFS_META_TMP_BUCKET, path, encoded.clone()) + .await + .expect("healthy temp shard should be written"); + let verified = verify_written_bitrot_shards( + &[Some(disk.clone())], + None, + BitrotSelfVerifyTarget { + operation: "put_object", + bucket: "bucket", + object: "object", + part_number: None, + volume: RUSTFS_META_TMP_BUCKET, + path, + logical_shard_size, + shard_size, + write_quorum: 1, + }, + ) + .await + .expect("healthy temp shard should verify"); + assert_eq!(verified, 1); + + let mut corrupt = encoded.to_vec(); + let last = corrupt.len() - 1; + corrupt[last] ^= 0x01; + disk.write_all(RUSTFS_META_TMP_BUCKET, path, Bytes::from(corrupt)) + .await + .expect("corrupt temp shard should be staged"); + + let err = verify_written_bitrot_shards( + &[Some(disk)], + None, + BitrotSelfVerifyTarget { + operation: "put_object", + bucket: "bucket", + object: "object", + part_number: None, + volume: RUSTFS_META_TMP_BUCKET, + path, + logical_shard_size, + shard_size, + write_quorum: 1, + }, + ) + .await + .expect_err("corrupt no-parity temp shard must not be committed"); + assert!(err.to_string().contains("bitrot self-verify failed")); + } +} diff --git a/crates/ecstore/src/set_disk/ops/mod.rs b/crates/ecstore/src/set_disk/ops/mod.rs index 44ea75cb9..4321ef3ed 100644 --- a/crates/ecstore/src/set_disk/ops/mod.rs +++ b/crates/ecstore/src/set_disk/ops/mod.rs @@ -17,6 +17,7 @@ //! borrows shared state through [`super::ctx::SetDisksCtx`]; the storage-api //! contract impls stay `for SetDisks`, so contract bounds are unchanged. +pub(crate) mod bitrot_self_verify; pub(crate) mod bucket; pub(crate) mod heal; pub(crate) mod heal_walk; diff --git a/crates/ecstore/src/set_disk/ops/multipart.rs b/crates/ecstore/src/set_disk/ops/multipart.rs index 1d922e09f..922c2e532 100644 --- a/crates/ecstore/src/set_disk/ops/multipart.rs +++ b/crates/ecstore/src/set_disk/ops/multipart.rs @@ -21,6 +21,7 @@ //! unchanged; method bodies are moved verbatim and runtime behavior is the same. use super::super::*; +use super::bitrot_self_verify::{BitrotSelfVerifyTarget, drop_failed_writer_disks, verify_written_bitrot_shards}; use crate::crash_inject::{self, CrashPoint}; use std::future::Future; use std::time::Duration; @@ -435,15 +436,16 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks { // return Err(to_object_err(Error::ErasureWriteQuorum, vec![bucket, object])); // } - let shuffle_disks = Self::shuffle_disks(&disks, &fi.erasure.distribution); + let mut shuffle_disks = Self::shuffle_disks(&disks, &fi.erasure.distribution); let part_suffix = format!("part.{part_id}"); let tmp_part = format!("{}x{}", Uuid::new_v4(), OffsetDateTime::now_utc().unix_timestamp()); let tmp_part_path = Arc::new(format!("{tmp_part}/{part_suffix}")); let result: Result = async { - let erasure = coding::Erasure::try_new(fi.erasure.data_blocks, fi.erasure.parity_blocks, fi.erasure.block_size) - .map_err(Error::from)?; + let erasure = + Arc::new(coding::Erasure::try_new(fi.erasure.data_blocks, fi.erasure.parity_blocks, fi.erasure.block_size) + .map_err(Error::from)?); let writer_setup_stage_start = rustfs_io_metrics::put_stage_metrics_enabled().then(Instant::now); let mut writers = Vec::with_capacity(shuffle_disks.len()); @@ -527,16 +529,14 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks { let (reader, w_size) = match write_path { SmallWritePath::SingleBlockNonInline => { - Arc::new(erasure) + Arc::clone(&erasure) .encode_single_block_non_inline(stream, &mut writers, write_quorum) .await? } SmallWritePath::PipelineBatchedLarge => { - Arc::new(erasure).encode_batched(stream, &mut writers, write_quorum).await? - } - SmallWritePath::Inline | SmallWritePath::Pipeline => { - Arc::new(erasure).encode(stream, &mut writers, write_quorum).await? + Arc::clone(&erasure).encode_batched(stream, &mut writers, write_quorum).await? } + SmallWritePath::Inline | SmallWritePath::Pipeline => Arc::clone(&erasure).encode(stream, &mut writers, write_quorum).await?, }; if let Some(stage_start) = encode_stage_start { @@ -568,6 +568,13 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks { )))?; } + let committed_shards = drop_failed_writer_disks(&mut shuffle_disks, &writers); + if committed_shards < write_quorum { + Err(Error::other(format!( + "put_object_part write quorum unavailable after encode: {committed_shards} shard(s) committed, need {write_quorum}" + )))?; + } + let index_op = data .stream .try_get_index() @@ -607,6 +614,28 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks { drop(writers); // drop writers to close all files + if fi.erasure.parity_blocks == 0 { + let written_size = i64::try_from(w_size).map_err(|_| Error::other("put_object_part written size overflows i64"))?; + let logical_shard_size = usize::try_from(erasure.shard_file_size(written_size)) + .map_err(|_| Error::other("put_object_part shard size overflows usize"))?; + verify_written_bitrot_shards( + &shuffle_disks, + None, + BitrotSelfVerifyTarget { + operation: "put_object_part", + bucket, + object, + part_number: Some(part_id), + volume: RUSTFS_META_TMP_BUCKET, + path: &tmp_part_path, + logical_shard_size, + shard_size: erasure.shard_size(), + write_quorum, + }, + ) + .await?; + } + let part_path = format!("{}/{}/{}", upload_id_path, fi.data_dir.unwrap_or_default(), part_suffix); // Serialize only the commit (rename_part), not the whole upload. Each @@ -645,7 +674,7 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks { let _ = self .rename_part( - &disks, + &shuffle_disks, RUSTFS_META_TMP_BUCKET, &tmp_part_path, RUSTFS_META_MULTIPART_BUCKET, diff --git a/crates/ecstore/src/set_disk/ops/object.rs b/crates/ecstore/src/set_disk/ops/object.rs index c4442fedc..5136cde92 100644 --- a/crates/ecstore/src/set_disk/ops/object.rs +++ b/crates/ecstore/src/set_disk/ops/object.rs @@ -20,6 +20,7 @@ //! SetDisks core (io_primitives) via inherent calls. use super::super::*; +use super::bitrot_self_verify::{BitrotSelfVerifyTarget, drop_failed_writer_disks, verify_written_bitrot_shards}; use crate::bucket::lifecycle::{ tier_delete_journal::{persist_tier_delete_journal_entry, remove_tier_delete_journal_entry}, @@ -787,7 +788,7 @@ impl SetDisks { let tmp_object = format!("{}/{}/part.1", tmp_dir, fi.data_dir.unwrap()); let result: Result<(ObjectInfo, Option)> = async { - let erasure = erasure_from_file_info(&fi, false)?; + let erasure = Arc::new(erasure_from_file_info(&fi, false)?); let put_object_size = known_put_object_storage_size(data.size()); let is_inline_buffer = storage_class_config.should_inline(erasure.shard_file_size(put_object_size), opts.versioned); @@ -875,7 +876,7 @@ impl SetDisks { let encode_stage_start = Instant::now(); let (reader, w_size) = match write_path { - SmallWritePath::Inline => match Arc::new(erasure) + SmallWritePath::Inline => match Arc::clone(&erasure) .encode_inline_small(stream, &mut writers, write_quorum) .await { @@ -885,7 +886,7 @@ impl SetDisks { return Err(e.into()); } }, - SmallWritePath::SingleBlockNonInline => match Arc::new(erasure) + SmallWritePath::SingleBlockNonInline => match Arc::clone(&erasure) .encode_single_block_non_inline(stream, &mut writers, write_quorum) .await { @@ -896,7 +897,7 @@ impl SetDisks { } }, SmallWritePath::PipelineBatchedLarge => { - match Arc::new(erasure).encode_batched(stream, &mut writers, write_quorum).await { + match Arc::clone(&erasure).encode_batched(stream, &mut writers, write_quorum).await { Ok((r, w)) => (r, w), Err(e) => { error!("encode_batched err {:?}", e); @@ -904,7 +905,7 @@ impl SetDisks { } } } - SmallWritePath::Pipeline => match Arc::new(erasure).encode(stream, &mut writers, write_quorum).await { + SmallWritePath::Pipeline => match Arc::clone(&erasure).encode(stream, &mut writers, write_quorum).await { Ok((r, w)) => (r, w), Err(e) => { error!("encode err {:?}", e); @@ -1023,6 +1024,32 @@ impl SetDisks { drop(writers); // drop writers to close all files, this is to prevent FileAccessDenied errors when renaming data + if fi.erasure.parity_blocks == 0 { + let written_size = i64::try_from(w_size).map_err(|_| Error::other("put_object written size overflows i64"))?; + let logical_shard_size = usize::try_from(erasure.shard_file_size(written_size)) + .map_err(|_| Error::other("put_object shard size overflows usize"))?; + verify_written_bitrot_shards( + &shuffle_disks, + if is_inline_buffer { + Some(parts_metadatas.as_slice()) + } else { + None + }, + BitrotSelfVerifyTarget { + operation: "put_object", + bucket, + object, + part_number: None, + volume: RUSTFS_META_TMP_BUCKET, + path: &tmp_object, + logical_shard_size, + shard_size, + write_quorum, + }, + ) + .await?; + } + if !opts.no_lock && object_lock_guard.is_none() { object_lock_guard = Some(self.acquire_write_lock_diag("put_object_commit", bucket, object).await?); } @@ -3880,22 +3907,6 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks { } } -/// Null out any disk whose shard writer failed (or was never created) so its -/// truncated/absent shard is not committed by `rename_data`, and return the -/// number of disks that still carry a fully-written shard. Extracted for unit -/// testing the write-quorum accounting (backlog#852 / #799 B3). -fn drop_failed_writer_disks(disks: &mut [Option], writers: &[Option]) -> usize { - let mut committed = 0usize; - for (slot, writer) in disks.iter_mut().zip(writers.iter()) { - if writer.is_none() { - *slot = None; - } else if slot.is_some() { - committed += 1; - } - } - committed -} - #[cfg(test)] mod erasure_construction_tests { use super::*; @@ -3920,38 +3931,6 @@ mod erasure_construction_tests { } } -#[cfg(test)] -mod b3_write_quorum_tests { - use super::drop_failed_writer_disks; - - #[test] - fn excludes_failed_writers_and_counts_committed() { - // Writer 2 failed (short write / error -> nulled). Its disk must be - // dropped so the truncated shard is not renamed or counted. - let mut disks = vec![Some(0u8), Some(1), Some(2), Some(3)]; - let writers = vec![Some(()), Some(()), None, Some(())]; - assert_eq!(drop_failed_writer_disks(&mut disks, &writers), 3); - assert_eq!(disks, vec![Some(0), Some(1), None, Some(3)]); - } - - #[test] - fn offline_disk_stays_excluded_and_uncounted() { - // Disk 1 was offline at setup (disk None, writer None): stays None, not counted. - let mut disks = vec![Some(0u8), None, Some(2)]; - let writers = vec![Some(()), None, Some(())]; - assert_eq!(drop_failed_writer_disks(&mut disks, &writers), 2); - assert_eq!(disks, vec![Some(0), None, Some(2)]); - } - - #[test] - fn all_writers_ok_keeps_every_disk() { - let mut disks = vec![Some(0u8), Some(1), Some(2)]; - let writers = vec![Some(()), Some(()), Some(())]; - assert_eq!(drop_failed_writer_disks(&mut disks, &writers), 3); - assert_eq!(disks, vec![Some(0), Some(1), Some(2)]); - } -} - #[cfg(test)] pub(in crate::set_disk::ops) mod hermetic_set_disks_support { //! Shared hermetic `SetDisks` construction for the ops tests below: the