From 31f446bed1d06afab7bdd6fe84d5bf4564a1d794 Mon Sep 17 00:00:00 2001 From: Hauser Date: Fri, 18 Sep 2026 08:07:37 +0800 Subject: [PATCH] fix(ecstore): preserve vectored bitrot writes (#7981) * fix(ecstore): preserve vectored bitrot writes Co-Authored-By: heihutu Co-Authored-By: zhi22915 * fix(ecstore): route test trait through contracts Co-Authored-By: heihutu Co-Authored-By: zhi22915 --------- Co-authored-by: zhi22915 --- crates/ecstore/src/disk/local.rs | 67 ++++++++++++++++++- .../src/set_disk/ops/bitrot_self_verify.rs | 16 +++++ crates/ecstore/src/set_disk/ops/object.rs | 6 +- rustfs/src/storage/rpc/node_service.rs | 5 +- 4 files changed, 91 insertions(+), 3 deletions(-) diff --git a/crates/ecstore/src/disk/local.rs b/crates/ecstore/src/disk/local.rs index 12099907b..12fce4480 100644 --- a/crates/ecstore/src/disk/local.rs +++ b/crates/ecstore/src/disk/local.rs @@ -1899,6 +1899,61 @@ impl AsyncWrite for DirectWriter { } } + fn poll_write_vectored( + self: std::pin::Pin<&mut Self>, + cx: &mut std::task::Context<'_>, + bufs: &[std::io::IoSlice<'_>], + ) -> std::task::Poll> { + // Bitrot writes arrive as [hash, data]. Coalesce both slices into the + // aligned bounce buffer so the O_DIRECT path has the same byte-stream + // contract as the buffered writer instead of relying on the trait's + // first-slice fallback. + let this = self.get_mut(); + loop { + match &mut this.state { + DirectWriteState::Busy(_) => { + std::task::ready!(this.poll_drive_busy(cx))?; + } + DirectWriteState::Idle(inner_opt) => { + let inner = inner_opt.as_mut().expect("idle direct writer must hold inner state"); + let capacity = inner.buf.len; + let space = capacity - inner.filled; + let mut remaining = space; + let mut written = 0; + for src in bufs { + if remaining == 0 { + break; + } + let take = src.len().min(remaining); + let start = inner.filled + written; + inner.buf.as_mut_slice()[start..start + take].copy_from_slice(&src[..take]); + written += take; + remaining -= take; + } + + if written == 0 { + return std::task::Poll::Ready(Ok(0)); + } + inner.filled += written; + + if inner.filled == capacity { + let mut inner = inner_opt.take().expect("idle direct writer must hold inner state"); + let handle = tokio::task::spawn_blocking(move || { + let res = inner.flush_batch(); + (inner, res) + }); + this.state = DirectWriteState::Busy(handle); + } + return std::task::Poll::Ready(Ok(written)); + } + } + } + } + + fn is_write_vectored(&self) -> bool { + true + } + fn poll_flush(self: std::pin::Pin<&mut Self>, cx: &mut std::task::Context<'_>) -> std::task::Poll> { // Only drive an in-flight batch to completion. Sub-alignment staged // bytes cannot be flushed mid-stream (they would misalign the next @@ -22134,7 +22189,7 @@ mod test { #[cfg(target_os = "linux")] #[tokio::test] async fn direct_writer_state_machine_round_trips_over_plain_file() { - use std::io::Read; + use std::io::{IoSlice, Read}; use tempfile::tempdir; let dir = tempdir().expect("tempdir"); @@ -22153,6 +22208,16 @@ mod test { let mut writer = DirectWriter::from_std_file_for_test(file, align, capacity); let mut off = 0; + if content.len() >= 2 { + let split = content.len().min(300); + let first = split / 2; + let written = writer + .write_vectored(&[IoSlice::new(&content[..first]), IoSlice::new(&content[first..split])]) + .await + .expect("vectored write"); + assert_eq!(written, split, "vectored write must consume both hash/data-like slices"); + off = split; + } while off < content.len() { let end = (off + 300).min(content.len()); writer.write_all(&content[off..end]).await.expect("write_all"); diff --git a/crates/ecstore/src/set_disk/ops/bitrot_self_verify.rs b/crates/ecstore/src/set_disk/ops/bitrot_self_verify.rs index 52af4a3d0..5a4cd45c6 100644 --- a/crates/ecstore/src/set_disk/ops/bitrot_self_verify.rs +++ b/crates/ecstore/src/set_disk/ops/bitrot_self_verify.rs @@ -141,6 +141,7 @@ pub(in crate::set_disk::ops) async fn verify_written_bitrot_shards( mod tests { use super::super::object::hermetic_set_disks_support::hermetic_set_disks_for_pool_with_default_parity; use super::*; + use crate::storage_api_contracts::object::ObjectIO 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); @@ -295,4 +296,19 @@ mod tests { assert!(err.to_string().contains("trailing data")); } } + + #[tokio::test] + async fn no_parity_put_round_trip_large_stream() { + let (_temp_dirs, disks, set_disks) = hermetic_set_disks_for_pool_with_default_parity(1, 0, 0).await; + let bucket = "no-parity-put-round-trip"; + for disk in &disks { + disk.make_volume(bucket).await.expect("bucket volume should be created"); + } + let payload = vec![0x5a; 13 * 1024 * 1024 + 727_213]; + let mut reader = crate::set_disk::PutObjReader::from_vec(payload); + set_disks + .put_object(bucket, "object", &mut reader, &crate::set_disk::ObjectOptions::default()) + .await + .expect("single-disk no-parity PUT should pass self-verify"); + } } diff --git a/crates/ecstore/src/set_disk/ops/object.rs b/crates/ecstore/src/set_disk/ops/object.rs index ddf9dfe38..f881457c5 100644 --- a/crates/ecstore/src/set_disk/ops/object.rs +++ b/crates/ecstore/src/set_disk/ops/object.rs @@ -7571,7 +7571,11 @@ impl SetDisks { let fi = &transported; let disks = self.disk_inventory().await; let namespace_owner = (!is_meta_bucketname(bucket)).then(|| self.ctx.begin_namespace_commit()); - let write_quorum = disks.len() / 2 + 1; + // Quorum is a property of the configured set, not of the current + // online snapshot. A decommission/offline refresh may temporarily + // shorten the snapshot; deriving quorum from it would turn a + // quorum-minus-one delete into an apparent success. + let write_quorum = self.set_drive_count / 2 + 1; let rollback_dir = Uuid::new_v4(); let mut futures = Vec::with_capacity(disks.len()); diff --git a/rustfs/src/storage/rpc/node_service.rs b/rustfs/src/storage/rpc/node_service.rs index 9df0e2dc7..72247fc08 100644 --- a/rustfs/src/storage/rpc/node_service.rs +++ b/rustfs/src/storage/rpc/node_service.rs @@ -6586,7 +6586,10 @@ mod tests { ); assert!(!remote_object_dir.join(rollback_dir.to_string()).exists()); - *set.disks.write().await = vec![Some(remote.clone()), Some(local_disks[1].clone()), None, None]; + // Model a transient membership snapshot that contains only two + // attached disks. The write quorum must still come from the configured + // four-drive set, otherwise this quorum-minus-one delete is accepted. + *set.disks.write().await = vec![Some(remote.clone()), Some(local_disks[1].clone())]; for _ in 0..3 { let marker = rustfs_filemeta::FileInfo { name: object.to_string(),