fix(ecstore): self-verify no-parity bitrot writes (#5179)

Co-authored-by: heihutu <heihutu@gmail.com>

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 {

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<D, W>(disks: &mut [Option<D>], writers: &[Option<W>]) -> 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<usize>,
+    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<DiskStore>],
+    inline_parts: Option<&[FileInfo]>,
+    target: BitrotSelfVerifyTarget<'_>,
+) -> Result<usize> {
+    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"));
+    }
+}

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;

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<PartInfo> = 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,

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<OldCurrentSize>)> = 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<D, W>(disks: &mut [Option<D>], writers: &[Option<W>]) -> 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