Feature: leader replicates without waiting for its own log flush (0.9)

openraft/src/core/notify.rs

@@ -1,8 +1,10 @@
 use crate::core::sm;
 use crate::raft::VoteResponse;
 use crate::replication;
+use crate::LogId;
 use crate::MessageSummary;
 use crate::RaftTypeConfig;
+use crate::StorageError;
 use crate::Vote;
 
 /// A message coming from the internal components.
@@ -47,6 +49,17 @@ where C: RaftTypeConfig
         /// ith tick
         i: u64,
     },
+
+    /// An append this node submitted as leader, without waiting for it,
+    /// is flushed (or failed). Sent in the order the appends were
+    /// submitted.
+    LocalLogFlushed {
+        /// The leader vote the append was submitted under.
+        vote: Vote<C::NodeId>,
+        /// The last log id of the append.
+        last_log_id: LogId<C::NodeId>,
+        result: Result<(), StorageError<C::NodeId>>,
+    },
 }
 
 impl<C> Notify<C>
@@ -88,6 +101,18 @@ where C: RaftTypeConfig
             Self::Tick { i } => {
                 format!("Tick {}", i)
             }
+            Self::LocalLogFlushed {
+                vote,
+                last_log_id,
+                result,
+            } => {
+                format!(
+                    "LocalLogFlushed: vote: {}, last_log_id: {}, ok: {}",
+                    vote,
+                    last_log_id,
+                    result.is_ok()
+                )
+            }
         }
     }
 }

openraft/src/core/raft_core.rs

@@ -89,6 +89,7 @@ use crate::storage::RaftLogReaderExt;
 use crate::storage::RaftLogStorage;
 use crate::storage::RaftStateMachine;
 use crate::type_config::alias::InstantOf;
+use crate::type_config::alias::OneshotReceiverOf;
 use crate::type_config::alias::ResponderOf;
 use crate::type_config::TypeConfigExt;
 use crate::AsyncRuntime;
@@ -105,6 +106,68 @@ use crate::StorageError;
 use crate::StorageIOError;
 use crate::Vote;
 
+/// The most appends a leader may have submitted but not yet seen
+/// flushed. Past it, the next append waits for its flush, so a stuck disk
+/// stops the leader instead of queueing entries without bound.
+pub(crate) const MAX_UNFLUSHED_LEADER_APPENDS: usize = 16;
+
+/// A leader's appends submitted without waiting for their flush, and
+/// the task that reports the flushes back to RaftCore, one by one in
+/// submission order.
+///
+/// A flush outstanding for longer than a heartbeat interval is a
+/// stalled disk: RaftCore then waits for its flushes before it sends
+/// anything to a follower (heartbeats included), as it does when every
+/// append is awaited, so the followers elect another leader
+/// instead of following one that cannot make anything durable.
+pub(crate) struct LeaderFlush<C: RaftTypeConfig> {
+    permits: Arc<tokio::sync::Semaphore>,
+    tx: Option<mpsc::UnboundedSender<PendingFlush<C>>>,
+    /// Appends submitted so far.
+    submitted: u64,
+    /// Submission number and time of the appends not known to be flushed.
+    times: std::collections::VecDeque<(u64, std::time::Instant)>,
+    /// Appends flushed so far (or failed), counted by the forwarder.
+    flushed_tx: Arc<watch::Sender<u64>>,
+    flushed_rx: watch::Receiver<u64>,
+    /// Whether the current stall was logged.
+    stall_logged: bool,
+}
+
+impl<C: RaftTypeConfig> Default for LeaderFlush<C> {
+    fn default() -> Self {
+        let (flushed_tx, flushed_rx) = watch::channel(0);
+        Self {
+            permits: Arc::new(tokio::sync::Semaphore::new(MAX_UNFLUSHED_LEADER_APPENDS)),
+            tx: None,
+            submitted: 0,
+            times: Default::default(),
+            flushed_tx: Arc::new(flushed_tx),
+            flushed_rx,
+            stall_logged: false,
+        }
+    }
+}
+
+impl<C: RaftTypeConfig> LeaderFlush<C> {
+    /// Whether the oldest unflushed append was submitted `after` ago or
+    /// longer.
+    fn stalled(&mut self, after: Duration) -> bool {
+        let done = *self.flushed_rx.borrow();
+        while self.times.front().is_some_and(|(seq, _)| *seq < done) {
+            self.times.pop_front();
+        }
+        self.times.front().is_some_and(|(_, t)| t.elapsed() >= after)
+    }
+}
+
+struct PendingFlush<C: RaftTypeConfig> {
+    vote: Vote<C::NodeId>,
+    last_log_id: LogId<C::NodeId>,
+    rx: OneshotReceiverOf<C, Result<LogIOId<C::NodeId>, std::io::Error>>,
+    _permit: tokio::sync::OwnedSemaphorePermit,
+}
+
 /// A temp struct to hold the data for a node that is being applied.
 #[derive(Debug)]
 pub(crate) struct ApplyingEntry<NID: NodeId, N: Node> {
@@ -206,6 +269,9 @@ where
 
     pub(crate) command_state: CommandState,
 
+    /// Leader appends submitted without waiting for their flush.
+    pub(crate) leader_flush: LeaderFlush<C>,
+
     pub(crate) span: Span,
 
     pub(crate) _p: PhantomData<SM>,
@@ -730,6 +796,103 @@ where
         Ok(())
     }
 
