rust volume: group-commit fsync uploads on the direct write path (#11860)

* rust volume: group-commit fsync uploads on the direct write path

Each fsync=true upload used to run its own .dat/.idx sync when the write
queue was off, which is the default. Route durable uploads through the
per-volume write workers so a batch shares one sync, matching Go's
writeNeedle2: non-fsync writes, and writes while the server is stopping,
stay on the synchronous path. The queue is now always constructed and
SEAWEED_WRITE_QUEUE is gone.

Generated with [Devin](https://devin.ai)

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

* rust volume: document post_handler write routing

Generated with [Devin](https://devin.ai)

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

---------

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

seaweed-volume/README.md

@@ -68,7 +68,10 @@ seaweed-volume --port 8080 --master localhost:9333 --dir /data/vol1 --max 7 \
 | `--pprof` | `false` | Enable pprof HTTP handlers |
 
 Set `RUST_LOG=debug` (or `trace`, `info`, `warn`) for log level control.
-Set `SEAWEED_WRITE_QUEUE=1` to enable batched async write processing.
+
+Durable (`fsync=true`) uploads always go through per-volume batching workers
+that share one `.dat`/`.idx` sync per batch, matching the Go volume server;
+non-fsync writes take the synchronous path.
 
 ## Features
 

seaweed-volume/src/config.rs

@@ -303,8 +303,6 @@ pub struct VolumeServerConfig {
     pub grpc_allowed_wildcard_domain: String,
     pub grpc_volume_allowed_common_names: Vec<String>,
     pub tls_policy: TlsPolicy,
-    /// Enable batched write queue for improved throughput under load.
-    pub enable_write_queue: bool,
     /// Path to security.toml — stored for SIGHUP reload.
     pub security_file: String,
     /// If true, FetchAndWriteNeedle skips remote S3 endpoint validation
@@ -968,8 +966,6 @@ fn resolve_config_with_env(cli: Cli, env: EnvLookup<'_>) -> VolumeServerConfig {
         grpc_allowed_wildcard_domain: sec.grpc_allowed_wildcard_domain,
         grpc_volume_allowed_common_names: sec.grpc_volume_allowed_common_names,
         tls_policy: sec.tls_policy,
-        enable_write_queue: env_string(env, "SEAWEED_WRITE_QUEUE")
-            .is_some_and(|v| v == "1" || v == "true"),
         security_file: cli.security_file,
         allow_untrusted_remote_endpoints: cli.allow_untrusted_remote_endpoints,
     }

seaweed-volume/src/main.rs

@@ -433,12 +433,11 @@ async fn run(
         }
     }
 
-    // Initialize the batched write queue if enabled
-    if config.enable_write_queue {
-        info!("Batched write queue enabled");
-        let wq = WriteQueue::new(state.clone(), 128);
-        let _ = state.write_queue.set(wq);
-    }
+    // The write queue is always on: durable (fsync) uploads go through it so
+    // a batch shares one .dat/.idx sync, the way Go's per-volume async writer
+    // group-commits. Non-fsync writes stay on the synchronous path.
+    let wq = WriteQueue::new(state.clone(), 128);
+    let _ = state.write_queue.set(wq);
 
     // Set initial metric gauges for concurrent limits and max volumes
     metrics::CONCURRENT_UPLOAD_LIMIT.set(state.concurrent_upload_limit);

seaweed-volume/src/server/grpc_client.rs

@@ -431,7 +431,6 @@ mod tests {
             grpc_allowed_wildcard_domain: String::new(),
             grpc_volume_allowed_common_names: vec![],
             tls_policy: TlsPolicy::default(),
-            enable_write_queue: false,
             security_file: String::new(),
         }
     }

seaweed-volume/src/server/handlers.rs

@@ -2433,6 +2433,9 @@ fn is_zero_u32(v: &u32) -> bool {
     *v == 0
 }
 
+/// Store an uploaded needle on the local volume — durable (fsync) uploads
+/// through the per-volume write queue, the rest synchronously — and then
+/// replicate it if the volume's placement calls for more copies.
 pub async fn post_handler(
     State(state): State<Arc<VolumeServerState>>,
     request: Request<Body>,
@@ -2953,15 +2956,7 @@ pub async fn post_handler(
     let needle_etag = n.etag();
     let needle_has_name = n.has_name();
 
-    let write_result = if let Some(wq) = state.write_queue.get() {
-        wq.submit(vid, n, fsync).await
-    } else {
-        let vol = {
-            let store = state.store.read().unwrap();
-            store.writable_volume(vid)
-        };
-        vol.and_then(|vol| vol.write_needle(&mut n, true, fsync))
-    };
+    let write_result = write_upload_needle(&state, vid, n, fsync).await;
 
     // Replicate to remote volume servers if this volume has replicas.
     // Matches Go's GetWritableRemoteReplications: skip if copy_count == 1.
@@ -3055,6 +3050,28 @@ pub async fn post_handler(
     resp
 }
 
+/// Append an uploaded needle the way Go's `writeNeedle2` does: a durable
+/// write goes to the volume's batching worker, which group-commits the
+/// batch's .dat/.idx syncs, while anything else — a non-durable write, or a
+/// durable one landing while the server is stopping and the workers are
+/// winding down — takes the synchronous path and fsyncs inline.
+async fn write_upload_needle(
+    state: &Arc<VolumeServerState>,
+    vid: VolumeId,
+    mut n: Needle,
+    fsync: bool,
+) -> Result<(u64, Size, bool), crate::storage::volume::VolumeError> {
+    let queued = fsync && !*state.is_stopping.read().unwrap();
+    if queued && let Some(wq) = state.write_queue.get() {
+        return wq.submit(vid, n, fsync).await;
+    }
+    let vol = {
+        let store = state.store.read().unwrap();
+        store.writable_volume(vid)
+    };
+    vol.and_then(|vol| vol.write_needle(&mut n, true, fsync))
+}
+
 // ============================================================================
 // Delete Handler
 // ============================================================================
@@ -7188,4 +7205,52 @@ mod tests {
             "the vacuum moves the needle once the stream drains"
         );
     }
+
+    /// Issue #801: an fsync upload takes the per-volume worker so its sync is
+    /// shared across the batch (Go's `writeNeedle2` split); a non-fsync write,
+    /// or an fsync write landing while the server is stopping, takes the
+    /// synchronous path instead.
+    #[tokio::test]
+    async fn test_write_upload_needle_routes_fsync_to_queue() {
+        let tmp = tempfile::TempDir::new().unwrap();
+        let state = volume_test_state(&tmp);
+        assert!(
+            state
+                .write_queue
+                .set(crate::server::write_queue::WriteQueue::new(
+                    state.clone(),
+                    128,
+                ))
+                .is_ok()
+        );
+        let queue = state.write_queue.get().unwrap();
+
+        let needle = |id: u64| {
+            let mut n = Needle {
+                id: NeedleId(id),
+                cookie: Cookie(TEST_COOKIE),
+                data: vec![id as u8; 8],
+                data_size: 8,
+                ..Needle::default()
+            };
+            n.checksum = CRC::new(&n.data);
+            n
+        };
+
+        write_upload_needle(&state, VolumeId(1), needle(0x6e7a_0b01), true)
+            .await
+            .unwrap();
+        assert_eq!(queue.submits_for_test(), 1);
+
+        write_upload_needle(&state, VolumeId(1), needle(0x6e7a_0b02), false)
+            .await
+            .unwrap();
+        assert_eq!(queue.submits_for_test(), 1);
+
+        *state.is_stopping.write().unwrap() = true;
+        write_upload_needle(&state, VolumeId(1), needle(0x6e7a_0b03), true)
+            .await
+            .unwrap();
+        assert_eq!(queue.submits_for_test(), 1);
+    }
 }

seaweed-volume/src/server/profiling.rs

@@ -170,7 +170,6 @@ mod tests {
             grpc_allowed_wildcard_domain: String::new(),
             grpc_volume_allowed_common_names: vec![],
             tls_policy: TlsPolicy::default(),
-            enable_write_queue: false,
             security_file: String::new(),
         }
     }

seaweed-volume/src/server/volume_server.rs

@@ -77,7 +77,9 @@ pub struct VolumeServerState {
     pub pre_stop_seconds: u32,
     /// Notify heartbeat to send an immediate update when volume state changes.
     pub volume_state_notify: tokio::sync::Notify,
-    /// Optional batched write queue for improved throughput under load.
+    /// Per-volume write workers that group-commit durable (fsync) uploads.
+    /// Set once at startup in production; a test state may leave it unset,
+    /// which sends every write down the synchronous path.
     pub write_queue: std::sync::OnceLock<WriteQueue>,
     /// Read mode: local, proxy, or redirect for non-local volumes.
     pub read_mode: ReadMode,

seaweed-volume/src/server/write_queue.rs

@@ -1,7 +1,8 @@
 //! Async batched write processing for the volume server.
 //!
-//! Instead of each upload handler directly calling `write_needle`, writes are
-//! submitted to a queue. Each volume gets its own FIFO worker (Go's
+//! Durable (fsync) uploads are submitted here instead of each handler calling
+//! `write_needle` directly — Go's `writeNeedle2` makes the same split, so the
+//! queue is always constructed. Each volume gets its own FIFO worker (Go's
 //! volume_write.go keeps one batching worker per volume), which drains its
 //! channel in batches of up to 128 entries or 4 MiB and shares the .dat/.idx
 //! flushes across durable writes (see `Volume::write_needles_grouped`).
@@ -66,6 +67,9 @@ 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>>>>,
+    /// Submissions accepted, for tests asserting which writes route here.
+    #[cfg(test)]
+    submits: Arc<std::sync::atomic::AtomicU64>,
 }
 
 impl WriteQueue {
@@ -78,9 +82,17 @@ impl WriteQueue {
             io_permits: Arc::new(Semaphore::new(MAX_CONCURRENT_VOLUME_WRITES)),
             queue_permits: Arc::new(Semaphore::new(capacity)),
             workers: Arc::new(Mutex::new(HashMap::new())),
+            #[cfg(test)]
+            submits: Arc::new(std::sync::atomic::AtomicU64::new(0)),
         }
     }
 
+    /// How many writes were submitted, for tests asserting upload routing.
+    #[cfg(test)]
+    pub fn submits_for_test(&self) -> u64 {
+        self.submits.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();
@@ -110,6 +122,9 @@ impl WriteQueue {
     pub async fn submit(&self, volume_id: VolumeId, needle: Needle, fsync: bool) -> WriteResult {
         // Workers spawn only for mounted volumes — a bogus id never creates
         // a channel, so volume-id sprays can't grow the worker map.
+        #[cfg(test)]
+        self.submits
+            .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
         {
             let store = self.state.store.read().unwrap();
             if store.find_volume(volume_id).is_none() {