perf(persistence): coalesce each AOF group-commit batch into one write

CHANGELOG.md

@@ -6,6 +6,32 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Fixed — Windows build: `accept_backoff` used unix-only `libc` errnos (PR #TBD)
+
+- `is_resource_exhaustion` (accept-loop backoff, PR #230) referenced
+  `libc::EMFILE`/`ENFILE`/`ENOBUFS`/`ENOMEM` unguarded; `libc` is not linked
+  on Windows, breaking the main-push Windows check (PRs never caught it —
+  Windows CI is skipped on PRs). Now cfg-split: unix keeps the errno match,
+  Windows matches WSAEMFILE (10024) / WSAENOBUFS (10055) /
+  `ErrorKind::OutOfMemory`.
+
+### Changed — AOF writer coalesces each group-commit batch into one write (PR #TBD)
+
+- The two monoio AOF writer paths (TopLevel via `commit_group_commit_batch`,
+  PerShard framed loop) wrote each record with its own `write(2)` on the raw
+  unbuffered `File` (the PerShard path even used a header+body pair). strace
+  during always-P16 SET: **127,812 write calls / 1.25 s** of an 8 s window vs
+  2,144 fdatasyncs / 0.2 s — the writer thread was write-syscall-bound, not
+  fsync-bound, while Redis batches ~120 records per write via `aof_buf`. Each
+  batch's records now coalesce into one contiguous buffer (reusable in the
+  PerShard loop, capped at 1 MB high-water) and are written with ONE
+  `write_all` before the single per-batch fsync. Single-message batches keep
+  the zero-copy direct write. Failure semantics unchanged: a failed batch
+  write acks every waiter `WriteFailed` and engages the torn-stream latch.
+- tokio writer loops already amortize via `BufWriter` — untouched.
+- `wal_group_commit` sink tests updated to the coalesced contract (one write,
+  channel-order bytes, fsync after it).
+
 ### Changed — AOF writer polls park-free under everysec/no (PR #TBD)
 
 - The two std-thread AOF writer loops (monoio TopLevel + PerShard) no longer

src/persistence/aof/group_commit.rs

@@ -230,10 +230,29 @@ pub fn commit_group_commit_batch<S: GroupCommitSink + ?Sized>(
     batch: &mut GroupCommitBatch,
     do_fsync: bool,
 ) -> CommitOutcome {
-    // Step 1 — write every data message's bytes in channel order.
-    for msg in &batch.data {
-        if sink.write_all(msg_body(msg)).is_err() {
-            return ack_batch(batch, BatchAck::WriteFailed);
+    // Step 1 — write the batch's bytes in channel order. Multi-message
+    // batches are coalesced into ONE contiguous `write_all` (the sink is a
+    // raw unbuffered File on the monoio paths, so per-message writes meant
+    // one write(2) PER RECORD — measured as the dominant writer-thread cost
+    // under always-P16: 127,812 write calls / 1.25s of an 8s strace window
+    // vs 2,144 fdatasyncs / 0.2s; Redis batches ~120 records per write via
+    // aof_buf). Single-message batches keep the zero-copy direct write.
+    match batch.data.len() {
+        0 => {}
+        1 => {
+            if sink.write_all(msg_body(&batch.data[0])).is_err() {
+                return ack_batch(batch, BatchAck::WriteFailed);
+            }
+        }
+        _ => {
+            let total: usize = batch.data.iter().map(|m| msg_body(m).len()).sum();
+            let mut buf = Vec::with_capacity(total);
+            for msg in &batch.data {
+                buf.extend_from_slice(msg_body(msg));
+            }
+            if sink.write_all(&buf).is_err() {
+                return ack_batch(batch, BatchAck::WriteFailed);
+            }
         }
     }
     // Step 2 — exactly ONE fsync, AFTER all writes, only under Always.

src/persistence/aof/writer_task.rs

@@ -1437,6 +1437,9 @@ pub async fn per_shard_aof_writer_task(
         let mut write_error = false;
         let mut _dbg_processed: u64 = 0;
         let _dbg_start = Instant::now();
+        // Reusable frame-coalescing buffer: one contiguous write_all per
+        // group-commit batch instead of a header+body write pair per record.
+        let mut batch_buf: Vec<u8> = Vec::new();
         // Idle-adaptive channel-poll wake cadence (RSS/CPU wave 5, item B) —
         // see `IdleWait` docs near the top of this file.
         let mut idle_wait = IdleWait::new();
@@ -1510,9 +1513,15 @@ pub async fn per_shard_aof_writer_task(
                         // waiter so callers error instead of acking a corrupt write.
                         let _ = group_commit::ack_batch(&mut batch, BatchAck::WriteFailed);
                     } else {
-                        // Write each record framed, header + body, in channel order;
-                        // the single fsync below covers them all.
-                        let mut write_failed = false;
+                        // Frame each record (`[u64 lsn][u32 len]` header + body, in
+                        // channel order) into the reusable batch buffer and issue
+                        // ONE write_all for the whole batch; the single fsync below
+                        // covers it all. The old per-record header+body write_all
+                        // pair on the raw File was the dominant writer-thread cost
+                        // under always-P16 (measured 127,812 write(2) calls / 1.25s
+                        // of an 8s strace window vs 2,144 fdatasyncs / 0.2s — Redis
+                        // batches ~120 records per write via aof_buf).
+                        batch_buf.clear();
                         for msg in &batch.data {
                             let (lsn, data) = match msg {
                                 AofMessage::Append { lsn, bytes }
@@ -1529,23 +1538,25 @@ pub async fn per_shard_aof_writer_task(
                             let mut header = [0u8; 12];
                             header[..8].copy_from_slice(&lsn.to_le_bytes());
                             header[8..].copy_from_slice(&(data.len() as u32).to_le_bytes());
-                            if let Err(e) = file.write_all(&header) {
-                                error!(
-                                    "AOF header write failed shard {} (seq {}): {}. Persistence degraded.",
-                                    shard_id, manifest.seq, e
-                                );
-                                write_failed = true;
-                                break;
-                            }
-                            if let Err(e) = file.write_all(data) {
+                            batch_buf.extend_from_slice(&header);
+                            batch_buf.extend_from_slice(data);
+                        }
+                        let mut write_failed = false;
+                        if !batch_buf.is_empty() {
+                            if let Err(e) = file.write_all(&batch_buf) {
                                 error!(
-                                    "AOF write failed shard {} (seq {}): {}. Persistence degraded.",
+                                    "AOF batch write failed shard {} (seq {}): {}. Persistence degraded.",
                                     shard_id, manifest.seq, e
                                 );
                                 write_failed = true;
-                                break;
                             }
                         }
+                        // Cap the reusable buffer's high-water mark: a rare burst of
+                        // large values (up to AOF_GROUP_COMMIT_MAX_BYTES) must not
+                        // pin megabytes on the writer thread forever.
+                        if batch_buf.capacity() > 1 << 20 {
+                            batch_buf = Vec::new();
+                        }
 
                         let do_fsync = matches!(fsync, FsyncPolicy::Always);
                         let verdict = if write_failed {

src/server/accept_backoff.rs

@@ -87,13 +87,23 @@ fn backoff_for(consecutive: u32) -> Duration {
 /// Errors that indicate the process/system is out of a resource needed to
 /// accept — retrying immediately would just spin. Everything else is treated
 /// as a transient per-connection error (logged, loop continues, no sleep).
+#[cfg(unix)]
 fn is_resource_exhaustion(err: &std::io::Error) -> bool {
     matches!(
         err.raw_os_error(),
         Some(libc::EMFILE) | Some(libc::ENFILE) | Some(libc::ENOBUFS) | Some(libc::ENOMEM)
     )
 }
 
+/// Windows variant — `libc` is not linked there. WSAEMFILE (10024) and
+/// WSAENOBUFS (10055) are winsock's fd/buffer exhaustion; `OutOfMemory`
+/// covers the ENOMEM-alikes the kind mapping recognizes.
+#[cfg(not(unix))]
+fn is_resource_exhaustion(err: &std::io::Error) -> bool {
+    matches!(err.raw_os_error(), Some(10024) | Some(10055))
+        || matches!(err.kind(), std::io::ErrorKind::OutOfMemory)
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

tests/wal_group_commit.rs

@@ -207,11 +207,21 @@ fn commit_one_fsync_many_acks() {
     let outcome = commit_group_commit_batch(&mut sink, &mut batch, true);
 
     assert_eq!(sink.sync_calls, 1, "exactly ONE fsync for the whole batch");
-    assert_eq!(sink.writes.len(), 3, "all three payloads written");
+    // Multi-message batches are coalesced into ONE contiguous write (the
+    // always-P16 write-syscall fix); channel order must be preserved inside it.
+    assert_eq!(
+        sink.writes.len(),
+        1,
+        "one coalesced write for the whole batch"
+    );
+    assert_eq!(
+        sink.writes[0], b"abc",
+        "all three payloads written in channel order"
+    );
     assert_eq!(
         sink.synced_at_write_count,
-        Some(3),
-        "the fsync runs AFTER all writes (ack-after-fsync ordering)"
+        Some(1),
+        "the fsync runs AFTER the batch write (ack-after-fsync ordering)"
     );
     assert_eq!(
         outcome,
@@ -237,7 +247,9 @@ fn commit_write_fail_acks_write_failed() {
         deferred_control: None,
     };
     let mut sink = CountingSink::new();
-    sink.fail_write_at = Some(1); // the 2nd write fails
+    // Multi-message batches coalesce into a single write — fail THAT write
+    // (index 0). The whole batch fails together: no torn half-batch acks.
+    sink.fail_write_at = Some(0);
 
     let outcome = commit_group_commit_batch(&mut sink, &mut batch, true);
 
@@ -304,7 +316,8 @@ fn commit_everysec_no_fsync() {
         sink.sync_calls, 0,
         "everysec (do_fsync=false) must not fsync per batch"
     );
-    assert_eq!(sink.writes.len(), 2, "bytes are still written in order");
+    assert_eq!(sink.writes.len(), 1, "one coalesced write for the batch");
+    assert_eq!(sink.writes[0], b"xy", "bytes are still written in order");
     assert_eq!(
         outcome.synced, 0,
         "no AppendSync waiters in an everysec batch"
@@ -330,10 +343,19 @@ fn commit_barrier_covers_preceding_appends() {
         sink.sync_calls, 1,
         "one fsync covers the appends + the barrier"
     );
+    assert_eq!(
+        sink.writes.len(),
+        1,
+        "the appends + zero-length barrier coalesce into one write"
+    );
+    assert_eq!(
+        sink.writes[0], b"onetwo",
+        "append bytes precede the fsync in channel order"
+    );
     assert_eq!(
         sink.synced_at_write_count,
-        Some(3),
-        "the fsync runs after the two appends AND the barrier are written"
+        Some(1),
+        "the fsync runs after the appends AND the barrier are written"
     );
     assert_eq!(outcome.synced, 1, "the barrier AppendSync is acked");
     assert_eq!(rx.try_recv(), Ok(AofAck::Synced));