+    /// Append as leader without waiting for the flush, so the `Replicate` commands that follow go
+    /// out at once and the followers' fsyncs overlap the leader's. The flush is
+    /// reported later as [`Notify::LocalLogFlushed`], and only then does
+    /// the leader count itself in the commit quorum, so a commit still
+    /// needs a quorum of durable copies.
+    #[tracing::instrument(level = "debug", skip_all)]
+    pub(crate) async fn submit_leader_append<I>(
+        &mut self,
+        entries: I,
+        vote: Vote<C::NodeId>,
+        last_log_id: LogId<C::NodeId>,
+    ) -> Result<(), StorageError<C::NodeId>>
+    where
+        I: IntoIterator<Item = C::Entry> + OptionalSend,
+        I::IntoIter: OptionalSend,
+    {
+        let permit = self
+            .leader_flush
+            .permits
+            .clone()
+            .acquire_owned()
+            .await
+            .map_err(|e| StorageIOError::write_logs(AnyError::error(e)))?;
+
+        let (tx, rx) = C::AsyncRuntime::oneshot();
+        let log_io_id = LogIOId::new(vote.clone(), Some(last_log_id.clone()));
+        let callback = LogFlushed::new(log_io_id, tx);
+        self.log_store.append(entries, callback).await?;
+
+        let pending = PendingFlush {
+            vote,
+            last_log_id,
+            rx,
+            _permit: permit,
+        };
+        self.leader_flush.times.push_back((self.leader_flush.submitted, std::time::Instant::now()));
+        self.leader_flush.submitted += 1;
+        let pending = match &self.leader_flush.tx {
+            Some(fwd) => match fwd.send(pending) {
+                Ok(()) => return Ok(()),
+                Err(e) => e.0,
+            },
+            None => pending,
+        };
+        // Start the forwarder: it waits for each flush in submission
+        // order and tells RaftCore, so the leader's own progress only
+        // ever moves forward.
+        let (fwd, mut rx_pending) = mpsc::unbounded_channel::<PendingFlush<C>>();
+        let tx_notify = self.tx_notify.clone();
+        let flushed = self.leader_flush.flushed_tx.clone();
+        #[allow(clippy::let_underscore_future)]
+        let _ = C::spawn(async move {
+            while let Some(p) = rx_pending.recv().await {
+                let result = match p.rx.await {
+                    Ok(Ok(_)) => Ok(()),
+                    Ok(Err(e)) => Err(StorageIOError::write_logs(AnyError::error(e)).into()),
+                    Err(e) => Err(StorageIOError::write_logs(AnyError::error(e)).into()),
+                };
+                // Counted even when it failed: RaftCore may be waiting for
+                // it (`wait_if_leader_flush_stalled`), and stops on the error.
+                flushed.send_modify(|n| *n += 1);
+                let sent = tx_notify.send(Notify::LocalLogFlushed {
+                    vote: p.vote,
+                    last_log_id: p.last_log_id,
+                    result,
+                });
+                drop(p._permit);
+                if sent.is_err() {
+                    break;
+                }
+            }
+        });
+        let _ = fwd.send(pending);
+        self.leader_flush.tx = Some(fwd);
+        Ok(())
+    }
+
+    /// If a leader append has been waiting for its flush for a heartbeat interval or longer, wait
+    /// for every submitted append's flush (see [`LeaderFlush`]).
+    async fn wait_if_leader_flush_stalled(&mut self) {
+        let after = Duration::from_millis(self.config.heartbeat_interval);
+        if !self.leader_flush.stalled(after) {
+            self.leader_flush.stall_logged = false;
+            return;
+        }
+        if !self.leader_flush.stall_logged {
+            tracing::warn!(
+                "the leader's log flush has taken {:?} or more: holding replication and heartbeats until it completes",
+                after
+            );
+            self.leader_flush.stall_logged = true;
+        }
+        let target = self.leader_flush.submitted;
+        let mut rx = self.leader_flush.flushed_rx.clone();
+        let _ = rx.wait_for(|n| *n >= target).await;
+    }
+
     #[tracing::instrument(level = "debug", skip_all)]
     pub(crate) async fn apply_to_state_machine(
         &mut self,
@@ -1327,6 +1490,24 @@ where
                 }
             }
 
