perf(persistence): move WAL v3 fsync off the shard event loop (#238)

WalWriterV3::flush_sync() ran fdatasync synchronously on the shard
event-loop thread — every fsync stalled the shard (no SPSC drain, no
conn I/O, no CDC fan-out until the disk acks). GCE pd baseline
(tmp/WALV3-OFFLOOP-FSYNC.md §8, c2d-standard-16, 3 reps): the everysec
1s timer froze every connection on the shard for 10-16 ms once per
second whenever the WAL held real bytes, and appendfsync=always paid
-20% RPS / ~2x tail (p999 5.4ms -> 9.8ms) on top of the AOF cost.

Design (spec §3, Option A — fsync offload with a durable-LSN watermark):
the writer keeps encode/buffer/page-cache-write/rotate/recycle on the
shard thread; ONLY the fsync moves.

- New src/persistence/wal_v3/sync_agent.rs: per-shard WalSyncAgent
  std::thread receives fd-dup'd SyncRequests over flume::bounded(8) and
  publishes a monotonic durable-LSN watermark (AtomicU64 fetch_max)
  after each successful fdatasync. fd-dup shares the file description,
  and write_all happens-before try_clone on the shard thread, so each
  fsync covers all previously written bytes; rotation stays safe because
  rotate_segment already fsyncs the old segment inline before switching.
- WalWriterV3::request_sync(): non-blocking initiation. Queue-full /
  dup-failure / agent-spawn-failure all fall back to inline fsync — a
  durability request is never dropped. Agent spawn is lazy (first call)
  and never retried after a failure (warn once, inline forever).
- WalWriterV3::wait_durable(lsn, timeout): bounded blocking wait, used
  ONLY by the two checkpoint ordering invariants and shutdown:
  * log-before-data (flush_dirty_pages page gate, persistence_tick)
  * WAL-before-manifest (checkpoint Finalize now waits on the actual
    checkpoint record LSN before manifest.commit())
  Both bounded by WAIT_DURABLE_TIMEOUT (5s); failure aborts that
  checkpoint step (retried next tick) so redo_lsn never advances past
  durability.
- Failure policy: an fsync error POISONS the agent permanently (POSIX
  post-error fsync semantics are undefined) with tracing::error!;
  subsequent request_sync/wait_durable fail loudly and the checkpoint
  stalls rather than silently opening a data-loss window.
- Call sites: timers::sync_wal_v3 (everysec) and both runtimes'
  appendfsync=always per-drain-batch syncs use request_sync(); shutdown
  paths keep inline flush_sync() (which now also publishes the
  watermark, as does rotation). everysec semantics become "sync
  initiated every 1s, durable typically ms later" — the same window the
  AOF everysec writers provide.

Testing (red/green per tmp/WALV3-OFFLOOP-FSYNC.md §4):
- 7 unit tests in sync_agent.rs with a gated/failing injectable fsync
  backend: watermark-advances-only-after-fsync, wait-blocks-then-
  returns, wait-timeout, poison-fails-loud (wakes parked waiters, never
  publishes, refuses new requests), backpressure-hands-request-back,
  drop-drains-pending-syncs, watermark-monotonic.
- 2 writer-level tests in segment.rs: request_sync+wait_durable covers
  all appends (bytes verified on disk), wait_durable fast paths.
- tests/loom_wal_sync_agent.rs: loom model of the watermark/poison
  monitor (no lost wakeup: publisher takes the mutex before notify,
  waiter re-checks under it) + std smoke variant when not under loom.

Gates (this worktree): fmt clean; clippy 0 warnings both feature sets
(1 pre-existing vendored-monoio warning); cargo test --lib 3864 passed
(monoio) + 3143 passed (tokio); cargo test --no-run compiles all
targets on both feature sets.

Refs: tmp/WALV3-OFFLOOP-FSYNC.md (spec + GCE baseline), stacked on
refactor/wal-v3-only (PR #236).

author: Tin Dang

Co-authored-by: Tin Dang <tindang.ht97@gmail.com>

Author: pilotspacex-byte

CHANGELOG.md

@@ -6,6 +6,31 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Changed — WAL v3 fsync moved off the shard event loop (PR #TBD)
+
+- **`src/persistence/wal_v3/sync_agent.rs` (new)**: per-shard `WalSyncAgent`
+  thread receives fd-dup'd sync requests over a bounded flume channel and
+  publishes a monotonic durable-LSN watermark after each `fdatasync`.
+  `WalWriterV3` gains `request_sync()` (non-blocking initiation; queue-full
+  falls back to inline fsync — a durability request is never dropped) and
+  `wait_durable(lsn, timeout)` (bounded blocking wait, used only by the two
+  checkpoint ordering invariants and shutdown). An fsync error poisons the
+  agent permanently and fails subsequent syncs loudly; the checkpoint then
+  refuses to advance `redo_lsn`.
+- **Why**: `flush_sync()` ran `fdatasync` on the shard event-loop thread.
+  Measured on GCE pd (tmp/WALV3-OFFLOOP-FSYNC.md): the everysec 1s timer
+  froze every connection on the shard for **10–16 ms once per second** when
+  the WAL held real bytes, and `appendfsync always` paid −20% RPS / ~2× tail
+  on top of the AOF cost.
+- **Call sites**: everysec timer (`timers::sync_wal_v3`) and the always-mode
+  per-drain-batch sync (both runtimes) now use `request_sync()`; the
+  checkpoint log-before-data page gate and WAL-before-manifest finalize use
+  `wait_durable` (5s bound); shutdown paths keep the inline `flush_sync()`.
+  everysec semantics are now "sync initiated every 1s, durable typically ms
+  later" — the same window the AOF everysec writers provide.
+- Loom model for the watermark/poison state machine in
+  `tests/loom_wal_sync_agent.rs`.
+
 ### Changed — consolidated dependency bumps wave 2 (PR #TBD, supersedes dependabot #223–227)
 
 - Patch-level bumps rolled into one `Cargo.lock` update (no `Cargo.toml`

src/persistence/wal_v3/mod.rs

@@ -3,6 +3,7 @@
 pub mod record;
 pub mod replay;
 pub mod segment;
+pub(crate) mod sync_agent;
 pub mod tail;
 
 pub use record::{WalRecord, WalRecordType, read_wal_v3_record, write_wal_v3_record};

src/persistence/wal_v3/segment.rs

@@ -124,8 +124,18 @@ pub struct WalWriterV3 {
     min_wal_bytes: u64,
     /// Maximum WAL size in bytes before aggressive recycling (design section 5.5: 256MB default).
     max_wal_bytes: u64,
+    /// Off-loop fsync agent (spawned lazily on the first `request_sync`).
+    /// See `sync_agent` module docs + tmp/WALV3-OFFLOOP-FSYNC.md.
+    sync_agent: Option<super::sync_agent::WalSyncAgent>,
+    /// Set when agent spawn failed once — never retried (inline fsync
+    /// fallback, logged once).
+    sync_agent_unavailable: bool,
 }
 
+/// Bound on every blocking durability wait (checkpoint ordering gates,
+/// shutdown drain). Design-for-failure: no unbounded waits.
+pub const WAIT_DURABLE_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(5);
+
 impl WalWriterV3 {
     /// Create a new WAL v3 writer for the given shard.
     ///
@@ -158,6 +168,8 @@ impl WalWriterV3 {
             epoch: 0,
             min_wal_bytes: DEFAULT_MIN_WAL_BYTES,
             max_wal_bytes: DEFAULT_MAX_WAL_BYTES,
+            sync_agent: None,
+            sync_agent_unavailable: false,
         };
 
         writer.open_new_segment()?;
@@ -203,17 +215,111 @@ impl WalWriterV3 {
         Ok(())
     }
 
-    /// Flush the in-memory buffer to disk and fsync.
+    /// Flush the in-memory buffer to disk and fsync — INLINE, on the caller's
+    /// thread. Kept for shutdown paths and as the no-agent fallback; latency-
+    /// sensitive callers use [`Self::request_sync`] instead.
     ///
     /// After this returns, all appended records are durable on stable storage.
     pub fn flush_sync(&mut self) -> std::io::Result<()> {
         self.flush_write()?;
         if let Some(ref mut file) = self.current_file {
             file.sync_data()?;
         }
+        // Keep the off-loop watermark honest: everything appended so far is
+        // now durable (fetch_max — never regresses a higher agent publish).
+        if let Some(agent) = &self.sync_agent {
+            agent.shared.publish(self.next_lsn.saturating_sub(1));
+        }
         Ok(())
     }
 
+    /// Spawn the sync agent on first use; on failure, log once and fall
+    /// back to inline fsync forever (never retried per call).
+    fn spawn_agent_if_needed(&mut self) {
+        if self.sync_agent.is_none() && !self.sync_agent_unavailable {
+            match super::sync_agent::WalSyncAgent::spawn(self.shard_id) {
+                Ok(agent) => self.sync_agent = Some(agent),
+                Err(e) => {
+                    tracing::warn!(
+                        shard_id = self.shard_id,
+                        "WAL v3 sync agent spawn failed — falling back to \
+                         inline fsync permanently: {e}"
+                    );
+                    self.sync_agent_unavailable = true;
+                }
+            }
+        }
+    }
+
+    /// Initiate durability for everything appended so far WITHOUT blocking
+    /// on the fsync: write the buffer to the page cache, then hand an
+    /// fd-dup to the off-loop sync agent. Queue-full / dup-failure fall
+    /// back to an inline fsync — a durability request is never dropped.
+    ///
+    /// Errors: propagates page-cache write failures and reports a poisoned
+    /// agent (a prior off-loop fsync failed — durability can no longer be
+    /// promised on this WAL; fail loud, never silently degrade).
+    pub fn request_sync(&mut self) -> std::io::Result<()> {
+        self.flush_write()?;
+        let upto_lsn = self.next_lsn.saturating_sub(1);
+        self.spawn_agent_if_needed();
+
+        if let Some(agent) = &self.sync_agent {
+            if agent.is_poisoned() {
+                return Err(std::io::Error::other(
+                    "WAL v3 sync agent poisoned by a prior fsync failure",
+                ));
+            }
+            if agent.durable_lsn() >= upto_lsn {
+                return Ok(()); // nothing new since the last durable point
+            }
+            let Some(file) = &self.current_file else {
+                return Ok(());
+            };
+            if let Ok(dup) = file.try_clone() {
+                if agent
+                    .try_send(super::sync_agent::SyncRequest {
+                        file: dup,
+                        upto_lsn,
+                    })
+                    .is_ok()
+                {
+                    return Ok(());
+                }
+                // Queue full (disk is the bottleneck) or agent racing
+                // poison — fall through to the inline path.
+            }
+        }
+
+        // Inline fallback: no agent / dup failed / queue full.
+        self.flush_sync()
+    }
+
+    /// Block until everything up to `lsn` is durable (bounded by `timeout`).
+    ///
+    /// Used ONLY by the checkpoint ordering invariants (log-before-data,
+    /// WAL-before-manifest) and shutdown drains — hot paths use
+    /// [`Self::request_sync`].
+    pub fn wait_durable(&mut self, lsn: u64, timeout: std::time::Duration) -> std::io::Result<()> {
+        if lsn == 0 {
+            return Ok(());
+        }
+        if let Some(agent) = &self.sync_agent {
+            if agent.durable_lsn() >= lsn {
+                return Ok(());
+            }
+        }
+        // Make sure a sync covering `lsn` is in flight (or completed
+        // inline, which publishes the watermark itself).
+        self.request_sync()?;
+        match &self.sync_agent {
+            Some(agent) => agent.wait_watermark(lsn, timeout),
+            // No agent: request_sync went through the inline flush_sync
+            // path, so durability already holds.
+            None => Ok(()),
+        }
+    }
+
     /// Flush if buffer exceeds a threshold — write only, no fsync.
     ///
     /// Matches WAL v2 pattern: frequent writes to OS page cache,
@@ -445,6 +551,13 @@ impl WalWriterV3 {
             }
             file.sync_data()?;
         }
+        // The old segment (holding every record < next_lsn) is now durable;
+        // the off-loop watermark can reflect that. This inline fsync at
+        // rotation is also what makes the agent's fd-dup scheme safe: a
+        // sync request only ever needs to cover the CURRENT segment.
+        if let Some(agent) = &self.sync_agent {
+            agent.shared.publish(self.next_lsn.saturating_sub(1));
+        }
 
         self.current_sequence += 1;
         crate::admin::metrics_setup::record_wal_rotation();
@@ -726,6 +839,65 @@ mod tests {
         assert_eq!(meta.len(), WAL_V3_HEADER_SIZE as u64);
     }
 
+    #[test]
+    fn test_request_sync_then_wait_durable_covers_all_appends() {
+        let tmp = tempfile::tempdir().unwrap();
+        let wal_dir = tmp.path().join("wal");
+        let mut writer = WalWriterV3::new(0, &wal_dir, DEFAULT_SEGMENT_SIZE).unwrap();
+
+        let mut last = 0;
+        for i in 0..10u32 {
+            last = writer.append(WalRecordType::Command, format!("SET k{i} v").as_bytes());
+        }
+        // Non-blocking initiation, then a bounded wait must observe
+        // durability for every appended record.
+        writer.request_sync().unwrap();
+        writer
+            .wait_durable(last, std::time::Duration::from_secs(5))
+            .unwrap();
+
+        // Bytes must actually be on disk (page-cache write happened before
+        // the agent's fsync request).
+        let data = fs::read(WalSegment::segment_path(&wal_dir, 1)).unwrap();
+        let mut offset = WAL_V3_HEADER_SIZE;
+        let mut count = 0;
+        while offset < data.len() {
+            let record = read_wal_v3_record(&data[offset..]).expect("record parses");
+            // record_len already includes its own 4-byte length prefix.
+            offset += u32::from_le_bytes([
+                data[offset],
+                data[offset + 1],
+                data[offset + 2],
+                data[offset + 3],
+            ]) as usize;
+            count += 1;
+            assert!(record.lsn <= last);
+        }
+        assert_eq!(count, 10);
+    }
+
+    #[test]
+    fn test_wait_durable_zero_and_already_durable_are_noops() {
+        let tmp = tempfile::tempdir().unwrap();
+        let wal_dir = tmp.path().join("wal");
+        let mut writer = WalWriterV3::new(0, &wal_dir, DEFAULT_SEGMENT_SIZE).unwrap();
+        // lsn 0 = "nothing to wait for" — must not spawn or block.
+        writer
+            .wait_durable(0, std::time::Duration::from_millis(10))
+            .unwrap();
+        let lsn = writer.append(WalRecordType::Command, b"SET a 1");
+        // Inline flush_sync publishes the watermark, so a subsequent
+        // wait_durable is a fast-path no-op even with an agent spawned.
+        writer.request_sync().unwrap(); // spawns agent
+        writer
+            .wait_durable(lsn, std::time::Duration::from_secs(5))
+            .unwrap();
+        writer.flush_sync().unwrap();
+        writer
+            .wait_durable(lsn, std::time::Duration::from_millis(10))
+            .unwrap();
+    }
+
     #[test]
     fn test_writer_append_and_flush() {
         let tmp = tempfile::tempdir().unwrap();