fix(ecstore): preserve vectored bitrot writes (#7981)

* fix(ecstore): preserve vectored bitrot writes

Co-Authored-By: heihutu <heihutu@gmail.com>

Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

* fix(ecstore): route test trait through contracts

Co-Authored-By: heihutu <heihutu@gmail.com>

Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

---------

Co-authored-by: zhi22915 <qiuzgang@gmail.com>

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<std::io::Result<usize>> {
+        // 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<std::io::Result<()>> {
         // 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");

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");
+    }
 }

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());

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(),