+            Notify::LocalLogFlushed {
+                vote,
+                last_log_id,
+                result,
+            } => {
+                if let Err(e) = result {
+                    tracing::error!(error = display(&e), "leader log flush failed: {}", func_name!());
+                    return Err(Fatal::from(e));
+                }
+                // A flush from an earlier leadership is not progress of
+                // this one.
+                if self.does_vote_match(&vote, "LocalLogFlushed") {
+                    if let Ok(mut lh) = self.engine.leader_handler() {
+                        lh.replication_handler().update_local_progress(Some(last_log_id));
+                    }
+                }
+            }
+
             Notify::Network { response } => {
                 //
                 match response {
@@ -1637,12 +1818,20 @@ where
                 let last_log_id = entries.last().unwrap().get_log_id().clone();
                 tracing::debug!("AppendInputEntries: {}", DisplaySlice::<_>(&entries),);
 
-                self.append_to_log(entries, vote, last_log_id.clone()).await?;
-
-                // The leader may have changed.
-                // But reporting to a different leader is not a problem.
-                if let Ok(mut lh) = self.engine.leader_handler() {
-                    lh.replication_handler().update_local_progress(Some(last_log_id));
+                // This node's own append as leader does not wait for
+                // its flush; a follower's still
+                // does (it answers AppendEntries only after it).
+                let own_leader_append = self.engine.leader.as_ref().is_some_and(|l| l.vote == vote);
+                if own_leader_append {
+                    self.submit_leader_append(entries, vote, last_log_id).await?;
+                } else {
+                    self.append_to_log(entries, vote, last_log_id.clone()).await?;
+
+                    // The leader may have changed.
+                    // But reporting to a different leader is not a problem.
+                    if let Ok(mut lh) = self.engine.leader_handler() {
+                        lh.replication_handler().update_local_progress(Some(last_log_id));
+                    }
                 }
             }
             Command::SaveVote { vote } => {
@@ -1700,6 +1889,7 @@ where
                 self.apply_to_state_machine(seq, already_committed.next_index(), upto.index).await?;
             }
             Command::Replicate { req, target } => {
+                self.wait_if_leader_flush_stalled().await;
                 let node = self.replications.get(&target).expect("replication to target node exists");
 
                 match req {

openraft/src/raft/mod.rs

@@ -299,6 +299,7 @@ where C: RaftTypeConfig
             tx_server_metrics,
 
             command_state: CommandState::default(),
+            leader_flush: Default::default(),
             span: core_span,
 
             _p: Default::default(),