feat(wal): overlap bounded replication with local persistence

crates/lance-context-wal/src/raft_log.rs

@@ -562,6 +562,17 @@ pub struct GroupLog {
 }
 
 impl GroupLog {
+    #[cfg(all(test, feature = "runtime"))]
+    pub(crate) fn block_sync_for_test(&self) -> std::sync::MutexGuard<'_, ()> {
+        self.node.sync_gate.lock().unwrap()
+    }
+
+    /// Engine append visibility is deliberately not a sync/durability receipt.
+    #[cfg(all(test, feature = "runtime"))]
+    pub(crate) fn appended_index_for_test(&self) -> Option<u64> {
+        self.node.engine.last_index(self.group)
+    }
+
     #[cfg(all(test, feature = "replication"))]
     pub(crate) fn fail_next_sync_for_test(&self) {
         self.node.fault.fail_sync_for_test();

crates/lance-context-wal/src/replication.rs

@@ -825,20 +825,55 @@ impl Node {
     /// Performs blocking storage I/O. The transport may drop output messages;
     /// Raft will retransmit. On error NONE of this call's events are released.
     pub fn poll(&mut self) -> Result<Events> {
+        self.poll_inner(None)
+    }
+
+    /// Dispatch ordinary non-snapshot Ready messages before local persistence.
+    /// The callback must not block. Messages requiring persistence, snapshots,
+    /// completions and quorum reads retain the synchronous poll boundary.
+    /// An error may follow an early message; it never establishes a durable ACK.
+    #[cfg(feature = "runtime")]
+    pub(crate) fn poll_with_early_messages(
+        &mut self,
+        mut send: impl FnMut(Packet),
+    ) -> Result<Events> {
+        self.poll_inner(Some(&mut send))
+    }
+
+    fn poll_inner(&mut self, send: Option<&mut dyn FnMut(Packet)>) -> Result<Events> {
         self.healthy()?;
-        let result = self.drive();
+        let result = self.drive(send);
         if result.is_err() {
             self.failed = true;
         }
         result
     }
 
-    fn drive(&mut self) -> Result<Events> {
+    fn drive(&mut self, mut early: Option<&mut dyn FnMut(Packet)>) -> Result<Events> {
         let mut events = Events::default();
         for (id, group) in &mut self.groups {
             let mut outbound = Vec::new();
             while group.raw.has_ready() {
                 let mut ready = group.raw.ready();
+                if let Some(send) = early.as_deref_mut() {
+                    // raft-rs separates these from persisted_messages. Keep
+                    // snapshots on the existing captured-image/feedback path.
+                    for message in ready.take_messages() {
+                        if message.snapshot.is_some()
+                            || message.get_msg_type() == MessageType::MsgSnapshot
+                        {
+                            outbound.push(message);
+                        } else {
+                            send(Packet {
+                                cluster: self.layout.cluster,
+                                layout: self.layout_hash,
+                                policy: self.policy,
+                                group: *id,
+                                message,
+                            });
+                        }
+                    }
+                }
                 if ready.snapshot().is_empty() {
                     group.raw.store().persist(ready.entries(), ready.hs())?;
                 } else {

crates/lance-context-wal/src/replication/runtime.rs

@@ -51,6 +51,10 @@ pub struct Options {
     pub tick_interval: Duration,
     /// Maximum commands from each lane before polling durable Raft progress.
     pub batch_requests: usize,
+    /// Overlap ordinary peer messages with local persistence. Persistence-bound
+    /// replies, snapshots and client completions retain their existing barriers.
+    /// Experimental; disabled until a controlled comparison supports it.
+    pub pipeline_replication: bool,
 }
 impl Default for Options {
     fn default() -> Self {
@@ -63,6 +67,7 @@ impl Default for Options {
             peer_bytes: 64 << 20,
             tick_interval: Duration::from_millis(100),
             batch_requests: 32,
+            pipeline_replication: false,
         }
     }
 }
@@ -751,7 +756,32 @@ impl Worker {
             }
             if activity {
                 self.observation.lock().unwrap().polling = true;
-                let events = self.node.poll()?;
+                let events = if self.options.pipeline_replication {
+                    let shared = &self.shared;
+                    let outbox = &self.outbox;
+                    self.node.poll_with_early_messages(|packet| {
+                        // Move the packet, including shared payload buffers,
+                        // through the same node-wide outbound credits. A full
+                        // transport never blocks the durable owner thread.
+                        let bytes = packet.message.compute_size() as usize;
+                        if let Ok(credit) = shared.outgoing.reserve(bytes) {
+                            if outbox
+                                .try_send(Outbound {
+                                    packet,
+                                    _credit: Some(credit),
+                                    image: None,
+                                    snapshot_feedback: None,
+                                })
+                                .is_ok()
+                            {
+                                return;
+                            }
+                        }
+                        shared.dropped_outgoing.fetch_add(1, Ordering::Relaxed);
+                    })?
+                } else {
+                    self.node.poll()?
+                };
                 for done in events.completions {
                     let pending = self
                         .writes

crates/lance-context-wal/src/replication/runtime/tests.rs

@@ -6,6 +6,7 @@ use crate::replication::{
 use std::path::Path;
 use tokio::time::{sleep, timeout};
 
+mod early;
 mod images;
 
 // Actual dedicated workers and durable engines. This relay deliberately drops

crates/lance-context-wal/src/replication/runtime/tests/early.rs

@@ -0,0 +1,158 @@
+use super::*;
+use raft_proto::eraftpb::MessageType;
+
+fn open_nodes(root: &Path) -> Result<Vec<Node>> {
+    (1..=3)
+        .map(|id| Node::open(&root.join(id.to_string()), layout(), id, Limits::default()))
+        .collect()
+}
+
+// The test thread owns the disk barrier. Actual dedicated group workers and
+// real log files perform the append; block_on only drives their async clients.
+#[test]
+fn pipelined_append_reaches_a_follower_during_blocked_leader_sync_without_ack() -> Result<()> {
+    assert!(!Options::default().pipeline_replication);
+    let executor = tokio::runtime::Runtime::new()?;
+    for fail_sync in [false, true] {
+        let directory = tempfile::tempdir()?;
+        let mut nodes = open_nodes(directory.path())?;
+        nodes[0].campaign(10)?;
+        crate::replication::tests::pump(&mut nodes)?;
+        let node = nodes.remove(0);
+        let store = node.groups[&10].raw.store().clone();
+        let (runtime, mut outbox) = Runtime::spawn(
+            node,
+            Options {
+                pipeline_replication: true,
+                peer_requests: 1,
+                peer_bytes: 16 * 1024,
+                tick_interval: Duration::from_secs(1),
+                ..Options::default()
+            },
+        )?;
+        let handle = runtime.handle();
+        let blocked = store.block_sync_for_test();
+        if fail_sync {
+            store.fail_next_sync_for_test();
+        }
+        let body = batch(1);
+        let pointer = body.body.as_ptr();
+        let pending = handle.commit(Command::Append {
+            group: 10,
+            batch: body.clone(),
+        })?;
+        let outgoing = executor.block_on(async {
+            timeout(Duration::from_secs(5), async {
+                loop {
+                    let value = outbox.recv().await.context("outbox closed")?;
+                    if value.packet().message.get_msg_type() == MessageType::MsgAppend
+                        && value
+                            .packet()
+                            .message
+                            .entries
+                            .iter()
+                            .any(|entry| entry.data == body.body)
+                    {
+                        return Ok::<_, anyhow::Error>(value);
+                    }
+                }
+            })
+            .await?
+        })?;
+        assert!(outgoing.snapshot_image().is_none());
+        let entry = outgoing
+            .packet()
+            .message
+            .entries
+            .iter()
+            .find(|entry| entry.data == body.body)
+            .unwrap();
+        let original_index = entry.index;
+        assert_eq!(
+            entry.data.as_ptr(),
+            pointer,
+            "early transport copied payload"
+        );
+        executor.block_on(wait_for(|| {
+            store.appended_index_for_test() == Some(original_index)
+        }))?;
+        // The second destination cannot bypass a held packet's node-wide slot.
+        assert!(handle.peer_drops().0 > 0);
+        assert_eq!(handle.shared.outgoing.slots.available_permits(), 0);
+        let follower = outgoing.packet().message.to as usize - 2;
+        nodes[follower].step(outgoing.packet().clone())?;
+        let response = nodes[follower].poll()?;
+        assert!(response.messages.iter().any(|packet| {
+            packet.message.get_msg_type() == MessageType::MsgAppendResponse
+                && !packet.message.reject
+                && packet.message.index == original_index
+        }));
+        for packet in response.messages {
+            handle.step(packet)?;
+        }
+        assert!(pending.0.is_empty(), "quorum response bypassed leader sync");
+        drop(outgoing);
+        assert_eq!(handle.shared.outgoing.slots.available_permits(), 1);
+        drop(blocked);
+        let outcome =
+            executor.block_on(async { timeout(Duration::from_secs(5), pending).await })?;
+        if fail_sync {
+            assert!(outcome.is_err(), "failed leader sync released a client ACK");
+            executor.block_on(wait_for(|| {
+                handle.observation(10).unwrap().stopped && handle.observation(20).unwrap().stopped
+            }))?;
+            assert!(executor.block_on(runtime.shutdown()).is_err());
+        } else {
+            assert_eq!(index(outcome?), original_index);
+            executor.block_on(runtime.shutdown())?;
+        }
+        drop(outbox);
+        drop(handle);
+        drop(store);
+        drop(nodes);
+
+        // The failed-sync case is uncertain, not rejected. Its follower has a
+        // durable append; elect that follower after reopening every directory.
+        let mut nodes = open_nodes(directory.path())?;
+        nodes[follower + 1].campaign(10)?;
+        crate::replication::tests::pump(&mut nodes)?;
+        for node in &nodes {
+            assert_eq!(node.read_accepted(10, original_index)?, Some(body.clone()));
+        }
+        nodes[follower + 1].propose(10, 99, body)?;
+        let retried = crate::replication::tests::pump(&mut nodes)?;
+        let retried = retried.into_iter().find(|done| done.request == 99).unwrap();
+        assert_eq!(index(retried.outcome), original_index);
+    }
+    Ok(())
+}
+
+#[test]
+fn pipelined_follower_sync_failure_releases_no_append_response() -> Result<()> {
+    let directory = tempfile::tempdir()?;
+    let mut nodes = open_nodes(directory.path())?;
+    nodes[0].campaign(10)?;
+    crate::replication::tests::pump(&mut nodes)?;
+    nodes[0].propose(10, 1, batch(1))?;
+    let mut packets = Vec::new();
+    let events = nodes[0].poll_with_early_messages(|packet| packets.push(packet))?;
+    assert!(events.completions.is_empty());
+    let append = packets
+        .into_iter()
+        .find(|packet| {
+            packet.message.to == 2
+                && packet.message.get_msg_type() == MessageType::MsgAppend
+                && !packet.message.entries.is_empty()
+        })
+        .context("missing early append")?;
+    nodes[1].step(append)?;
+    nodes[1].groups[&10].raw.store().fail_next_sync_for_test();
+    let mut responses = Vec::new();
+    assert!(nodes[1]
+        .poll_with_early_messages(|packet| responses.push(packet))
+        .is_err());
+    assert!(responses.is_empty(), "follower replied before durable sync");
+    assert!(nodes[0].poll()?.completions.is_empty());
+    assert!(nodes[1].status(20).is_err());
+    Ok(())
+}

crates/lance-context-wal/src/replication/runtime/tests/images.rs

@@ -72,8 +72,21 @@ async fn collect(handle: &Handle) -> Result<u64> {
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn runtime_snapshot_catches_up_compacted_follower_and_preserves_restart_retries() -> Result<()>
 {
+    snapshot_restart_retries(false).await
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn pipelined_replication_keeps_snapshot_ownership_and_restart_retries() -> Result<()> {
+    snapshot_restart_retries(true).await
+}
+
+async fn snapshot_restart_retries(pipeline_replication: bool) -> Result<()> {
+    let options = Options {
+        pipeline_replication,
+        ..Options::default()
+    };
     let root = tempfile::tempdir()?;
-    let c = Cluster::open(root.path(), Options::default())?;
+    let c = Cluster::open(root.path(), options.clone())?;
     c.elect(1, 10).await?;
     c.elect(1, 20).await?;
     c.isolated.store(3, Ordering::Release);
@@ -129,7 +142,7 @@ async fn runtime_snapshot_catches_up_compacted_follower_and_preserves_restart_re
             assert_eq!(node.read_accepted(*group, *i)?, Some(b.clone()));
         }
     }
-    let c = Cluster::open(root.path(), Options::default())?;
+    let c = Cluster::open(root.path(), options)?;
     c.elect(3, 10).await?;
     c.elect(3, 20).await?;
     // Only the latest identity is exactly retryable; sequence5 must not have

docs/design/wal-replication-pipeline.md

@@ -0,0 +1,35 @@
+# Overlap replication messages and local persistence
+
+The WAL runtime normally delivers peer packets after the entire synchronous
+Raft poll finishes. An append therefore waits for the leader's local storage
+before a follower can begin storing that append.
+
+The experimental node setting `runtime.pipeline_replication: true` dispatches
+ordinary, non-snapshot `Ready.messages` through the existing bounded outbound
+queue before the local persistence call. Packets move their shared payload
+buffers into that queue. Byte and request credits remain owned through transport;
+an exhausted queue drops a packet for Raft to retransmit and never blocks the
+group owner.
+
+The default is false. This is a candidate for controlled comparison, not an
+established throughput improvement. It does not make filesystem I/O asynchronous
+or let the owner process timer ticks while a filesystem operation blocks.
+
+`Ready.persisted_messages` still wait for durable HardState, entries and snapshot
+installation. All current sync barriers, commit/application ordering, client
+completion and quorum-read boundaries remain. Snapshots use the existing
+captured-image ownership and reserved completion lane after the poll. The
+synchronous public `Node::poll` retains its existing behavior.
+
+With the option enabled, safe replication packets can already be in flight when
+the leader's local persistence fails. The node still fences its shared engine
+and completes outstanding callers with an uncertain outcome, never a successful
+ACK from that failed poll. Another replica can retain the original proposal;
+recovery must reconcile the same producer identity instead of assuming rejection.
+
+Qualification must demonstrate a follower receiving an append while the leader's
+real sync barrier is blocked, with no client ACK before release; injected leader
+and follower sync failures; bounded outgoing credits; snapshot catch-up; and
+complete payload/exact-retry recovery after reopening the original directories.
+Independent-process recovery and a same-binary performance comparison remain
+necessary before recommending this setting.