volume: keep a below-minimum fsync batch open for stragglers (#12016)

* storage: hold a below-minimum fsync batch open for stragglers

When the durable-write queue drained, the worker flushed whatever it
held, so low-concurrency fsync writes collapsed to one sync per
request -- a seek plus a rotation on rotational media no matter the
payload. Once the queue empties with the batch still below the
configured minimum bytes and requests, the batch now stays open up to
SyncBatchOption.Wait, armed once per batch so the added latency stays
bounded. DiskLocation carries the option and defaults it; Store exposes
a setter the server calls once at startup.

* volume server: expose the fsync batch window as -fsync* flags

-fsyncMinBatchBytes, -fsyncMinBatchRequests and -fsyncBatchWait carry
the operator's batching floor into every disk's SyncBatchOption.
-fsyncBatchWait=0 restores the previous flush-on-drain behavior.

* server/mini: register the fsync batch window flags

weed server and weed mini build their own VolumeServerOptions and left
the three new fsync flag pointers nil, so startVolumeServer panicked
dereferencing them at startup.

* storage: bound the fsync batch window by Wait while the queue refills

Once the straggler timer was armed, a steady trickle of queued writes
kept taking the len(requests)>0 continue path, so the first write could
wait past Wait until the size or count cap flushed the batch. Track the
armed deadline and flush once it passes even with requests buffered.

* storage: sync the straggler test on the worker taking the first write

The fixed sleep let both writes queue together (passing under the old
flush-on-drain behavior) or the straggler arrive after the window closed
(flaky failure). Poll the drained queue instead so the second write can
only land inside the armed window or a later batch.

* volume server: default the fsync batch window when its options are unset

weed server and weed mini build VolumeServerOptions without the fsync
flag pointers, and outside embedders can too. Resolve the batch window
from the defaults so startVolumeServer never dereferences nil options.

* storage: drive the batch-window test through an injectable timer

The window is armed by the worker now, so the straggler provably joins
an open batch and expiry is fired by hand instead of racing a sleep.

* seaweed-volume: accept -fsyncMinBatch/-fsyncBatchWait flags

Same flags and defaults as weed volume so the Rust server's durable
write batching can be tuned identically.

* seaweed-volume: hold a below-minimum fsync batch open for stragglers

Mirrors the Go worker's batchWait: once the queue drains under the
minimum batch, the window stays open for stragglers and an expired
window still bounds the batch while the queue refills. Paused-time
coverage replaces any sleep-based assertion.

* seaweed-volume: sync the straggler test on the armed batch window

The test fired both writes together, so the straggler could already be
queued when the worker drained — a regression that drops mid-window
writes from the open batch would still pass. Wait until the worker
arms the window before sending the last write (Go's test polls the
drained channel for the same guarantee).

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* seaweed-volume: reject fsyncMinBatchBytes that does not fit usize

---------

Co-authored-by: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

seaweed-volume/src/server/write_queue.rs

@@ -56,6 +56,27 @@ const MAX_CONCURRENT_VOLUME_WRITES: usize = 8;
 /// retirement is the volume leaving the store.
 const IDLE_WORKER_POLL: std::time::Duration = std::time::Duration::from_secs(60);
 
+/// Tunes the per-volume durable-write worker. Once the queue drains with the
+/// batch still below min_bytes and min_requests, the batch stays open up to
+/// wait for stragglers instead of paying an fsync per request. Mirrors Go's
+/// storage.SyncBatchOption.
+#[derive(Clone, Copy)]
+pub struct SyncBatchOption {
+    pub min_bytes: usize,
+    pub min_requests: usize,
+    pub wait: std::time::Duration,
+}
+
+impl Default for SyncBatchOption {
+    fn default() -> Self {
+        SyncBatchOption {
+            min_bytes: 1024 * 1024,
+            min_requests: MAX_BATCH_SIZE,
+            wait: std::time::Duration::from_micros(500),
+        }
+    }
+}
+
 /// Handle for submitting write requests to the per-volume workers.
 #[derive(Clone)]
 pub struct WriteQueue {
@@ -67,23 +88,35 @@ pub struct WriteQueue {
     /// alone would multiply the old single queue's capacity by volume count.
     queue_permits: Arc<Semaphore>,
     workers: Arc<Mutex<HashMap<VolumeId, mpsc::Sender<WriteRequest>>>>,
+    sync_batch: SyncBatchOption,
     /// Submissions accepted, for tests asserting which writes route here.
     #[cfg(test)]
     submits: Arc<std::sync::atomic::AtomicU64>,
+    /// Times a worker armed the straggler window, for tests that must land a
+    /// write inside an already-open batch.
+    #[cfg(test)]
+    windows_armed: Arc<std::sync::atomic::AtomicU64>,
 }
 
 impl WriteQueue {
     /// Create a new write queue. Per-volume workers are spawned lazily on the
     /// first submission for that volume.
-    pub fn new(state: Arc<VolumeServerState>, capacity: usize) -> Self {
+    pub fn new(
+        state: Arc<VolumeServerState>,
+        capacity: usize,
+        sync_batch: SyncBatchOption,
+    ) -> Self {
         WriteQueue {
             state,
             capacity,
             io_permits: Arc::new(Semaphore::new(MAX_CONCURRENT_VOLUME_WRITES)),
             queue_permits: Arc::new(Semaphore::new(capacity)),
             workers: Arc::new(Mutex::new(HashMap::new())),
+            sync_batch,
             #[cfg(test)]
             submits: Arc::new(std::sync::atomic::AtomicU64::new(0)),
+            #[cfg(test)]
+            windows_armed: Arc::new(std::sync::atomic::AtomicU64::new(0)),
         }
     }
 
@@ -93,6 +126,12 @@ impl WriteQueue {
         self.submits.load(std::sync::atomic::Ordering::Relaxed)
     }
 
+    /// How many batch windows workers have armed so far.
+    #[cfg(test)]
+    fn windows_armed_for_test(&self) -> u64 {
+        self.windows_armed.load(std::sync::atomic::Ordering::Relaxed)
+    }
+
     /// Sender for the volume's worker, spawning one on first use.
     fn worker_for(&self, volume_id: VolumeId) -> mpsc::Sender<WriteRequest> {
         let mut workers = self.workers.lock().unwrap();
@@ -107,6 +146,9 @@ impl WriteQueue {
                         state: self.state.clone(),
                         io_permits: self.io_permits.clone(),
                         workers: self.workers.clone(),
+                        sync_batch: self.sync_batch,
+                        #[cfg(test)]
+                        windows_armed: self.windows_armed.clone(),
                     }
                     .run(),
                 );
@@ -191,6 +233,9 @@ struct VolumeWriteWorker {
     state: Arc<VolumeServerState>,
     io_permits: Arc<Semaphore>,
     workers: Arc<Mutex<HashMap<VolumeId, mpsc::Sender<WriteRequest>>>>,
+    sync_batch: SyncBatchOption,
+    #[cfg(test)]
+    windows_armed: Arc<std::sync::atomic::AtomicU64>,
 }
 
 impl VolumeWriteWorker {
@@ -238,14 +283,40 @@ impl VolumeWriteWorker {
         let mut batch = Vec::with_capacity(MAX_BATCH_SIZE);
         let mut batch_bytes: usize = first.needle.data.len();
         batch.push(first);
+        // Armed once the queue first drains below the minimum batch; holds the
+        // batch open for stragglers so the wait bound is per batch, not per
+        // lull (Go's batchWait).
+        let mut window_deadline = None;
         while batch.len() < MAX_BATCH_SIZE && batch_bytes < MAX_BATCH_BYTES {
-            match self.rx.try_recv() {
-                Ok(req) => {
-                    batch_bytes += req.needle.data.len();
-                    batch.push(req);
-                }
-                Err(_) => break,
+            if window_deadline.is_some_and(|d| tokio::time::Instant::now() >= d) {
+                break;
             }
+            let req = match self.rx.try_recv() {
+                Ok(req) => req,
+                Err(mpsc::error::TryRecvError::Empty) => {
+                    let sync = self.sync_batch;
+                    if sync.wait.is_zero()
+                        || batch_bytes >= sync.min_bytes
+                        || batch.len() >= sync.min_requests
+                    {
+                        // a drained queue flushes right away once it holds enough
+                        break;
+                    }
+                    let deadline = *window_deadline.get_or_insert_with(|| {
+                        #[cfg(test)]
+                        self.windows_armed
+                            .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
+                        tokio::time::Instant::now() + sync.wait
+                    });
+                    match tokio::time::timeout_at(deadline, self.rx.recv()).await {
+                        Ok(Some(req)) => req,
+                        Ok(None) | Err(_) => break,
+                    }
+                }
+                Err(mpsc::error::TryRecvError::Disconnected) => break,
+            };
+            batch_bytes += req.needle.data.len();
+            batch.push(req);
         }
         debug!("processing write batch of {} requests", batch.len());
 
@@ -380,7 +451,7 @@ mod tests {
     async fn test_write_queue_submit_no_volume() {
         // Submit a write to a non-existent volume -- should return VolumeError::NotFound.
         let state = make_test_state();
-        let queue = WriteQueue::new(state, MAX_BATCH_SIZE);
+        let queue = WriteQueue::new(state, MAX_BATCH_SIZE, SyncBatchOption::default());
 
         let needle = Needle {
             id: 1.into(),
@@ -402,7 +473,7 @@ mod tests {
     async fn test_write_queue_concurrent_submissions() {
         // Submit multiple concurrent writes -- all should complete (with errors since no volume).
         let state = make_test_state();
-        let queue = WriteQueue::new(state, MAX_BATCH_SIZE);
+        let queue = WriteQueue::new(state, MAX_BATCH_SIZE, SyncBatchOption::default());
 
         let mut handles = Vec::new();
         for i in 0..10u64 {
@@ -430,7 +501,7 @@ mod tests {
     async fn test_write_queue_batching() {
         // Verify that many concurrent writes get processed (testing the batching path).
         let state = make_test_state();
-        let queue = WriteQueue::new(state, MAX_BATCH_SIZE);
+        let queue = WriteQueue::new(state, MAX_BATCH_SIZE, SyncBatchOption::default());
 
         // Submit MAX_BATCH_SIZE requests concurrently
         let mut handles = Vec::new();
@@ -522,7 +593,7 @@ mod tests {
     async fn test_write_queue_dropped_sender() {
         // When the queue is dropped, subsequent submits should fail gracefully.
         let state = make_test_state();
-        let queue = WriteQueue::new(state, 1);
+        let queue = WriteQueue::new(state, 1, SyncBatchOption::default());
 
         // Clone then drop the original -- the worker keeps running via its rx handle.
         let queue2 = queue.clone();
@@ -665,7 +736,7 @@ mod tests {
                     .unwrap();
             }
         }
-        let queue = WriteQueue::new(state, MAX_BATCH_SIZE);
+        let queue = WriteQueue::new(state, MAX_BATCH_SIZE, SyncBatchOption::default());
 
         let in_append = Arc::new(Barrier::new(2));
         let release = Arc::new(Barrier::new(2));
@@ -734,4 +805,85 @@ mod tests {
         assert!(a1.await.unwrap().is_ok());
         assert!(a2.await.unwrap().is_ok());
     }
+
+    /// A durable write below the minimum batch rides out its window so a
+    /// straggler joins the same sync instead of each paying an fsync alone —
+    /// Go's batchWait. Paused time makes the hold observable: a lone write can
+    /// only complete once the runtime auto-advances to the window's deadline.
+    #[tokio::test(start_paused = true)]
+    async fn test_batch_window_defers_flush_for_stragglers() {
+        use crate::config::MinFreeSpace;
+        use crate::storage::types::{DiskType, NeedleId};
+        use crate::storage::volume::VolumeSpec;
+        use std::time::Duration;
+
+        let tmp = tempfile::TempDir::new().unwrap();
+        let dir = tmp.path().to_str().unwrap();
+        let state = make_test_state();
+        {
+            let mut store = state.store.write().unwrap();
+            store
+                .add_location(
+                    dir,
+                    dir,
+                    10,
+                    DiskType::HardDrive,
+                    MinFreeSpace::Percent(1.0),
+                    Vec::new(),
+                )
+                .unwrap();
+            store
+                .add_volume(VolumeId(1), DiskType::HardDrive, &VolumeSpec::default())
+                .unwrap();
+        }
+
+        let queue = WriteQueue::new(
+            state,
+            MAX_BATCH_SIZE,
+            SyncBatchOption {
+                min_bytes: usize::MAX,
+                min_requests: MAX_BATCH_SIZE,
+                wait: Duration::from_secs(30),
+            },
+        );
+
+        let needle = |id: u64| Needle {
+            id: NeedleId(id),
+            cookie: 0x1111.into(),
+            data: vec![id as u8; 8],
+            data_size: 8,
+            ..Needle::default()
+        };
+
+        let start = tokio::time::Instant::now();
+        queue.submit(VolumeId(1), needle(1), true).await.unwrap();
+        assert!(start.elapsed() >= Duration::from_secs(30));
+
+        // Wait for the worker to take the second write and open its batch
+        // window, so the straggler provably lands inside an open batch
+        // instead of arriving during the initial drain.
+        let armed = queue.windows_armed_for_test();
+        let second = tokio::spawn({
+            let queue = queue.clone();
+            async move { queue.submit(VolumeId(1), needle(2), true).await }
+        });
+        for _ in 0..10_000 {
+            if queue.windows_armed_for_test() > armed {
+                break;
+            }
+            tokio::task::yield_now().await;
+        }
+        assert!(queue.windows_armed_for_test() > armed);
+        let third = tokio::spawn({
+            let queue = queue.clone();
+            async move { queue.submit(VolumeId(1), needle(3), true).await }
+        });
+        let (second, third) = tokio::join!(second, third);
+        second.unwrap().unwrap();
+        third.unwrap().unwrap();
+        // One shared window ends both writes ~30s after their batch armed;
+        // back-to-back windows would keep the third out to ~90s.
+        let elapsed = start.elapsed();
+        assert!(elapsed >= Duration::from_secs(60) && elapsed < Duration::from_secs(90));
+    }
 }