diff --git a/CHANGELOG.md b/CHANGELOG.md
index 6015118f..d312d1eb 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -8,12 +8,9 @@ All notable changes to this project will be documented in this file.
 
 ### Added
 
-- **ReadActor fast path for Eventual/LeaseRead** (#392): Dedicated `ReadActor` task serves `EventualConsistency` and `LeaseRead` without entering the Raft loop, eliminating channel contention under high read concurrency.
-  - `Arc<SM>` is owned exclusively by ReadActor — guarantees RocksDB LOCK release before Raft shutdown on `stop()`, fixing `test_snapshot_recovery_embedded`
-  - `ReadLease.revoke()`: atomic lease invalidation on leader demotion (replaces `invalidate()`)
-  - `EmbeddedClient` routes Eventual/Lease → ReadActor; falls back to `cmd_tx` on `LeaseInvalid`/`SmStopped`
-  - New `[raft.read_actor]` config section: `channel_capacity` (default 512), `max_drain` (default 100)
-  - **Benchmark (local embedded, 100 concurrent clients)**: Lease Read +10.9%, Eventual Read +13.1% vs v0.2.4; write throughput unchanged (see `benches/reports/v0.2.5/`)
+- **AppendEntries coalescing + speculative replication** (#407): Consecutive same-term `AppendEntries` in the receive buffer are merged before dispatch (heartbeats absorbed via `max(leader_commit_index)`); leader advances `next_index` speculatively on send, eliminating stop-and-wait per replication round.
+
+- **ReadActor fast path for Eventual/LeaseRead** (#392): Dedicated `ReadActor` serves `EventualConsistency` and `LeaseRead` off the Raft loop, eliminating channel contention under high read concurrency. Lease Read +10.9%, Eventual Read +13.1% vs v0.2.4 (100 concurrent clients, local embedded).
 
 ### Fixed
 
diff --git a/benches/standalone-bench/src/main.rs b/benches/standalone-bench/src/main.rs
index 7cb07661..f2b9b20a 100644
--- a/benches/standalone-bench/src/main.rs
+++ b/benches/standalone-bench/src/main.rs
@@ -1,3 +1,10 @@
+use std::sync::Arc;
+use std::sync::Mutex;
+use std::sync::atomic::AtomicU64;
+use std::sync::atomic::Ordering;
+use std::time::Duration;
+use std::time::Instant;
+
 use clap::Parser;
 use clap::Subcommand;
 use d_engine::Client;
@@ -9,12 +16,6 @@ use hdrhistogram::Histogram;
 use rand::Rng;
 use rand::SeedableRng;
 use rand::distributions::Alphanumeric;
-use std::sync::Arc;
-use std::sync::Mutex;
-use std::sync::atomic::AtomicU64;
-use std::sync::atomic::Ordering;
-use std::time::Duration;
-use std::time::Instant;
 use tokio::sync::Semaphore;
 
 #[derive(Parser, Debug, Clone)]
@@ -209,7 +210,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
 
     // Initialize the connection pool
     let endpoints = cli.endpoints.clone();
-    println!("Initializing client connection with: {:?}", &endpoints);
+    println!("Initializing client connection with: {:?}", endpoints);
     let client_pool = ClientPool::new(endpoints, cli.conns)
         .await
         .expect("Failed to create client pool");
diff --git a/d-engine-core/src/config/raft.rs b/d-engine-core/src/config/raft.rs
index 03a826c5..aad2cc64 100644
--- a/d-engine-core/src/config/raft.rs
+++ b/d-engine-core/src/config/raft.rs
@@ -333,12 +333,19 @@ pub struct BatchingConfig {
     /// **Default**: 100
     #[serde(default = "default_max_batch_size")]
     pub max_batch_size: usize,
+
+    /// Maximum total entries to accumulate in a single merge_append_entries pass.
+    /// Prevents unbounded memory growth when the buffer has many large batches.
+    /// Default: 1000
+    #[serde(default = "default_max_merge_entries")]
+    pub max_merge_entries: usize,
 }
 
 impl Default for BatchingConfig {
     fn default() -> Self {
         Self {
             max_batch_size: default_max_batch_size(),
+            max_merge_entries: default_max_merge_entries(),
         }
     }
 }
@@ -350,6 +357,11 @@ impl BatchingConfig {
                 "batching.max_batch_size must be > 0".into(),
             )));
         }
+        if self.max_merge_entries == 0 {
+            return Err(Error::Config(ConfigError::Message(
+                "batching.max_merge_entries must be > 0".into(),
+            )));
+        }
         Ok(())
     }
 }
@@ -360,6 +372,11 @@ fn default_append_interval() -> u64 {
 fn default_max_batch_size() -> usize {
     100
 }
+
+fn default_max_merge_entries() -> usize {
+    1000
+}
+
 fn default_entries_per_replication() -> u64 {
     100
 }
diff --git a/d-engine-core/src/event.rs b/d-engine-core/src/event.rs
index 24c0f2d3..80e8291f 100644
--- a/d-engine-core/src/event.rs
+++ b/d-engine-core/src/event.rs
@@ -146,7 +146,7 @@ pub enum RaftEvent {
 
     AppendEntries(
         AppendEntriesRequest,
-        MaybeCloneOneshotSender<std::result::Result<AppendEntriesResponse, Status>>,
+        Vec<MaybeCloneOneshotSender<std::result::Result<AppendEntriesResponse, Status>>>,
     ),
 
     // Response snapshot stream from Leader
@@ -193,6 +193,11 @@ pub enum RaftEvent {
     },
 }
 
+// SAFETY: tonic::Streaming variants are only accessed on the thread that created them;
+// no shared references to snapshot variants cross thread boundaries.
+// Full decoupling tracked in #409.
+unsafe impl Sync for RaftEvent {}
+
 #[cfg(test)]
 #[cfg_attr(test, derive(Debug, Clone))]
 #[allow(unused)]
diff --git a/d-engine-core/src/raft.rs b/d-engine-core/src/raft.rs
index 6fd1b1f1..6af9c249 100644
--- a/d-engine-core/src/raft.rs
+++ b/d-engine-core/src/raft.rs
@@ -27,6 +27,7 @@ use crate::Result;
 use crate::TypeConfig;
 use crate::alias::MOF;
 use crate::alias::TROF;
+use std::collections::VecDeque;
 use std::sync::Arc;
 
 pub struct Raft<T>
@@ -40,6 +41,7 @@ where
     // Network & Storage events
     event_tx: mpsc::Sender<RaftEvent>,
     event_rx: mpsc::Receiver<RaftEvent>,
+    buffered_raft_event: VecDeque<RaftEvent>,
 
     // Client commands (drain-driven)
     cmd_tx: mpsc::Sender<super::ClientCmd>,
@@ -48,6 +50,7 @@ where
     // Timer
     role_tx: mpsc::UnboundedSender<RoleEvent>,
     role_rx: mpsc::UnboundedReceiver<RoleEvent>,
+    buffered_role_event: VecDeque<RoleEvent>,
 
     // For business logic to apply logs into state machine
     new_commit_listener: Vec<mpsc::UnboundedSender<NewCommitData>>,
@@ -134,12 +137,14 @@ where
 
             event_tx: signal_params.event_tx,
             event_rx: signal_params.event_rx,
+            buffered_raft_event: VecDeque::new(),
 
             cmd_tx: signal_params.cmd_tx,
             cmd_rx: signal_params.cmd_rx,
 
             role_tx: signal_params.role_tx,
             role_rx: signal_params.role_rx,
+            buffered_role_event: VecDeque::new(),
 
             new_commit_listener: Vec::new(),
 
@@ -259,6 +264,7 @@ where
             tokio::select! {
                 // Use biased to ensure branch order
                 biased;
+
                 // P0: shutdown received;
                 _ = self.shutdown_signal.changed() => {
                     info!("[Raft:{}] shutdown signal received.", self.node_id);
@@ -271,9 +277,9 @@ where
                     self.event_rx.close();
                     return Ok(());
                 }
+
                 // P1: Tick: start Heartbeat(replication) or start Election
                 _ = tick => {
-
                     trace!("receive tick");
                     let role_tx = &self.role_tx;
                     let event_tx = &self.event_tx;
@@ -288,35 +294,28 @@ where
                 // P2: Role events — handle first, drain rest after select
                 Some(role_event) = self.role_rx.recv() => {
                     debug!(%self.node_id, ?role_event, "receive role event");
+                    self.buffered_role_event.push_back(role_event);
+                    self.drain_role_events().await?;
+
 
-                    if let Err(e) = self.handle_role_event(role_event).await {
-                        error!(%self.node_id, ?e, "handle_role_event error");
-                    }
                 }
 
                 // P3: Client commands — push first, drain rest after select
                 Some(first_cmd) = self.cmd_rx.recv() => {
                     trace!(%self.node_id, "receive first client command");
                     self.role.push_client_cmd(first_cmd, &self.ctx);
+                    self.drain_client_cmds().await?;
                 }
 
                 // P4: Other events — handle first, drain rest after select
                 Some(raft_event) = self.event_rx.recv() => {
                     trace!(%self.node_id, ?raft_event, "receive raft event");
+                    self.buffered_raft_event.push_back(raft_event);
+                    self.drain_raft_events().await?;
+
 
-                    #[cfg(test)]
-                    let event = raft_event_to_test_event(&raft_event);
 
-                    if let Err(e) = self.role.handle_raft_event(raft_event, &self.ctx, self.role_tx.clone()).await {
-                        if e.is_fatal() {
-                            error!(%self.node_id, ?e, "Fatal error in handle_raft_event, shutting down");
-                            return Err(e);
-                        }
-                        warn!(%self.node_id, ?e, "Non-fatal error in handle_raft_event, continuing");
-                    }
 
-                    #[cfg(test)]
-                    self.notify_raft_event(event);
                 }
 
             }
@@ -326,15 +325,31 @@ where
             // naturally yielding at .await points so client tasks can enqueue their
             // next writes into cmd_rx before drain_client_cmds runs.
             tokio::task::yield_now().await;
-            self.drain_role_events().await?;
+            self.process_role_events().await?;
             // Yield after role events so woken client tasks can enqueue their next
             // writes into cmd_rx before drain_client_cmds runs.
             tokio::task::yield_now().await;
-            self.drain_client_cmds().await?;
-            self.drain_raft_events().await?;
+            self.process_client_cmds().await?;
+            self.process_raft_events().await?;
         }
     }
 
+    /// Drain all pending raft events (up to max_batch_size).
+    async fn drain_raft_events(&mut self) -> Result<()> {
+        let max = self.ctx.node_config.raft.batching.max_batch_size;
+        let mut count = 0;
+        while count < max {
+            match self.event_rx.try_recv() {
+                Ok(raft_event) => {
+                    self.buffered_raft_event.push_back(raft_event);
+                    count += 1;
+                }
+                Err(_) => break,
+            }
+        }
+        Ok(())
+    }
+
     /// Drain all pending role events (up to max_batch_size).
     async fn drain_role_events(&mut self) -> Result<()> {
         let max = self.ctx.node_config.raft.batching.max_batch_size;
@@ -342,9 +357,7 @@ where
         while count < max {
             match self.role_rx.try_recv() {
                 Ok(role_event) => {
-                    if let Err(e) = self.handle_role_event(role_event).await {
-                        error!(%self.node_id, ?e, "drain_role_events: handle_role_event error");
-                    }
+                    self.buffered_role_event.push_back(role_event);
                     count += 1;
                 }
                 Err(_) => break,
@@ -369,39 +382,107 @@ where
         if count > 0 {
             trace!("Drained {} client commands", count);
         }
-        // Always flush: the P3 select arm may have pushed a command before this drain ran.
-        // flush_cmd_buffers checks is_empty() internally and is a no-op when nothing is buffered.
-        if let Err(e) = self.role.flush_cmd_buffers(&self.ctx, &self.role_tx).await {
-            error!(%self.node_id, ?e, "drain_client_cmds: flush_cmd_buffers error");
-            return Err(e);
+        Ok(())
+    }
+
+    async fn process_role_events(&mut self) -> Result<()> {
+        while let Some(event) = self.buffered_role_event.pop_front() {
+            if let Err(e) = self.handle_role_event(event).await {
+                if e.is_fatal() {
+                    error!(%self.node_id, ?e, "Fatal error in process_role_events, shutting down");
+                    return Err(e);
+                }
+                warn!(%self.node_id, ?e, "Non-fatal error in process_role_events, continuing");
+            }
         }
         Ok(())
     }
 
+    async fn process_raft_events(&mut self) -> Result<()> {
+        while !self.buffered_raft_event.is_empty() {
+            // Avoid none AE event pop and push into queue
+            if matches!(
+                self.buffered_raft_event.front(),
+                Some(RaftEvent::AppendEntries(..))
+            ) {
+                self.merge_append_entries();
+            }
+
+            let Some(event) = self.buffered_raft_event.pop_front() else {
+                break;
+            };
+
+            #[cfg(test)]
+            let test_event = raft_event_to_test_event(&event);
+
+            if let Err(e) =
+                self.role.handle_raft_event(event, &self.ctx, self.role_tx.clone()).await
+            {
+                if e.is_fatal() {
+                    error!(%self.node_id, ?e, "Fatal error in drain_raft_events, shutting down");
+                    return Err(e);
+                }
+                warn!(%self.node_id, ?e, "Non-fatal error in drain_raft_events, continuing");
+            }
+
+            #[cfg(test)]
+            self.notify_raft_event(test_event);
+        }
+        Ok(())
+    }
+
+    async fn process_client_cmds(&mut self) -> Result<()> {
+        // Always flush: the P3 select arm may have pushed a command before this drain ran.
+        // flush_cmd_buffers checks is_empty() internally and is a no-op when nothing is buffered.
+        self.role.flush_cmd_buffers(&self.ctx, &self.role_tx).await
+    }
+
     /// Drain all pending raft events (up to max_batch_size).
-    async fn drain_raft_events(&mut self) -> Result<()> {
-        let max = self.ctx.node_config.raft.batching.max_batch_size;
-        let mut count = 0;
-        while count < max {
-            match self.event_rx.try_recv() {
-                Ok(raft_event) => {
-                    if let Err(e) = self
-                        .role
-                        .handle_raft_event(raft_event, &self.ctx, self.role_tx.clone())
-                        .await
-                    {
-                        if e.is_fatal() {
-                            error!(%self.node_id, ?e, "Fatal error in drain_raft_events, shutting down");
-                            return Err(e);
-                        }
-                        warn!(%self.node_id, ?e, "Non-fatal error in drain_raft_events, continuing");
+    fn merge_append_entries(&mut self) {
+        let Some(event) = self.buffered_raft_event.pop_front() else {
+            return;
+        };
+
+        let RaftEvent::AppendEntries(mut first_req, first_senders) = event else {
+            self.buffered_raft_event.push_front(event);
+            return;
+        };
+
+        let max = self.ctx.node_config.raft.batching.max_merge_entries;
+
+        let mut next_prev = first_req.prev_log_index + first_req.entries.len() as u64;
+        let mut merged_entries = first_req.entries;
+        let term = first_req.term;
+        let mut merged_senders = first_senders;
+
+        while let Some(event) = self.buffered_raft_event.pop_front() {
+            match event {
+                RaftEvent::AppendEntries(mut req, mut senders)
+                    if next_prev == req.prev_log_index && term == req.term =>
+                {
+                    if merged_entries.len() + req.entries.len() > max {
+                        self.buffered_raft_event.push_front(RaftEvent::AppendEntries(req, senders));
+                        break;
                     }
-                    count += 1;
+
+                    // Take max(leader_commit_index) to handle two cases:
+                    // case 1: heartbeat (empty entries) may carry a higher commit index
+                    // case 2: leader's commit index may advance between consecutive AE sends
+                    first_req.leader_commit_index =
+                        first_req.leader_commit_index.max(req.leader_commit_index);
+                    next_prev += req.entries.len() as u64;
+                    merged_entries.append(&mut req.entries);
+                    merged_senders.append(&mut senders);
+                }
+                _ => {
+                    self.buffered_raft_event.push_front(event);
+                    break;
                 }
-                Err(_) => break,
             }
         }
-        Ok(())
+        first_req.entries = merged_entries;
+        self.buffered_raft_event
+            .push_front(RaftEvent::AppendEntries(first_req, merged_senders));
     }
 
     /// `handle_role_event` will be responsbile to process role trasnsition and
@@ -751,3 +832,19 @@ where
         info!("Graceful shutdown node state ...");
     }
 }
+
+#[cfg(test)]
+#[path = "raft_test/leader_change_tests.rs"]
+mod leader_change_tests;
+#[cfg(test)]
+#[path = "raft_test/leader_discovered_tests.rs"]
+mod leader_discovered_tests;
+#[cfg(test)]
+#[path = "raft_test/merge_append_entries_tests.rs"]
+mod merge_append_entries_tests;
+#[cfg(test)]
+#[path = "raft_test/process_raft_events_tests.rs"]
+mod process_raft_events_tests;
+#[cfg(test)]
+#[path = "raft_test/raft_comprehensive_tests.rs"]
+mod raft_comprehensive_tests;
diff --git a/d-engine-core/src/raft_role/candidate_state.rs b/d-engine-core/src/raft_role/candidate_state.rs
index 948dfdec..3b87ecd5 100644
--- a/d-engine-core/src/raft_role/candidate_state.rs
+++ b/d-engine-core/src/raft_role/candidate_state.rs
@@ -319,7 +319,7 @@ impl<T: TypeConfig> RaftRoleState for CandidateState<T> {
                 }
             }
 
-            RaftEvent::AppendEntries(append_entries_request, sender) => {
+            RaftEvent::AppendEntries(append_entries_request, senders) => {
                 debug!(
                     "handle_raft_event::RaftEvent::AppendEntries: {:?}",
                     &append_entries_request
@@ -349,7 +349,7 @@ impl<T: TypeConfig> RaftRoleState for CandidateState<T> {
                     );
                     self.send_replay_raft_event(
                         &role_tx,
-                        RaftEvent::AppendEntries(append_entries_request, sender),
+                        RaftEvent::AppendEntries(append_entries_request, senders),
                     )?;
                 } else {
                     // request.term < my_term: stale leader, reject.
@@ -359,12 +359,14 @@ impl<T: TypeConfig> RaftRoleState for CandidateState<T> {
                         ctx.raft_log(),
                     );
                     debug!("Rejecting AppendEntries from stale leader: {:?}", &response);
-                    if let Err(e) = sender.send(Ok(response)) {
-                        // Receiver timed out and dropped — this is normal, do not crash the node
-                        error!(
-                            "Failed to send AppendEntries rejection (receiver dropped): {:?}",
-                            e
-                        );
+                    for sender in senders {
+                        if let Err(e) = sender.send(Ok(response)) {
+                            // Receiver timed out and dropped — this is normal, do not crash the node
+                            error!(
+                                "Failed to send AppendEntries rejection (receiver dropped): {:?}",
+                                e
+                            );
+                        }
                     }
                 }
             }
diff --git a/d-engine-core/src/raft_role/candidate_state_test.rs b/d-engine-core/src/raft_role/candidate_state_test.rs
index a0dcc1eb..497f578a 100644
--- a/d-engine-core/src/raft_role/candidate_state_test.rs
+++ b/d-engine-core/src/raft_role/candidate_state_test.rs
@@ -571,7 +571,7 @@ async fn test_handle_append_entries_steps_down_to_follower() {
         leader_commit_index: new_leader_commit,
     };
     let (resp_tx, mut resp_rx) = MaybeCloneOneshot::new();
-    let raft_event = RaftEvent::AppendEntries(append_entries_request, resp_tx);
+    let raft_event = RaftEvent::AppendEntries(append_entries_request, vec![resp_tx]);
     let (role_tx, mut role_rx) = mpsc::unbounded_channel();
 
     assert!(state.handle_raft_event(raft_event, &context, role_tx).await.is_ok());
@@ -619,7 +619,7 @@ async fn test_handle_append_entries_rejects_lower_term() {
         leader_commit_index: 0,
     };
     let (resp_tx, mut resp_rx) = MaybeCloneOneshot::new();
-    let raft_event = RaftEvent::AppendEntries(append_entries_request, resp_tx);
+    let raft_event = RaftEvent::AppendEntries(append_entries_request, vec![resp_tx]);
     let (role_tx, mut role_rx) = mpsc::unbounded_channel();
 
     assert!(state.handle_raft_event(raft_event, &context, role_tx).await.is_ok());
@@ -673,7 +673,7 @@ async fn test_handle_append_entries_rejects_stale_leader() {
         leader_commit_index: 0,
     };
     let (resp_tx, mut resp_rx) = MaybeCloneOneshot::new();
-    let raft_event = RaftEvent::AppendEntries(append_entries_request, resp_tx);
+    let raft_event = RaftEvent::AppendEntries(append_entries_request, vec![resp_tx]);
     let (role_tx, mut role_rx) = mpsc::unbounded_channel();
 
     assert!(state.handle_raft_event(raft_event, &context, role_tx).await.is_ok());
@@ -736,7 +736,7 @@ async fn test_handle_append_entries_same_term_log_conflict_steps_down() {
         leader_commit_index: 10,
     };
     let (resp_tx, mut resp_rx) = MaybeCloneOneshot::new();
-    let raft_event = RaftEvent::AppendEntries(append_entries_request, resp_tx);
+    let raft_event = RaftEvent::AppendEntries(append_entries_request, vec![resp_tx]);
     let (role_tx, mut role_rx) = mpsc::unbounded_channel();
 
     assert!(state.handle_raft_event(raft_event, &context, role_tx).await.is_ok());
@@ -807,7 +807,7 @@ async fn test_handle_append_entries_higher_term_log_conflict_steps_down_and_upda
         leader_commit_index: 10,
     };
     let (resp_tx, mut resp_rx) = MaybeCloneOneshot::new();
-    let raft_event = RaftEvent::AppendEntries(append_entries_request, resp_tx);
+    let raft_event = RaftEvent::AppendEntries(append_entries_request, vec![resp_tx]);
     let (role_tx, mut role_rx) = mpsc::unbounded_channel();
 
     assert!(state.handle_raft_event(raft_event, &context, role_tx).await.is_ok());
diff --git a/d-engine-core/src/raft_role/follower_state_test.rs b/d-engine-core/src/raft_role/follower_state_test.rs
index 41d686b9..95d274ce 100644
--- a/d-engine-core/src/raft_role/follower_state_test.rs
+++ b/d-engine-core/src/raft_role/follower_state_test.rs
@@ -556,7 +556,7 @@ async fn test_handle_append_entries_success_from_new_leader() {
         leader_commit_index: 0,
     };
     let (resp_tx, mut resp_rx) = MaybeCloneOneshot::new();
-    let raft_event = RaftEvent::AppendEntries(append_entries_request, resp_tx);
+    let raft_event = RaftEvent::AppendEntries(append_entries_request, vec![resp_tx]);
 
     let (role_tx, mut role_rx) = mpsc::unbounded_channel();
 
@@ -653,7 +653,7 @@ async fn test_handle_append_entries_rejects_stale_term() {
         leader_commit_index: 0,
     };
     let (resp_tx, mut resp_rx) = MaybeCloneOneshot::new();
-    let raft_event = RaftEvent::AppendEntries(append_entries_request, resp_tx);
+    let raft_event = RaftEvent::AppendEntries(append_entries_request, vec![resp_tx]);
 
     let (role_tx, mut role_rx) = mpsc::unbounded_channel();
 
@@ -730,7 +730,7 @@ async fn test_handle_append_entries_with_handler_error() {
         leader_commit_index: 0,
     };
     let (resp_tx, mut resp_rx) = MaybeCloneOneshot::new();
-    let raft_event = RaftEvent::AppendEntries(append_entries_request, resp_tx);
+    let raft_event = RaftEvent::AppendEntries(append_entries_request, vec![resp_tx]);
 
     let (role_tx, mut role_rx) = mpsc::unbounded_channel();
 
@@ -2409,7 +2409,7 @@ async fn test_follower_acks_immediately_after_memory_write() {
         leader_commit_index: 0,
     };
     let (resp_tx, mut resp_rx) = MaybeCloneOneshot::new();
-    let raft_event = RaftEvent::AppendEntries(append_request, resp_tx);
+    let raft_event = RaftEvent::AppendEntries(append_request, vec![resp_tx]);
     let (role_tx, _role_rx) = mpsc::unbounded_channel();
 
     assert!(state.handle_raft_event(raft_event, &context, role_tx).await.is_ok());
@@ -2450,7 +2450,7 @@ async fn test_follower_acks_immediately_for_heartbeat() {
         leader_commit_index: 0,
     };
     let (resp_tx, mut resp_rx) = MaybeCloneOneshot::new();
-    let raft_event = RaftEvent::AppendEntries(append_request, resp_tx);
+    let raft_event = RaftEvent::AppendEntries(append_request, vec![resp_tx]);
     let (role_tx, _role_rx) = mpsc::unbounded_channel();
 
     assert!(state.handle_raft_event(raft_event, &context, role_tx).await.is_ok());
@@ -2499,7 +2499,7 @@ async fn test_follower_commit_index_and_ack_both_sent_immediately() {
         leader_commit_index: new_commit,
     };
     let (resp_tx, mut resp_rx) = MaybeCloneOneshot::new();
-    let raft_event = RaftEvent::AppendEntries(append_request, resp_tx);
+    let raft_event = RaftEvent::AppendEntries(append_request, vec![resp_tx]);
     let (role_tx, _role_rx) = mpsc::unbounded_channel();
 
     assert!(state.handle_raft_event(raft_event, &context, role_tx).await.is_ok());
diff --git a/d-engine-core/src/raft_role/leader_state.rs b/d-engine-core/src/raft_role/leader_state.rs
index 10e7ec90..923c0e2c 100644
--- a/d-engine-core/src/raft_role/leader_state.rs
+++ b/d-engine-core/src/raft_role/leader_state.rs
@@ -1211,7 +1211,7 @@ impl<T: TypeConfig> RaftRoleState for LeaderState<T> {
                 }
             }
 
-            RaftEvent::AppendEntries(append_entries_request, sender) => {
+            RaftEvent::AppendEntries(append_entries_request, senders) => {
                 debug!(
                     "handle_raft_event::RaftEvent::AppendEntries: {:?}",
                     &append_entries_request
@@ -1221,11 +1221,11 @@ impl<T: TypeConfig> RaftRoleState for LeaderState<T> {
                 if my_term >= append_entries_request.term {
                     let response = AppendEntriesResponse::higher_term(my_id, my_term);
 
-                    sender.send(Ok(response)).map_err(|e| {
-                        let error_str = format!("{e:?}");
-                        error!("Failed to send: {}", error_str);
-                        NetworkError::SingalSendFailed(error_str)
-                    })?;
+                    for sender in senders {
+                        if let Err(e) = sender.send(Ok(response)) {
+                            error!("Failed to send: {:?}", e);
+                        }
+                    }
                 } else {
                     // Step down as Follower as new Leader found
                     info!(
@@ -1245,7 +1245,7 @@ impl<T: TypeConfig> RaftRoleState for LeaderState<T> {
                     );
                     send_replay_raft_event(
                         &role_tx,
-                        RaftEvent::AppendEntries(append_entries_request, sender),
+                        RaftEvent::AppendEntries(append_entries_request, senders),
                     )?;
                 }
             }
@@ -1830,8 +1830,7 @@ impl<T: TypeConfig> RaftRoleState for LeaderState<T> {
         };
 
         // Update next_index and match_index for this peer.
-        let peer_updates = HashMap::from([(follower_id, peer_update.clone())]);
-        self.update_peer_indexes(&peer_updates);
+        self.update_peer_index(follower_id, &peer_update);
 
         // Check learner promotion progress.
         if !is_voter {
@@ -2565,26 +2564,48 @@ impl<T: TypeConfig> LeaderState<T> {
         }
     }
 
-    /// Update peer node index
+    /// Updates next_index and match_index for a peer after receiving an AppendEntries response.
+    ///
+    /// Two distinct update strategies based on response type:
+    /// - Success: `next_index = max(current, peer_match + 1)` — never regress a speculative
+    ///   advance; the ACK is a lower-bound confirmation, not a correction.
+    /// - Conflict: `next_index = conflict_hint` (floor-guarded by `match_index + 1`) —
+    ///   the follower explicitly corrects the leader's assumption; retreat is required.
     #[instrument(skip(self))]
-    fn update_peer_indexes(
+    pub(super) fn update_peer_index(
         &mut self,
-        peer_updates: &HashMap<u32, PeerUpdate>,
+        follower_id: u32,
+        update: &PeerUpdate,
     ) {
-        for (peer_id, update) in peer_updates {
-            if let Err(e) = self.update_next_index(*peer_id, update.next_index) {
+        if update.success {
+            // Success: trust speculative advance — never regress next_index below what
+            // the leader has already pipeline-sent. ACK confirms a lower bound only;
+            // the leader may have sent further batches speculatively.
+            let current = self.next_index.get(&follower_id).copied().unwrap_or(1);
+            if let Err(e) = self.update_next_index(follower_id, update.next_index.max(current)) {
+                error!("Failed to update next index: {:?}", e);
+            }
+        } else {
+            // Conflict: follower explicitly corrects our assumption — allow next_index
+            // to retreat to the hint. Floor guard (match_index + 1) inside
+            // update_next_index prevents retreating below confirmed entries.
+            if let Err(e) = self.update_next_index(follower_id, update.next_index) {
                 error!("Failed to update next index: {:?}", e);
             }
+        }
+
+        trace!(
+            "Updated next index for peer {}-{}",
+            follower_id, update.next_index
+        );
+        if let Some(match_index) = update.match_index {
+            if let Err(e) = self.update_match_index(follower_id, match_index) {
+                error!("Failed to update match index: {:?}", e);
+            }
             trace!(
-                "Updated next index for peer {}-{}",
-                peer_id, update.next_index
+                "Updated match index for peer {}-{}",
+                follower_id, match_index
             );
-            if let Some(match_index) = update.match_index {
-                if let Err(e) = self.update_match_index(*peer_id, match_index) {
-                    error!("Failed to update match index: {:?}", e);
-                }
-                trace!("Updated match index for peer {}-{}", peer_id, match_index);
-            }
         }
     }
 
@@ -3536,6 +3557,14 @@ impl<T: TypeConfig> LeaderState<T> {
 
         // Phase 5: fire AppendEntries requests to per-follower workers (non-blocking).
         for (peer_id, request) in requests.append_requests {
+            let speculative_next_index = request.prev_log_index + request.entries.len() as u64 + 1;
+            if let Err(e) = self.update_next_index(peer_id, speculative_next_index) {
+                error!(
+                    "speculative next_index advance peer={} failed: {:?}",
+                    peer_id, e
+                );
+            }
+
             self.send_to_worker_or_spawn(
                 peer_id,
                 ReplicationTask::Append(request),
diff --git a/d-engine-core/src/raft_role/leader_state_test/become_follower_test.rs b/d-engine-core/src/raft_role/leader_state_test/become_follower_test.rs
index 96c2b42e..4027cd32 100644
--- a/d-engine-core/src/raft_role/leader_state_test/become_follower_test.rs
+++ b/d-engine-core/src/raft_role/leader_state_test/become_follower_test.rs
@@ -143,7 +143,7 @@ async fn test_receive_higher_term_append_entries_revokes_lease() {
             entries: vec![],
             leader_commit_index: 0,
         },
-        resp_tx,
+        vec![resp_tx],
     );
     state.handle_raft_event(event, &context, role_tx).await.ok();
 
@@ -247,7 +247,7 @@ async fn test_append_entries_higher_term_lease_revoked_before_become_follower()
             entries: vec![],
             leader_commit_index: 0,
         },
-        resp_tx,
+        vec![resp_tx],
     );
     state.handle_raft_event(event, &context, role_tx).await.ok();
 
diff --git a/d-engine-core/src/raft_role/leader_state_test/client_write_test.rs b/d-engine-core/src/raft_role/leader_state_test/client_write_test.rs
index 30748a35..0b77eb4b 100644
--- a/d-engine-core/src/raft_role/leader_state_test/client_write_test.rs
+++ b/d-engine-core/src/raft_role/leader_state_test/client_write_test.rs
@@ -7,12 +7,14 @@ use crate::ApplyResult;
 use crate::ClientCmd;
 use crate::MockRaftLog;
 use crate::MockReplicationCore;
+use crate::PrepareResult;
 use crate::RaftRequestWithSignal;
 use crate::client::{ClientWriteRequest, WriteOperation};
 use crate::client_command_to_entry_payloads;
 use crate::event::{NewCommitData, RoleEvent};
 use crate::maybe_clone_oneshot::MaybeCloneOneshot;
 use crate::maybe_clone_oneshot::RaftOneshot;
+use crate::mock_raft_context;
 use crate::raft_context::RaftContext;
 use crate::raft_role::leader_state::LeaderState;
 use crate::role_state::RaftRoleState;
@@ -21,9 +23,13 @@ use crate::test_utils::mock::MockTypeConfig;
 use crate::test_utils::node_config;
 use bytes::Bytes;
 use d_engine_proto::common::AddNode;
+use d_engine_proto::common::Entry;
 use d_engine_proto::common::EntryPayload;
+use d_engine_proto::common::NodeRole::Follower;
 use d_engine_proto::common::NodeStatus;
 use d_engine_proto::common::membership_change::Change;
+use d_engine_proto::server::cluster::NodeMeta;
+use d_engine_proto::server::replication::AppendEntriesRequest;
 use rand::distr::SampleString;
 use std::collections::VecDeque;
 use std::sync::Arc;
@@ -1415,3 +1421,85 @@ async fn test_single_voter_client_write_completes_after_log_flushed() {
         "client must receive Success error code"
     );
 }
+
+// Test speculative next index logic
+// Criteria: new next_index should be pushed forward before receiving append result
+//
+#[tokio::test]
+async fn test_speculative_next_index_before_receiving_append_result() {
+    let (_graceful_tx, graceful_rx) = watch::channel(());
+
+    let mut ctx = mock_raft_context(
+        "/tmp/test_speculative_next_index_before_receving_append_result",
+        graceful_rx,
+        None,
+    );
+
+    let mut membership = crate::MockMembership::<MockTypeConfig>::new();
+    membership.expect_is_single_node_cluster().returning(|| false);
+    membership.expect_voters().returning(|| {
+        vec![
+            NodeMeta {
+                id: 2,
+                address: "".into(),
+                status: NodeStatus::Active as i32,
+                role: Follower.into(),
+            },
+            NodeMeta {
+                id: 3,
+                address: "".into(),
+                status: NodeStatus::Active as i32,
+                role: Follower.into(),
+            },
+        ]
+    });
+    membership.expect_replication_peers().returning(Vec::new);
+    ctx.membership = Arc::new(membership);
+
+    let mut raft_log = MockRaftLog::new();
+    raft_log.expect_last_entry_id().returning(|| 1);
+    raft_log.expect_durable_index().returning(|| 0);
+    // Quorum blocked: calculate_majority_matched_index returns None (leader durable=0)
+    raft_log.expect_calculate_majority_matched_index().returning(|_, _, _| None);
+    ctx.storage.raft_log = Arc::new(raft_log);
+
+    let mut state = LeaderState::<MockTypeConfig>::new(1, ctx.node_config.clone());
+    state.init_cluster_metadata(&ctx.membership).await.unwrap();
+    assert!(!state.cluster_metadata.single_voter);
+
+    let follower_2_prev_log_index = 11;
+    ctx.handlers
+        .replication_handler
+        .expect_prepare_batch_requests()
+        .times(1)
+        .returning(move |_, _, _, _, _| {
+            Ok(PrepareResult {
+                append_requests: [(
+                    2,
+                    AppendEntriesRequest {
+                        term: 1,
+                        leader_id: 1,
+                        prev_log_index: follower_2_prev_log_index,
+                        prev_log_term: 1,
+                        entries: vec![Entry {
+                            index: 12,
+                            term: 1,
+                            payload: None,
+                        }],
+                        leader_commit_index: 3,
+                    },
+                )]
+                .to_vec(),
+                snapshot_targets: Vec::new(),
+            })
+        });
+
+    let (req, _rx) = write_request();
+    let (role_tx, _role_rx) = mpsc::unbounded_channel();
+    state.process_batch(VecDeque::from(vec![req]), &role_tx, &ctx).await.unwrap();
+    assert_eq!(
+        state.next_index(2),
+        Some(follower_2_prev_log_index + 1 + 1),
+        "follower(id=2) speculative next index is: prev_log_index + entries.len() + 1"
+    );
+}
diff --git a/d-engine-core/src/raft_role/leader_state_test/commit_index_test.rs b/d-engine-core/src/raft_role/leader_state_test/commit_index_test.rs
index 2fce443c..20734bde 100644
--- a/d-engine-core/src/raft_role/leader_state_test/commit_index_test.rs
+++ b/d-engine-core/src/raft_role/leader_state_test/commit_index_test.rs
@@ -4,11 +4,6 @@
 //! never `last_entry_id` (in-memory). Covers both the single-voter fast path and
 //! the multi-voter quorum path. Fixed by ticket #329.
 
-use std::collections::VecDeque;
-use std::sync::Arc;
-
-use tokio::sync::{mpsc, watch};
-
 use crate::event::RoleEvent;
 use crate::maybe_clone_oneshot::MaybeCloneOneshot;
 use crate::maybe_clone_oneshot::RaftOneshot;
@@ -18,9 +13,14 @@ use crate::test_utils::mock::{MockTypeConfig, mock_raft_context};
 use crate::{MockMembership, MockRaftLog, PeerUpdate, RaftRequestWithSignal};
 use bytes::Bytes;
 use d_engine_proto::common::{EntryPayload, LogId};
+use d_engine_proto::common::{NodeRole::Follower, NodeStatus};
+use d_engine_proto::server::cluster::NodeMeta;
 use d_engine_proto::server::replication::append_entries_response;
 use d_engine_proto::server::replication::{AppendEntriesResponse, SuccessResult};
 use rand::distr::SampleString;
+use std::collections::VecDeque;
+use std::sync::Arc;
+use tokio::sync::{mpsc, watch};
 
 fn success_response(
     term: u64,
@@ -186,8 +186,6 @@ async fn test_multi_voter_commit_respects_quorum_result() {
     let mut membership = crate::MockMembership::<MockTypeConfig>::new();
     membership.expect_is_single_node_cluster().returning(|| false);
     membership.expect_voters().returning(|| {
-        use d_engine_proto::common::{NodeRole::Follower, NodeStatus};
-        use d_engine_proto::server::cluster::NodeMeta;
         vec![
             NodeMeta {
                 id: 2,
diff --git a/d-engine-core/src/raft_role/leader_state_test/event_handling_test.rs b/d-engine-core/src/raft_role/leader_state_test/event_handling_test.rs
index 059dd4b0..ae61215a 100644
--- a/d-engine-core/src/raft_role/leader_state_test/event_handling_test.rs
+++ b/d-engine-core/src/raft_role/leader_state_test/event_handling_test.rs
@@ -389,7 +389,7 @@ async fn test_handle_append_entries_reject_same_term() {
     };
     use crate::maybe_clone_oneshot::MaybeCloneOneshot;
     let (resp_tx, mut resp_rx) = <MaybeCloneOneshot as RaftOneshot<_>>::new();
-    let raft_event = RaftEvent::AppendEntries(append_entries_request, resp_tx);
+    let raft_event = RaftEvent::AppendEntries(append_entries_request, vec![resp_tx]);
 
     let (role_tx, _role_rx) = mpsc::unbounded_channel();
 
@@ -453,7 +453,7 @@ async fn test_handle_append_entries_step_down_on_higher_term() {
     };
     use crate::maybe_clone_oneshot::MaybeCloneOneshot;
     let (resp_tx, mut resp_rx) = <MaybeCloneOneshot as RaftOneshot<_>>::new();
-    let raft_event = RaftEvent::AppendEntries(append_entries_request, resp_tx);
+    let raft_event = RaftEvent::AppendEntries(append_entries_request, vec![resp_tx]);
 
     let (role_tx, mut role_rx) = mpsc::unbounded_channel();
 
diff --git a/d-engine-core/src/raft_role/leader_state_test/replication_test.rs b/d-engine-core/src/raft_role/leader_state_test/replication_test.rs
index 38fd79f0..6dc2bfd1 100644
--- a/d-engine-core/src/raft_role/leader_state_test/replication_test.rs
+++ b/d-engine-core/src/raft_role/leader_state_test/replication_test.rs
@@ -4,34 +4,35 @@
 //! `execute_request_immediately` methods which handle log replication,
 //! commit index calculation, and client response delivery.
 
-use std::collections::VecDeque;
-use std::sync::Arc;
-use std::sync::atomic::{AtomicU64, Ordering};
-
-use bytes::Bytes;
-use rand::distr::SampleString;
-use tokio::sync::{mpsc, watch};
-use tonic::Status;
-use tracing_test::traced_test;
-
 use crate::Error;
 use crate::MockMembership;
 use crate::MockRaftLog;
-use crate::{MaybeCloneOneshot, PeerUpdate};
-
+use crate::MockReplicationCore;
+use crate::RaftContext;
 use crate::RaftRequestWithSignal;
-
 use crate::client::ClientResponse;
 use crate::event::RoleEvent;
 use crate::maybe_clone_oneshot::RaftOneshot;
 use crate::raft_role::leader_state::LeaderState;
 use crate::role_state::RaftRoleState;
+use crate::test_utils::mock::MockBuilder;
 use crate::test_utils::mock::MockTypeConfig;
 use crate::test_utils::mock::mock_raft_context;
+use crate::{MaybeCloneOneshot, PeerUpdate};
+use bytes::Bytes;
 use d_engine_proto::common::{EntryPayload, LogId, NodeRole::Follower, NodeStatus};
 use d_engine_proto::server::cluster::{ClusterMembership, NodeMeta};
+use d_engine_proto::server::replication::ConflictResult;
 use d_engine_proto::server::replication::append_entries_response;
 use d_engine_proto::server::replication::{AppendEntriesResponse, SuccessResult};
+use rand::distr::SampleString;
+use std::collections::VecDeque;
+use std::sync::Arc;
+use std::sync::atomic::{AtomicU64, Ordering};
+use tokio::sync::mpsc;
+use tokio::sync::watch;
+use tonic::Status;
+use tracing_test::traced_test;
 
 // ============================================================================
 // Test Helper Structures and Functions
@@ -2149,11 +2150,6 @@ async fn test_handle_append_result_rpc_failure_ignored() {
 #[tokio::test]
 #[traced_test]
 async fn test_stale_conflict_ack_does_not_regress_next_index() {
-    use crate::MockReplicationCore;
-    use crate::test_utils::mock::MockBuilder;
-    use d_engine_proto::server::replication::ConflictResult;
-    use tokio::sync::watch;
-
     let peer_id = 2u32;
 
     // Build a replication handler that returns next_index=1 for the conflict —
@@ -2258,3 +2254,259 @@ async fn test_stale_conflict_ack_does_not_regress_next_index() {
         "match_index must remain unchanged after a conflict response"
     );
 }
+
+async fn setup_state_with_next_and_match(
+    match_index: u64,
+    next_index: u64,
+    speculative_next_index: u64,
+) -> (
+    LeaderState<MockTypeConfig>,
+    AppendEntriesResponse,
+    RaftContext<MockTypeConfig>,
+) {
+    let follower2_id = 2u32;
+
+    // Build a replication handler that returns next_index=1 for the conflict —
+    // simulating what happens when a stale conflict from an empty-log follower arrives.
+    let mut rep = MockReplicationCore::<MockTypeConfig>::new();
+    rep.expect_handle_conflict_response().returning(move |_, _, _, _| {
+        Ok(crate::PeerUpdate {
+            match_index: None,
+            next_index,
+            success: false,
+        })
+    });
+
+    let (_graceful_tx, graceful_rx) = watch::channel(());
+    let mut context_inner =
+        MockBuilder::new(graceful_rx).with_replication_handler(rep).build_context();
+
+    // Set up membership with peer 2 and 3 as voters (same as setup_commit_index_test_context).
+    let mut membership = MockMembership::new();
+    membership.expect_can_rejoin().returning(|_, _| Ok(()));
+    membership.expect_voters().returning(|| {
+        vec![
+            NodeMeta {
+                id: 2,
+                address: "".into(),
+                status: NodeStatus::Active as i32,
+                role: Follower.into(),
+            },
+            NodeMeta {
+                id: 3,
+                address: "".into(),
+                status: NodeStatus::Active as i32,
+                role: Follower.into(),
+            },
+        ]
+    });
+    membership.expect_get_peers_id_with_condition().returning(|_| vec![]);
+    membership.expect_members().returning(Vec::new);
+    membership.expect_check_cluster_is_ready().returning(|| Ok(()));
+    membership.expect_retrieve_cluster_membership_config().returning(|_| {
+        d_engine_proto::server::cluster::ClusterMembership {
+            version: 1,
+            nodes: vec![],
+            current_leader_id: None,
+        }
+    });
+    membership.expect_pre_warm_connections().returning(|| Ok(()));
+    membership.expect_replication_peers().returning(|| {
+        vec![
+            NodeMeta {
+                id: 2,
+                address: "".into(),
+                status: NodeStatus::Active as i32,
+                role: Follower.into(),
+            },
+            NodeMeta {
+                id: 3,
+                address: "".into(),
+                status: NodeStatus::Active as i32,
+                role: Follower.into(),
+            },
+        ]
+    });
+    membership.expect_initial_cluster_size().returning(|| 3);
+    context_inner.membership = Arc::new(membership);
+
+    let mut state = LeaderState::<MockTypeConfig>::new(1, context_inner.node_config.clone());
+    state.update_commit_index(5).unwrap();
+    state.init_cluster_metadata(&context_inner.membership).await.unwrap();
+
+    // Simulate: SUCCESS from Batch A already processed — match_index = 100, next_index = 101.
+    state.match_index.insert(follower2_id, match_index);
+    state.next_index.insert(follower2_id, speculative_next_index);
+
+    // Stale CONFLICT from Batch B: follower's log was empty when it arrived.
+    let stale_conflict = AppendEntriesResponse {
+        node_id: follower2_id,
+        term: 1,
+        result: Some(append_entries_response::Result::Conflict(ConflictResult {
+            conflict_term: None,
+            conflict_index: Some(1),
+        })),
+    };
+
+    (state, stale_conflict, context_inner)
+}
+
+// Leader speculatively advances next_index to 13 before receiving an ACK.
+// When a fresh conflict response carries hint=7, next_index must roll back
+// from the speculative value (13) to the follower's hint (7) so the
+// retransmission starts from the correct entry.
+#[tokio::test]
+async fn test_stale_conflict_floor_guard_prevents_regression() {
+    let follower2_id = 2u32;
+    let follower2_speculative_next_index = 13;
+    let follower2_returned_next_index = 7;
+    let follower2_returned_match_index = 100;
+    let (mut state, stale_conflict, context_inner) = setup_state_with_next_and_match(
+        follower2_returned_match_index,
+        follower2_returned_next_index,
+        follower2_speculative_next_index,
+    )
+    .await;
+
+    let (role_tx, _role_rx) = mpsc::unbounded_channel();
+    let result = state
+        .handle_append_result(follower2_id, Ok(stale_conflict), &context_inner, &role_tx)
+        .await;
+
+    assert!(result.is_ok(), "stale conflict must not return an error");
+
+    // Floor guard: next_index must not retreat below match_index + 1 = 101.
+    assert_eq!(
+        state.next_index[&follower2_id],
+        follower2_returned_match_index + 1,
+        "stale conflict must not regress next_index below match_index + 1"
+    );
+}
+// Stale conflict arrives after match_index is already at 100.
+// The floor guard (match_index + 1 = 101) must override the conflict hint (7),
+// so next_index never retreats below what quorum already confirmed.
+#[tokio::test]
+async fn test_conflict_response_resets_speculative_next_index() {
+    let follower2_id = 2u32;
+    let follower2_speculative_next_index = 13;
+    let follower2_returned_next_index = 7;
+    let follower2_returned_match_index = 1;
+
+    let (mut state, stale_conflict, context_inner) = setup_state_with_next_and_match(
+        follower2_returned_match_index,
+        follower2_returned_next_index,
+        follower2_speculative_next_index,
+    )
+    .await;
+
+    let (role_tx, _role_rx) = mpsc::unbounded_channel();
+
+    let result = state
+        .handle_append_result(follower2_id, Ok(stale_conflict), &context_inner, &role_tx)
+        .await;
+
+    assert!(result.is_ok(), "stale conflict must not return an error");
+
+    // Floor guard: next_index must not retreat below match_index + 1 = 101.
+    assert_eq!(
+        state.next_index[&follower2_id], follower2_returned_next_index,
+        "stale conflict must not regress next_index below match_index + 1"
+    );
+}
+
+mod test_leader_update_next_index {
+    use super::*;
+
+    // On a success ACK, next_index must not regress below the current speculative value.
+    // The ACK carries peer_match + 1 which may lag behind what the leader has
+    // already pipeline-sent; the leader must keep the higher speculative value.
+    #[tokio::test]
+    async fn test_update_peer_index_success_preserves_speculative_next_index() {
+        let (_graceful_tx, graceful_rx) = watch::channel(());
+        let context_inner = MockBuilder::new(graceful_rx).build_context();
+
+        let mut state = LeaderState::<MockTypeConfig>::new(1, context_inner.node_config.clone());
+        state.update_next_index(2, 100).unwrap();
+        state.update_match_index(2, 7).unwrap();
+        state.update_peer_index(
+            2,
+            &PeerUpdate {
+                match_index: Some(11),
+                next_index: 12,
+                success: true,
+            },
+        );
+        assert_eq!(state.next_index.get(&2), Some(&100));
+        assert_eq!(state.match_index.get(&2), Some(&11));
+    }
+
+    // On a conflict response, next_index must retreat to the follower's hint,
+    // allowing the leader to retransmit from the correct position.
+    // The floor guard (match_index + 1) prevents retreating below confirmed entries.
+    #[tokio::test]
+    async fn test_update_peer_index_conflict_retreats_to_hint() {
+        let (_graceful_tx, graceful_rx) = watch::channel(());
+        let context_inner = MockBuilder::new(graceful_rx).build_context();
+
+        let mut state = LeaderState::<MockTypeConfig>::new(1, context_inner.node_config.clone());
+        state.update_next_index(2, 100).unwrap();
+        state.update_match_index(2, 7).unwrap();
+        state.update_peer_index(
+            2,
+            &PeerUpdate {
+                match_index: None,
+                next_index: 12,
+                success: false,
+            },
+        );
+        assert_eq!(state.next_index.get(&2), Some(&12));
+        assert_eq!(state.match_index.get(&2), Some(&7));
+    }
+
+    // On a conflict response, if the computed retry point (PeerUpdate::next_index)
+    // is below the confirmed match_index + 1, the floor guard must clamp next_index to match_index + 1.
+    // Retreating below already-confirmed entries would violate Raft's Log Matching Property.
+    #[tokio::test]
+    async fn test_update_peer_index_conflict_floor_guard_prevents_retreat_below_match_index() {
+        let (_graceful_tx, graceful_rx) = watch::channel(());
+        let context_inner = MockBuilder::new(graceful_rx).build_context();
+
+        let mut state = LeaderState::<MockTypeConfig>::new(1, context_inner.node_config.clone());
+        state.update_next_index(2, 100).unwrap();
+        state.update_match_index(2, 50).unwrap();
+        state.update_peer_index(
+            2,
+            &PeerUpdate {
+                match_index: None,
+                next_index: 12,
+                success: false,
+            },
+        );
+        assert_eq!(state.next_index.get(&2), Some(&(50 + 1)));
+        assert_eq!(state.match_index.get(&2), Some(&50));
+    }
+
+    // On a success ACK, if the ACK carries a next_index larger than the current
+    // speculative value (e.g. after reconnect or snapshot install where next_index
+    // was never advanced), next_index must be updated to the ACK value.
+    // Failing to do so causes the leader to re-send the same entries forever.
+    #[tokio::test]
+    async fn test_update_peer_index_success_advances_next_index_when_ack_is_ahead() {
+        let (_graceful_tx, graceful_rx) = watch::channel(());
+        let context_inner = MockBuilder::new(graceful_rx).build_context();
+
+        let mut state = LeaderState::<MockTypeConfig>::new(1, context_inner.node_config.clone());
+        state.update_next_index(2, 1).unwrap();
+        state.update_match_index(2, 1).unwrap();
+        state.update_peer_index(
+            2,
+            &PeerUpdate {
+                match_index: Some(11),
+                next_index: 12,
+                success: true,
+            },
+        );
+        assert_eq!(state.next_index.get(&2), Some(&12));
+        assert_eq!(state.match_index.get(&2), Some(&11));
+    }
+}
diff --git a/d-engine-core/src/raft_role/learner_state_test.rs b/d-engine-core/src/raft_role/learner_state_test.rs
index 25169c52..8531c336 100644
--- a/d-engine-core/src/raft_role/learner_state_test.rs
+++ b/d-engine-core/src/raft_role/learner_state_test.rs
@@ -318,7 +318,7 @@ async fn test_learner_handles_append_entries_success() {
         leader_commit_index: 0,
     };
     let (resp_tx, mut resp_rx) = MaybeCloneOneshot::new();
-    let raft_event = RaftEvent::AppendEntries(append_request, resp_tx);
+    let raft_event = RaftEvent::AppendEntries(append_request, vec![resp_tx]);
     let (role_tx, mut role_rx) = mpsc::unbounded_channel();
 
     assert!(
@@ -379,7 +379,7 @@ async fn test_learner_rejects_append_entries_stale_term() {
         leader_commit_index: 0,
     };
     let (resp_tx, mut resp_rx) = MaybeCloneOneshot::new();
-    let raft_event = RaftEvent::AppendEntries(append_request, resp_tx);
+    let raft_event = RaftEvent::AppendEntries(append_request, vec![resp_tx]);
     let (role_tx, mut role_rx) = mpsc::unbounded_channel();
 
     assert!(
@@ -436,7 +436,7 @@ async fn test_learner_handles_append_entries_handler_error() {
         leader_commit_index: 0,
     };
     let (resp_tx, mut resp_rx) = MaybeCloneOneshot::new();
-    let raft_event = RaftEvent::AppendEntries(append_request, resp_tx);
+    let raft_event = RaftEvent::AppendEntries(append_request, vec![resp_tx]);
     let (role_tx, mut role_rx) = mpsc::unbounded_channel();
 
     assert!(
@@ -1740,7 +1740,7 @@ async fn test_learner_acks_immediately_after_memory_write() {
         leader_commit_index: 0,
     };
     let (resp_tx, mut resp_rx) = MaybeCloneOneshot::new();
-    let raft_event = RaftEvent::AppendEntries(append_request, resp_tx);
+    let raft_event = RaftEvent::AppendEntries(append_request, vec![resp_tx]);
     let (role_tx, _role_rx) = mpsc::unbounded_channel();
 
     assert!(state.handle_raft_event(raft_event, &context, role_tx).await.is_ok());
diff --git a/d-engine-core/src/raft_role/role_state.rs b/d-engine-core/src/raft_role/role_state.rs
index e154a8cd..2e7f98a5 100644
--- a/d-engine-core/src/raft_role/role_state.rs
+++ b/d-engine-core/src/raft_role/role_state.rs
@@ -449,7 +449,7 @@ pub trait RaftRoleState: Send + Sync + 'static {
     async fn handle_append_entries_request_workflow(
         &mut self,
         append_entries_request: AppendEntriesRequest,
-        sender: MaybeCloneOneshotSender<std::result::Result<AppendEntriesResponse, Status>>,
+        senders: Vec<MaybeCloneOneshotSender<std::result::Result<AppendEntriesResponse, Status>>>,
         ctx: &RaftContext<Self::T>,
         role_tx: mpsc::UnboundedSender<RoleEvent>,
         state_snapshot: &StateSnapshot,
@@ -469,11 +469,12 @@ pub trait RaftRoleState: Send + Sync + 'static {
             let response = AppendEntriesResponse::higher_term(self.node_id(), my_term);
             debug!("AppendEntriesResponse: {:?}", response);
 
-            sender.send(Ok(response)).map_err(|e| {
-                let error_str = format!("{e:?}");
-                error!("Failed to send: {}", error_str);
-                NetworkError::SingalSendFailed(error_str)
-            })?;
+            for sender in senders {
+                if let Err(e) = sender.send(Ok(response)) {
+                    error!("Failed to send client: {:?}", e);
+                }
+            }
+
             return Ok(());
         }
 
@@ -546,11 +547,12 @@ pub trait RaftRoleState: Send + Sync + 'static {
                 // MemFirst: ACK immediately after memory write. IO thread fsyncs async.
                 // Safety: quorum uses last_entry_id (in-memory); crash safety is guaranteed by
                 // majority replication, not per-follower durability.
-                sender.send(Ok(response)).map_err(|e| {
-                    let error_str = format!("{e:?}");
-                    error!("Failed to send: {}", error_str);
-                    NetworkError::SingalSendFailed(error_str)
-                })?;
+
+                for sender in senders {
+                    if let Err(e) = sender.send(Ok(response)) {
+                        error!("Failed to send: {:?}", e);
+                    }
+                }
             }
             Err(e) => {
                 // Conservatively fallback to a safe position, forcing the leader to retry or
@@ -567,11 +569,12 @@ pub trait RaftRoleState: Send + Sync + 'static {
                 );
                 debug!("AppendEntriesResponse: {:?}", response);
 
-                sender.send(Ok(response)).map_err(|e| {
-                    let error_str = format!("{e:?}");
-                    error!("Failed to send: {}", error_str);
-                    NetworkError::SingalSendFailed(error_str)
-                })?;
+                for sender in senders {
+                    if let Err(e) = sender.send(Ok(response)) {
+                        error!("Failed to send: {:?}", e);
+                    }
+                }
+
                 error("handle_raft_event", &e);
                 return Err(e);
             }
diff --git a/d-engine-core/src/raft_test/drain_based_batch_architecture_tests.rs b/d-engine-core/src/raft_test/drain_based_batch_architecture_tests.rs
index a089b664..71037e93 100644
--- a/d-engine-core/src/raft_test/drain_based_batch_architecture_tests.rs
+++ b/d-engine-core/src/raft_test/drain_based_batch_architecture_tests.rs
@@ -25,6 +25,7 @@
 //!
 
 use crate::RaftOneshot;
+use crate::client::{ClientReadRequest, ClientWriteRequest, ReadConsistencyPolicy};
 use crate::maybe_clone_oneshot::MaybeCloneOneshot;
 use crate::{
     MockBuilder,
@@ -32,10 +33,8 @@ use crate::{
     raft_role::{RaftRole, role_state::RaftRoleState},
 };
 use bytes::Bytes;
+use d_engine_proto::client::WriteCommand;
 use d_engine_proto::client::write_command::Operation;
-use d_engine_proto::client::{
-    ClientReadRequest, ClientWriteRequest, ReadConsistencyPolicy, WriteCommand,
-};
 use std::sync::Arc;
 use std::time::Duration;
 use tokio::sync::watch;
diff --git a/d-engine-core/src/raft_test/merge_append_entries_tests.rs b/d-engine-core/src/raft_test/merge_append_entries_tests.rs
new file mode 100644
index 00000000..da63adf0
--- /dev/null
+++ b/d-engine-core/src/raft_test/merge_append_entries_tests.rs
@@ -0,0 +1,887 @@
+//! # AppendEntries Merge Tests
+//!
+//! Tests for `Raft::merge_append_entries()` — the function that coalesces
+//! consecutive contiguous `RaftEvent::AppendEntries` events in the buffer
+//! into a single event before dispatch.
+//!
+//! ## Design decisions under test
+//! - Only the leading run of contiguous same-term AppendEntries is merged.
+//! - Heartbeats (empty entries, same term and prev_log_index as the next slot) are absorbed
+//!   into the merge group: their sender is collected and their `leader_commit_index` is folded
+//!   in via `max()`, advancing the commit index without a separate dispatch.
+//! - Non-AppendEntries events stop the merge and remain in FIFO order.
+//! - All senders from merged events are collected into a Vec for broadcast.
+//!
+//! ## Setup pattern
+//! Each test builds a `Raft<MockTypeConfig>` via `MockBuilder`, pushes events
+//! directly into `raft.buffered_raft_event`, calls `raft.merge_append_entries()`,
+//! then inspects the resulting buffer.
+
+use crate::maybe_clone_oneshot::MaybeCloneOneshot;
+use crate::test_utils::MockBuilder;
+use crate::{MaybeCloneOneshotReceiver, RaftEvent, RaftNodeConfig, RaftOneshot, mock_entries};
+use d_engine_proto::common::LogId;
+use d_engine_proto::server::election::{VoteRequest, VoteResponse};
+use d_engine_proto::server::replication::{AppendEntriesRequest, AppendEntriesResponse};
+use std::collections::VecDeque;
+use tokio::sync::watch;
+
+// -----------------------------------------------------------------------
+// Helpers
+// -----------------------------------------------------------------------
+
+/// Build an `AppendEntriesRequest` covering `count` entries starting after
+/// `prev_log_index`, all with the given `term`.
+///
+/// Entry indices are `prev_log_index + 1 ..= prev_log_index + count`.
+fn make_request(
+    term: u64,
+    prev_log_index: u64,
+    count: u64,
+    leader_commit: u64,
+) -> AppendEntriesRequest {
+    AppendEntriesRequest {
+        term,
+        leader_id: 1,
+        prev_log_index,
+        prev_log_term: term,
+        entries: mock_entries(1, count, term),
+        leader_commit_index: leader_commit,
+    }
+}
+
+/// Wrap a request and a fresh oneshot sender into `RaftEvent::AppendEntries`.
+/// Returns `(event, receiver)` — the receiver lets tests verify send results.
+fn make_event(
+    req: AppendEntriesRequest
+) -> (
+    RaftEvent,
+    MaybeCloneOneshotReceiver<Result<AppendEntriesResponse, tonic::Status>>,
+) {
+    let (tx, rx) = MaybeCloneOneshot::new();
+    (RaftEvent::AppendEntries(req, vec![tx]), rx)
+}
+
+fn make_none_append_entries_event() -> (
+    RaftEvent,
+    MaybeCloneOneshotReceiver<Result<VoteResponse, tonic::Status>>,
+) {
+    let (tx, rx) = MaybeCloneOneshot::new();
+    (
+        RaftEvent::ReceiveVoteRequest(
+            VoteRequest {
+                term: 1,
+                candidate_id: 2,
+                last_log_index: 11,
+                last_log_term: 3,
+            },
+            tx,
+        ),
+        rx,
+    )
+}
+
+// -----------------------------------------------------------------------
+// Edge cases — buffer state before any merge opportunity
+// -----------------------------------------------------------------------
+
+/// Calling `merge_append_entries` on an empty buffer is a no-op.
+/// Buffer remains empty; function returns without panicking.
+///
+/// Buffer before: []
+///
+/// After merge:
+/// - buffer.len() == 0
+#[test]
+fn test_empty_buffer_is_noop() {
+    let (_graceful_tx, graceful_rx) = watch::channel(());
+    let mut raft = MockBuilder::new(graceful_rx).build_raft();
+    raft.buffered_raft_event = VecDeque::new();
+
+    raft.merge_append_entries();
+    assert_eq!(raft.buffered_raft_event.len(), 0);
+}
+
+/// A buffer with exactly one AppendEntries event is returned unchanged.
+/// The event is popped and immediately pushed back as-is; no entries are
+/// added, no senders are lost.
+///
+/// Buffer before: [AE(prev=9, [10..14], term=5)]
+///
+/// After merge:
+/// - buffer.len() == 1
+/// - entries.len() == 5  (unchanged)
+/// - senders.len() == 1  (unchanged)
+/// - prev_log_index == 9 (unchanged)
+#[test]
+fn test_single_append_entries_event_returned_unchanged() {
+    let (_graceful_tx, graceful_rx) = watch::channel(());
+    let mut raft = MockBuilder::new(graceful_rx).build_raft();
+    let mut queue = VecDeque::new();
+    let request = make_request(5, 9u64, 5, 1);
+    let (event, _receiver) = make_event(request);
+    queue.push_back(event);
+
+    raft.buffered_raft_event = queue;
+
+    raft.merge_append_entries();
+    assert_eq!(raft.buffered_raft_event.len(), 1);
+    match raft.buffered_raft_event.pop_front() {
+        Some(RaftEvent::AppendEntries(request, senders)) => {
+            let entries = request.entries;
+            assert_eq!(entries.len(), 5);
+            assert_eq!(senders.len(), 1);
+            assert_eq!(request.term, 5);
+            assert_eq!(request.prev_log_index, 9);
+        }
+        _ => panic!("expected RaftEvent::AppendEntries, got something else"),
+    }
+}
+
+/// When the first event is not AppendEntries (e.g. VoteRequest), the function
+/// pushes it back to the front and returns without touching anything else.
+/// Buffer order and length are unchanged.
+///
+/// Buffer before: [VoteRequest(term=1), AE(prev=9, [10..14], term=5)]
+///
+/// After merge:
+/// - buffer.len() == 2
+/// - buffer[0]: VoteRequest (unchanged, still at front)
+/// - buffer[1]: AE (unchanged, not merged)
+#[test]
+fn test_first_non_append_entries_event_is_noop() {
+    let (_graceful_tx, graceful_rx) = watch::channel(());
+    let mut raft = MockBuilder::new(graceful_rx).build_raft();
+    let mut queue = VecDeque::new();
+    for (is_ae, prev, count, term) in [(false, 9u64, 5, 15), (true, 9u64, 5, 5)] {
+        if is_ae {
+            let request = make_request(term, prev, count, 1);
+            let (event, _receiver) = make_event(request);
+            queue.push_back(event);
+        } else {
+            let (event, _receiver) = make_none_append_entries_event();
+            queue.push_back(event);
+        }
+    }
+    raft.buffered_raft_event = queue;
+
+    raft.merge_append_entries();
+    assert_eq!(raft.buffered_raft_event.len(), 2);
+
+    // VoteRequest
+    if let Some(RaftEvent::ReceiveVoteRequest(request, _senders)) =
+        raft.buffered_raft_event.pop_front()
+    {
+        assert_eq!(request.last_log_index, 11);
+    } else {
+        panic!("expected RaftEvent::ReceiveVoteRequest, got something else");
+    }
+
+    // AE
+    match raft.buffered_raft_event.pop_front() {
+        Some(RaftEvent::AppendEntries(request, senders)) => {
+            let entries = request.entries;
+            assert_eq!(entries.len(), 5);
+            assert_eq!(senders.len(), 1);
+            assert_eq!(request.prev_log_index, 9);
+        }
+        _ => panic!("expected RaftEvent::AppendEntries, got something else"),
+    }
+}
+
+// -----------------------------------------------------------------------
+// Basic merge
+// -----------------------------------------------------------------------
+
+/// Three contiguous AppendEntries from the same term are merged into one.
+///
+/// Buffer before: [AE(prev=9, entries=[10..14], term=5),
+///                  AE(prev=14, entries=[15..19], term=5),
+///                  AE(prev=19, entries=[20..24], term=5)]
+///
+/// After merge:
+/// - buffer.len() == 1
+/// - merged entries.len() == 15
+/// - merged senders.len() == 3
+/// - prev_log_index == 9  (taken from first request)
+/// - term == 5
+#[test]
+fn test_merge_three_contiguous_entries_same_term() {
+    let (_graceful_tx, graceful_rx) = watch::channel(());
+    let mut raft = MockBuilder::new(graceful_rx).build_raft();
+    let mut queue = VecDeque::new();
+    for prev in [9u64, 14, 19] {
+        let request = make_request(5, prev, 5, 1);
+        let (event, _receiver) = make_event(request);
+        queue.push_back(event);
+    }
+
+    raft.buffered_raft_event = queue;
+
+    raft.merge_append_entries();
+    assert_eq!(raft.buffered_raft_event.len(), 1);
+    match raft.buffered_raft_event.pop_front() {
+        Some(RaftEvent::AppendEntries(request, senders)) => {
+            let entries = request.entries;
+            assert_eq!(entries.len(), 15);
+            assert_eq!(senders.len(), 3);
+            assert_eq!(request.term, 5);
+            assert_eq!(request.prev_log_index, 9);
+        }
+        _ => panic!("expected RaftEvent::AppendEntries, got something else"),
+    }
+}
+
+/// The merged event preserves the metadata (term, leader_id, prev_log_index,
+/// prev_log_term, leader_commit_index) from the first request unchanged.
+/// Only `entries` and `senders` are accumulated from subsequent events.
+///
+/// Buffer before: [AE(term=5, leader_id=1, prev=9, prev_term=5, commit=42, [10..14]),
+///                  AE(term=5, leader_id=1, prev=14, prev_term=5, commit=99, [15..19])]
+///
+/// After merge:
+/// - term == 5               (from first)
+/// - leader_id == 1          (from first)
+/// - prev_log_index == 9     (from first)
+/// - prev_log_term == 5      (from first)
+/// - leader_commit_index == 99  (max leader commit index)
+/// - entries.len() == 10
+#[test]
+fn test_merge_preserves_first_request_metadata() {
+    let (_graceful_tx, graceful_rx) = watch::channel(());
+    let mut raft = MockBuilder::new(graceful_rx).build_raft();
+    let mut queue = VecDeque::new();
+    for (prev, term, commit) in [(9u64, 5, 42), (14, 5, 99)] {
+        let request = make_request(term, prev, 5, commit);
+        let (event, _receiver) = make_event(request);
+        queue.push_back(event);
+    }
+    raft.buffered_raft_event = queue;
+
+    raft.merge_append_entries();
+    assert_eq!(raft.buffered_raft_event.len(), 1);
+
+    // front
+    match raft.buffered_raft_event.pop_front() {
+        Some(RaftEvent::AppendEntries(request, senders)) => {
+            let entries = request.entries;
+            assert_eq!(entries.len(), 10);
+            assert_eq!(senders.len(), 2);
+            assert_eq!(request.term, 5);
+            assert_eq!(request.prev_log_index, 9);
+            assert_eq!(request.leader_commit_index, 99);
+        }
+        _ => panic!("expected RaftEvent::AppendEntries, got something else"),
+    }
+}
+
+// -----------------------------------------------------------------------
+// Merge boundary: term mismatch
+// -----------------------------------------------------------------------
+
+/// When term changes mid-sequence, only the prefix with matching term is merged.
+/// The event with the new term is pushed back and remains as the next item.
+///
+/// Buffer before: [AE(term=5, prev=9,  [10..14]),
+///                  AE(term=5, prev=14, [15..19]),
+///                  AE(term=6, prev=19, [20..24])]
+///
+/// After merge:
+/// - buffer.len() == 2
+/// - front:  merged AE(term=5, entries=[10..19], senders=[s1,s2])
+/// - back:   AE(term=6, prev=19, entries=[20..24], senders=[s3])  — untouched
+#[test]
+fn test_term_mismatch_stops_merge() {
+    let (_graceful_tx, graceful_rx) = watch::channel(());
+    let mut raft = MockBuilder::new(graceful_rx).build_raft();
+    let mut queue = VecDeque::new();
+    for (prev, term) in [(9u64, 5), (14, 5), (19, 6)] {
+        let request = make_request(term, prev, 5, 1);
+        let (event, _receiver) = make_event(request);
+        queue.push_back(event);
+    }
+    raft.buffered_raft_event = queue;
+
+    raft.merge_append_entries();
+    assert_eq!(raft.buffered_raft_event.len(), 2);
+
+    // front
+    match raft.buffered_raft_event.pop_front() {
+        Some(RaftEvent::AppendEntries(request, senders)) => {
+            let entries = request.entries;
+            assert_eq!(entries.len(), 10);
+            assert_eq!(senders.len(), 2);
+            assert_eq!(request.term, 5);
+            assert_eq!(request.prev_log_index, 9);
+        }
+        _ => panic!("expected RaftEvent::AppendEntries, got something else"),
+    }
+
+    // back
+    match raft.buffered_raft_event.pop_back() {
+        Some(RaftEvent::AppendEntries(request, senders)) => {
+            let entries = request.entries;
+            assert_eq!(entries.len(), 5);
+            assert_eq!(senders.len(), 1);
+            assert_eq!(request.term, 6);
+            assert_eq!(request.prev_log_index, 19);
+        }
+        _ => panic!("expected RaftEvent::AppendEntries, got something else"),
+    }
+}
+
+// -----------------------------------------------------------------------
+// Merge boundary: non-contiguous log range
+// -----------------------------------------------------------------------
+
+/// When `prev_log_index` of the next event does not equal the end of the
+/// accumulated entries, the merge stops.
+///
+/// Buffer before: [AE(prev=9, [10..14], term=5),
+///                  AE(prev=20, [21..24], term=5)]   ← gap at 15-20
+///
+/// After merge:
+/// - buffer.len() == 2
+/// - front: AE with 5 entries, prev=9
+/// - back:  AE with prev=20, unchanged
+#[test]
+fn test_non_contiguous_prev_log_index_stops_merge() {
+    let (_graceful_tx, graceful_rx) = watch::channel(());
+    let mut raft = MockBuilder::new(graceful_rx).build_raft();
+    let mut queue = VecDeque::new();
+    for (prev, term) in [(9u64, 5), (20, 5)] {
+        let request = make_request(term, prev, 5, 1);
+        let (event, _receiver) = make_event(request);
+        queue.push_back(event);
+    }
+    raft.buffered_raft_event = queue;
+
+    raft.merge_append_entries();
+    assert_eq!(raft.buffered_raft_event.len(), 2);
+
+    // front
+    match raft.buffered_raft_event.pop_front() {
+        Some(RaftEvent::AppendEntries(request, senders)) => {
+            let entries = request.entries;
+            assert_eq!(entries.len(), 5);
+            assert_eq!(senders.len(), 1);
+            assert_eq!(request.term, 5);
+            assert_eq!(request.prev_log_index, 9);
+        }
+        _ => panic!("expected RaftEvent::AppendEntries, got something else"),
+    }
+
+    // back
+    match raft.buffered_raft_event.pop_back() {
+        Some(RaftEvent::AppendEntries(request, senders)) => {
+            let entries = request.entries;
+            assert_eq!(entries.len(), 5);
+            assert_eq!(senders.len(), 1);
+            assert_eq!(request.term, 5);
+            assert_eq!(request.prev_log_index, 20);
+        }
+        _ => panic!("expected RaftEvent::AppendEntries, got something else"),
+    }
+}
+
+// -----------------------------------------------------------------------
+// Merge boundary: heartbeat (empty entries) — Method B absorption
+// -----------------------------------------------------------------------
+
+/// A heartbeat (entries=[]) that chains with the preceding AE is absorbed
+/// into the merge group rather than stopping it (Method B).
+///
+/// The heartbeat contributes no entries but its senders are collected and
+/// `next_prev` is unchanged, allowing subsequent AEs to continue chaining.
+///
+/// Buffer before: [AE(prev=9,  entries=[10..14], commit=10),
+///                  HB(prev=14, entries=[],        commit=15),
+///                  AE(prev=14, entries=[15..19],  commit=10)]
+///
+/// After merge:
+/// - buffer.len() == 1
+/// - entries.len() == 10  (5 from first AE + 0 from HB + 5 from last AE)
+/// - senders.len() == 3   (one per original event)
+/// - prev_log_index == 9  (from first request)
+/// - leader_commit_index == 15  (max across all three)
+#[test]
+fn test_heartbeat_absorbed_into_merge() {
+    let (_graceful_tx, graceful_rx) = watch::channel(());
+    let mut raft = MockBuilder::new(graceful_rx).build_raft();
+    let mut queue = VecDeque::new();
+    for (prev, count, commit) in [(9u64, 5, 10), (14, 0, 15), (14, 5, 10)] {
+        let request = make_request(5, prev, count, commit);
+        let (event, _receiver) = make_event(request);
+        queue.push_back(event);
+    }
+    raft.buffered_raft_event = queue;
+
+    raft.merge_append_entries();
+    assert_eq!(raft.buffered_raft_event.len(), 1);
+
+    // front
+    match raft.buffered_raft_event.pop_front() {
+        Some(RaftEvent::AppendEntries(request, senders)) => {
+            let entries = request.entries;
+            assert_eq!(entries.len(), 10);
+            assert_eq!(senders.len(), 3);
+            assert_eq!(request.prev_log_index, 9);
+            assert_eq!(request.leader_commit_index, 15);
+        }
+        _ => panic!("expected RaftEvent::AppendEntries, got something else"),
+    }
+}
+
+/// Verify that a heartbeat's `leader_commit_index` is promoted via `max()`
+/// into the merged result, not silently dropped.
+///
+/// When the HB carries a higher commit than surrounding AEs, the merged
+/// event must reflect that higher value so the follower advances its
+/// commit index correctly on the single `workflow()` call.
+///
+/// Buffer before: [AE(prev=9, [10..14], commit=10),
+///                  HB(prev=14, [],       commit=99),   ← high commit
+///                  AE(prev=14, [15..19], commit=10)]
+///
+/// After merge:
+/// - buffer.len() == 1
+/// - merged.leader_commit_index == 99  (HB value wins via max)
+#[test]
+fn test_heartbeat_commit_index_promoted_via_max() {
+    let (_graceful_tx, graceful_rx) = watch::channel(());
+    let mut raft = MockBuilder::new(graceful_rx).build_raft();
+    let mut queue = VecDeque::new();
+    for (prev, count, commit) in [(9u64, 5, 10), (14, 0, 99), (14, 5, 10)] {
+        let request = make_request(5, prev, count, commit);
+        let (event, _receiver) = make_event(request);
+        queue.push_back(event);
+    }
+    raft.buffered_raft_event = queue;
+
+    raft.merge_append_entries();
+    assert_eq!(raft.buffered_raft_event.len(), 1);
+
+    // front
+    match raft.buffered_raft_event.pop_front() {
+        Some(RaftEvent::AppendEntries(request, senders)) => {
+            let entries = request.entries;
+            assert_eq!(entries.len(), 10);
+            assert_eq!(senders.len(), 3);
+            assert_eq!(request.prev_log_index, 9);
+            assert_eq!(request.leader_commit_index, 99);
+        }
+        _ => panic!("expected RaftEvent::AppendEntries, got something else"),
+    }
+}
+
+// -----------------------------------------------------------------------
+// Merge boundary: non-AppendEntries event in sequence (FIFO invariant)
+// -----------------------------------------------------------------------
+
+/// A non-AppendEntries event (e.g. VoteRequest) between two AppendEntries
+/// stops the merge and preserves full FIFO ordering.
+///
+/// Rationale: skipping over non-AppendEntries events to merge surrounding
+/// AEs would change relative processing order, which can break election
+/// safety (§5.2) — a VoteRequest must be processed before a later AE.
+///
+/// Buffer before: [AE(prev=9, [10..14], term=5),
+///                  VoteRequest(...),
+///                  AE(prev=14, [15..19], term=5)]
+///
+/// After merge:
+/// - buffer.len() == 3
+/// - buffer[0]: AE with 5 entries
+/// - buffer[1]: VoteRequest (unchanged, in original position)
+/// - buffer[2]: AE with entries=[15..19]  (NOT merged with buffer[0])
+#[test]
+fn test_non_append_entries_event_stops_merge_fifo_preserved() {
+    let (_graceful_tx, graceful_rx) = watch::channel(());
+    let mut raft = MockBuilder::new(graceful_rx).build_raft();
+    let mut queue = VecDeque::new();
+    for (is_ae, prev, count, term) in [(true, 9u64, 5, 5), (false, 14, 0, 0), (true, 14, 5, 5)] {
+        if is_ae {
+            let request = make_request(term, prev, count, 1);
+            let (event, _receiver) = make_event(request);
+            queue.push_back(event);
+        } else {
+            let (event, _receiver) = make_none_append_entries_event();
+            queue.push_back(event);
+        }
+    }
+    raft.buffered_raft_event = queue;
+
+    raft.merge_append_entries();
+    assert_eq!(raft.buffered_raft_event.len(), 3);
+
+    // AE
+    match raft.buffered_raft_event.pop_front() {
+        Some(RaftEvent::AppendEntries(request, senders)) => {
+            let entries = request.entries;
+            assert_eq!(entries.len(), 5);
+            assert_eq!(senders.len(), 1);
+            assert_eq!(request.prev_log_index, 9);
+        }
+        _ => panic!("expected RaftEvent::AppendEntries, got something else"),
+    }
+
+    // VoteRequest
+    if let Some(RaftEvent::ReceiveVoteRequest(request, _senders)) =
+        raft.buffered_raft_event.pop_front()
+    {
+        assert_eq!(request.last_log_index, 11);
+    } else {
+        panic!("expected RaftEvent::ReceiveVoteRequest, got something else");
+    }
+
+    // AE
+    match raft.buffered_raft_event.pop_front() {
+        Some(RaftEvent::AppendEntries(request, senders)) => {
+            let entries = request.entries;
+            assert_eq!(entries.len(), 5);
+            assert_eq!(senders.len(), 1);
+            assert_eq!(request.prev_log_index, 14);
+        }
+        _ => panic!("expected RaftEvent::AppendEntries, got something else"),
+    }
+}
+
+// -----------------------------------------------------------------------
+// Sender collection
+// -----------------------------------------------------------------------
+
+/// All senders from merged events appear in the resulting Vec, in order.
+/// Sender count == number of original events that were merged.
+///
+/// Buffer before: [AE(prev=9, [10..14]), AE(prev=14, [15..19]), AE(prev=19, [20..24])]
+///
+/// After merge:
+/// - senders.len() == 3
+/// - senders appear in original push order (s1, s2, s3)
+#[tokio::test]
+async fn test_all_senders_collected_in_merge_order() {
+    let (_graceful_tx, graceful_rx) = watch::channel(());
+    let mut raft = MockBuilder::new(graceful_rx).build_raft();
+    let mut queue = VecDeque::new();
+    let mut receivers = vec![];
+
+    for (prev, count) in [(9u64, 5), (14, 5), (19, 5)] {
+        let request = make_request(5, prev, count, 1);
+        let (event, rx) = make_event(request);
+        receivers.push(rx);
+        queue.push_back(event);
+    }
+    raft.buffered_raft_event = queue;
+
+    raft.merge_append_entries();
+    assert_eq!(raft.buffered_raft_event.len(), 1);
+
+    let Some(RaftEvent::AppendEntries(_, senders)) = raft.buffered_raft_event.pop_front() else {
+        panic!("expected AppendEntries");
+    };
+    assert_eq!(senders.len(), 3);
+
+    for (i, sender) in senders.iter().enumerate() {
+        let matched = (i as u64 + 1) * 10; // 10, 20, 30
+        sender
+            .send(Ok(AppendEntriesResponse::success(
+                (i + 1) as u32,
+                5,
+                Some(LogId {
+                    term: matched,
+                    index: matched,
+                }),
+            )))
+            .ok();
+    }
+
+    let mut iter = receivers.into_iter();
+    let r1 = iter.next().unwrap().await.unwrap().unwrap();
+    let r2 = iter.next().unwrap().await.unwrap().unwrap();
+    let r3 = iter.next().unwrap().await.unwrap().unwrap();
+    assert_eq!(r1.node_id, 1);
+    assert_eq!(r2.node_id, 2);
+    assert_eq!(r3.node_id, 3);
+}
+
+/// Dropping a receiver before merge does not prevent or corrupt the merge.
+/// The merge is a structural operation on RaftEvent variants; receiver
+/// validity is only checked at dispatch time (send()), not during merge.
+///
+/// Buffer before: [AE(prev=9, [10..14], rx dropped), AE(prev=14, [15..19])]
+///
+/// After merge:
+/// - buffer.len() == 1
+/// - senders.len() == 2  (dropped-receiver sender still present in Vec)
+/// - entries.len() == 10
+#[test]
+fn test_dropped_receiver_does_not_affect_merge() {
+    let (_graceful_tx, graceful_rx) = watch::channel(());
+    let mut raft = MockBuilder::new(graceful_rx).build_raft();
+    let mut queue = VecDeque::new();
+    for (rx_dropped, prev, count) in [(true, 9u64, 5), (false, 14, 5)] {
+        let request = make_request(5, prev, count, 1);
+        if rx_dropped {
+            let (event, receiver) = make_event(request);
+            drop(receiver);
+            queue.push_back(event);
+        } else {
+            let (event, _receiver) = make_event(request);
+            queue.push_back(event);
+        }
+    }
+    raft.buffered_raft_event = queue;
+
+    raft.merge_append_entries();
+    assert_eq!(raft.buffered_raft_event.len(), 1);
+
+    // front
+    match raft.buffered_raft_event.pop_front() {
+        Some(RaftEvent::AppendEntries(request, senders)) => {
+            let entries = request.entries;
+            assert_eq!(entries.len(), 10);
+            assert_eq!(senders.len(), 2);
+            assert_eq!(request.prev_log_index, 9);
+        }
+        _ => panic!("expected RaftEvent::AppendEntries, got something else"),
+    }
+}
+
+// -----------------------------------------------------------------------
+// Heartbeat position variants
+// -----------------------------------------------------------------------
+
+/// A heartbeat at the end of the sequence (not between two AEs) is absorbed.
+/// The merged result has entries from the preceding AE, the HB contributes
+/// only its sender and commit index via max().
+///
+/// Buffer before: [AE(prev=9, [10..14], commit=10),
+///                  HB(prev=14, entries=[], commit=50)]
+///
+/// After merge:
+/// - buffer.len() == 1
+/// - entries.len() == 5        (from AE only)
+/// - senders.len() == 2        (AE sender + HB sender)
+/// - leader_commit_index == 50 (HB's higher value wins via max)
+#[test]
+fn test_heartbeat_at_sequence_end_absorbed() {
+    let (_graceful_tx, graceful_rx) = watch::channel(());
+    let mut raft = MockBuilder::new(graceful_rx).build_raft();
+    let mut queue = VecDeque::new();
+    for (prev, count, commit) in [(9u64, 5, 10), (14, 0, 50)] {
+        let request = make_request(5, prev, count, commit);
+        let (event, _receiver) = make_event(request);
+        queue.push_back(event);
+    }
+    raft.buffered_raft_event = queue;
+
+    raft.merge_append_entries();
+    assert_eq!(raft.buffered_raft_event.len(), 1);
+
+    // front
+    match raft.buffered_raft_event.pop_front() {
+        Some(RaftEvent::AppendEntries(request, senders)) => {
+            let entries = request.entries;
+            assert_eq!(entries.len(), 5);
+            assert_eq!(senders.len(), 2);
+            assert_eq!(request.prev_log_index, 9);
+            assert_eq!(request.leader_commit_index, 50);
+        }
+        _ => panic!("expected RaftEvent::AppendEntries, got something else"),
+    }
+}
+
+/// Two consecutive heartbeats between two AEs are both absorbed.
+/// Each HB contributes its sender; commit index is the running max across all.
+///
+/// Buffer before: [AE(prev=9,  [10..14], commit=10),
+///                  HB(prev=14, [],       commit=20),
+///                  HB(prev=14, [],       commit=30),
+///                  AE(prev=14, [15..19], commit=10)]
+///
+/// After merge:
+/// - buffer.len() == 1
+/// - entries.len() == 10       (5 from first AE + 0 + 0 + 5 from last AE)
+/// - senders.len() == 4
+/// - leader_commit_index == 30 (max across all four)
+#[test]
+fn test_consecutive_heartbeats_absorbed() {
+    let (_graceful_tx, graceful_rx) = watch::channel(());
+    let mut raft = MockBuilder::new(graceful_rx).build_raft();
+    let mut queue = VecDeque::new();
+    // [AE(5 entries, commit=10), HB(commit=20), HB(commit=30), AE(5 entries, commit=10)]
+    for (prev, count, commit) in [(9u64, 5, 10), (14, 0, 20), (14, 0, 30), (14, 5, 10)] {
+        let request = make_request(5, prev, count, commit);
+        let (event, _receiver) = make_event(request);
+        queue.push_back(event);
+    }
+    raft.buffered_raft_event = queue;
+
+    raft.merge_append_entries();
+    assert_eq!(raft.buffered_raft_event.len(), 1);
+
+    match raft.buffered_raft_event.pop_front() {
+        Some(RaftEvent::AppendEntries(request, senders)) => {
+            assert_eq!(request.entries.len(), 10); // 5 + 0 + 0 + 5
+            assert_eq!(senders.len(), 4);
+            assert_eq!(request.prev_log_index, 9);
+            assert_eq!(request.leader_commit_index, 30); // max(10, 20, 30, 10)
+        }
+        _ => panic!("expected RaftEvent::AppendEntries, got something else"),
+    }
+}
+
+/// When all events in the buffer are heartbeats, they are merged into one
+/// empty-entries event with all senders collected and commit = max().
+///
+/// Buffer before: [HB(prev=9, [], commit=10),
+///                  HB(prev=9, [], commit=99)]
+///
+/// After merge:
+/// - buffer.len() == 1
+/// - entries.len() == 0
+/// - senders.len() == 2
+/// - leader_commit_index == 99
+#[test]
+fn test_all_heartbeats_merged_into_one() {
+    let (_graceful_tx, graceful_rx) = watch::channel(());
+    let mut raft = MockBuilder::new(graceful_rx).build_raft();
+    let mut queue = VecDeque::new();
+    for (prev, count, commit) in [(9u64, 0, 10), (9, 0, 99)] {
+        let request = make_request(5, prev, count, commit);
+        let (event, _receiver) = make_event(request);
+        queue.push_back(event);
+    }
+    raft.buffered_raft_event = queue;
+
+    raft.merge_append_entries();
+    assert_eq!(raft.buffered_raft_event.len(), 1);
+
+    // front
+    match raft.buffered_raft_event.pop_front() {
+        Some(RaftEvent::AppendEntries(request, senders)) => {
+            let entries = request.entries;
+            assert_eq!(entries.len(), 0);
+            assert_eq!(senders.len(), 2);
+            assert_eq!(request.prev_log_index, 9);
+            assert_eq!(request.leader_commit_index, 99);
+        }
+        _ => panic!("expected RaftEvent::AppendEntries, got something else"),
+    }
+}
+
+// -----------------------------------------------------------------------
+// Term ordering: newer-first
+// -----------------------------------------------------------------------
+
+/// When the buffer has a newer-term AE followed by an older-term AE
+/// (network reorder), the merge stops after the first event.
+/// The older-term AE is left at the back of the buffer untouched.
+///
+/// Rationale: merge only checks `term == first_req.term`; a lower term
+/// in the second slot fails that check and falls through to the `_ =>` arm.
+///
+/// Buffer before: [AE(term=6, prev=9,  [10..14]),
+///                  AE(term=5, prev=14, [15..19])]
+///
+/// After merge:
+/// - buffer.len() == 2
+/// - front: AE(term=6, entries=[10..14]) — processed alone
+/// - back:  AE(term=5, prev=14)         — untouched, will be rejected by follower
+#[test]
+fn test_newer_term_first_older_term_second_stops_merge() {
+    let (_graceful_tx, graceful_rx) = watch::channel(());
+    let mut raft = MockBuilder::new(graceful_rx).build_raft();
+    let mut queue = VecDeque::new();
+    for (prev, count, term) in [(9u64, 5, 6), (14, 5, 5)] {
+        let request = make_request(term, prev, count, 1);
+        let (event, _receiver) = make_event(request);
+        queue.push_back(event);
+    }
+    raft.buffered_raft_event = queue;
+
+    raft.merge_append_entries();
+    assert_eq!(raft.buffered_raft_event.len(), 2);
+
+    // front
+    match raft.buffered_raft_event.pop_front() {
+        Some(RaftEvent::AppendEntries(request, senders)) => {
+            let entries = request.entries;
+            assert_eq!(entries.len(), 5);
+            assert_eq!(senders.len(), 1);
+            assert_eq!(request.prev_log_index, 9);
+            assert_eq!(request.term, 6);
+        }
+        _ => panic!("expected RaftEvent::AppendEntries, got something else"),
+    }
+    // back
+    match raft.buffered_raft_event.pop_back() {
+        Some(RaftEvent::AppendEntries(request, senders)) => {
+            let entries = request.entries;
+            assert_eq!(entries.len(), 5);
+            assert_eq!(senders.len(), 1);
+            assert_eq!(request.prev_log_index, 14);
+            assert_eq!(request.term, 5);
+        }
+        _ => panic!("expected RaftEvent::AppendEntries, got something else"),
+    }
+}
+
+// -----------------------------------------------------------------------
+// max_merge_entries upper bound (requires config wiring)
+// -----------------------------------------------------------------------
+
+/// When the accumulated entry count would exceed `max_merge_entries`,
+/// the merge stops before the event that would breach the limit.
+/// The exceeding event is pushed back and remains as the next item.
+///
+/// Config: max_merge_entries = 8
+///
+/// Buffer before: [AE(prev=9,  [10..14], 5 entries),
+///                  AE(prev=14, [15..19], 5 entries)]   ← would make total 10 > 8
+///
+/// After merge:
+/// - buffer.len() == 2
+/// - front: AE with 5 entries  (first batch alone, did not exceed limit)
+/// - back:  AE with prev=14    (held back to respect the limit)
+#[test]
+fn test_max_merge_entries_limit_stops_merge() {
+    let (_graceful_tx, graceful_rx) = watch::channel(());
+    let mut node_config = RaftNodeConfig::new().expect("valid config");
+    node_config.raft.batching.max_merge_entries = 8;
+
+    let mut raft = MockBuilder::new(graceful_rx).with_node_config(node_config).build_raft();
+    let mut queue = VecDeque::new();
+    for (prev, count, term) in [(9u64, 5, 6), (14, 5, 6)] {
+        let request = make_request(term, prev, count, 1);
+        let (event, _receiver) = make_event(request);
+        queue.push_back(event);
+    }
+    raft.buffered_raft_event = queue;
+
+    raft.merge_append_entries();
+    assert_eq!(raft.buffered_raft_event.len(), 2);
+
+    // front
+    match raft.buffered_raft_event.pop_front() {
+        Some(RaftEvent::AppendEntries(request, senders)) => {
+            let entries = request.entries;
+            assert_eq!(entries.len(), 5);
+            assert_eq!(senders.len(), 1);
+            assert_eq!(request.prev_log_index, 9);
+            assert_eq!(request.term, 6);
+        }
+        _ => panic!("expected RaftEvent::AppendEntries, got something else"),
+    }
+    // back
+    match raft.buffered_raft_event.pop_back() {
+        Some(RaftEvent::AppendEntries(request, senders)) => {
+            let entries = request.entries;
+            assert_eq!(entries.len(), 5);
+            assert_eq!(senders.len(), 1);
+            assert_eq!(request.prev_log_index, 14);
+            assert_eq!(request.term, 6);
+        }
+        _ => panic!("expected RaftEvent::AppendEntries, got something else"),
+    }
+}
diff --git a/d-engine-core/src/raft_test/mod.rs b/d-engine-core/src/raft_test/mod.rs
index de00d540..2537adac 100644
--- a/d-engine-core/src/raft_test/mod.rs
+++ b/d-engine-core/src/raft_test/mod.rs
@@ -345,8 +345,3 @@
 //! - Fast: Complete suite should run in <10 seconds
 //! - Comprehensive: Cover normal path, error path, and edge cases
 //! - Clear: Test names and comments explain what is being tested and why
-
-mod drain_based_batch_architecture_tests;
-mod leader_change_tests;
-mod leader_discovered_tests;
-mod raft_comprehensive_tests;
diff --git a/d-engine-core/src/raft_test/process_raft_events_tests.rs b/d-engine-core/src/raft_test/process_raft_events_tests.rs
new file mode 100644
index 00000000..3ce4e8d8
--- /dev/null
+++ b/d-engine-core/src/raft_test/process_raft_events_tests.rs
@@ -0,0 +1,285 @@
+//! # process_raft_events drain loop tests
+//!
+//! Tests for the in-loop `matches!()` merge guard in `Raft::process_raft_events`.
+//! These tests verify that `merge_append_entries` fires per-iteration when the
+//! front of the queue is an `AppendEntries` event — not only once at the start
+//! of the loop (which was the pre-fix behavior flagged in CodeRabbit review #407).
+//!
+//! ## Assertion style
+//! The drain loop empties the buffer, so buffer length is not meaningful here.
+//! Dispatch counts are asserted via mock `times(N)`:
+//! - `handle_append_entries.times(1)` proves two consecutive AEs were merged.
+//! - `handle_append_entries.times(2)` would prove they were dispatched separately.
+//!   Mock expectations are verified on drop at the end of each test.
+
+use crate::AppendResponseWithUpdates;
+use crate::RoleEvent;
+use crate::StateUpdate;
+use crate::maybe_clone_oneshot::MaybeCloneOneshot;
+use crate::maybe_clone_oneshot::RaftOneshot;
+use crate::test_utils::MockBuilder;
+use crate::{MockElectionCore, MockReplicationCore, RaftEvent, mock_entries};
+use d_engine_proto::server::election::VoteRequest;
+use d_engine_proto::server::replication::{AppendEntriesRequest, AppendEntriesResponse};
+use std::collections::VecDeque;
+use tokio::sync::watch;
+
+// -----------------------------------------------------------------------
+// Helpers
+// -----------------------------------------------------------------------
+
+fn make_request(
+    term: u64,
+    prev_log_index: u64,
+    count: u64,
+    leader_commit: u64,
+) -> AppendEntriesRequest {
+    AppendEntriesRequest {
+        term,
+        leader_id: 1,
+        prev_log_index,
+        prev_log_term: term,
+        entries: mock_entries(1, count, term),
+        leader_commit_index: leader_commit,
+    }
+}
+
+fn make_ae_event(req: AppendEntriesRequest) -> RaftEvent {
+    let (tx, _rx) = MaybeCloneOneshot::new();
+    RaftEvent::AppendEntries(req, vec![tx])
+}
+
+fn make_fatal_event() -> RoleEvent {
+    RoleEvent::FatalError {
+        source: "core".to_string(),
+        error: "fatal-error".to_string(),
+    }
+}
+
+fn make_vr_event() -> RaftEvent {
+    let (tx, _rx) = MaybeCloneOneshot::new();
+    RaftEvent::ReceiveVoteRequest(
+        VoteRequest {
+            term: 1,
+            candidate_id: 2,
+            last_log_index: 11,
+            last_log_term: 3,
+        },
+        tx,
+    )
+}
+
+fn default_election_mock() -> MockElectionCore<crate::test_utils::MockTypeConfig> {
+    let mut m = MockElectionCore::new();
+    m.expect_broadcast_vote_requests().returning(|_, _, _, _, _| Ok(()));
+    m
+}
+
+fn ok_append_response() -> crate::AppendResponseWithUpdates {
+    AppendResponseWithUpdates {
+        response: AppendEntriesResponse::success(1, 5, None),
+        commit_index_update: None,
+    }
+}
+
+// -----------------------------------------------------------------------
+// Drain loop correctness
+// -----------------------------------------------------------------------
+
+/// A non-AE at the front does NOT block AE merging once it is drained.
+/// After the VoteRequest is dispatched, the subsequent AE run is merged.
+///
+/// Buffer before: [VoteRequest, AE(prev=9, 5 entries, term=5), AE(prev=14, 5 entries, term=5)]
+///
+/// Drain iterations:
+/// 1. front=VR  → merge guard skips → dispatch VR        → handle_vote_request ×1
+/// 2. front=AE  → merge guard fires → AE+AE → merged AE  → handle_append_entries ×1
+///
+/// times(1) on handle_append_entries proves merge happened (not ×2 separate dispatches).
+#[tokio::test]
+async fn test_non_ae_prefix_does_not_block_ae_merge_in_drain() {
+    let (_graceful_tx, graceful_rx) = watch::channel(());
+
+    let mut election_handler = default_election_mock();
+    election_handler.expect_handle_vote_request().times(1).returning(|_, _, _, _| {
+        Ok(StateUpdate {
+            new_voted_for: None,
+            term_update: None,
+        })
+    });
+
+    let mut replication_handler = MockReplicationCore::new();
+    replication_handler
+        .expect_handle_append_entries()
+        .times(1) // ← the assertion: 2 AEs merged → 1 dispatch
+        .returning(|_, _, _| Ok(ok_append_response()));
+
+    let mut raft = MockBuilder::new(graceful_rx)
+        .with_election_handler(election_handler)
+        .with_replication_handler(replication_handler)
+        .build_raft();
+
+    let mut queue = VecDeque::new();
+    queue.push_back(make_vr_event());
+    queue.push_back(make_ae_event(make_request(5, 9, 5, 1)));
+    queue.push_back(make_ae_event(make_request(5, 14, 5, 1)));
+    raft.buffered_raft_event = queue;
+
+    raft.process_raft_events().await.unwrap();
+    // mock drop verifies: handle_vote_request×1, handle_append_entries×1
+}
+
+/// When a non-AE splits two AE runs, BOTH runs are independently merged.
+///
+/// Buffer before: [AE(prev=9,  entries 10-14),
+///                 AE(prev=14, entries 15-19),
+///                 VoteRequest,
+///                 AE(prev=19, entries 20-24),
+///                 AE(prev=24, entries 25-29)]
+///
+/// Expected dispatches:
+/// - handle_append_entries × 2   ← each run merged separately, not ×4
+/// - handle_vote_request × 1
+#[tokio::test]
+async fn test_ae_runs_on_both_sides_of_non_ae_are_each_merged() {
+    let (_graceful_tx, graceful_rx) = watch::channel(());
+
+    let mut election_handler = default_election_mock();
+    election_handler.expect_handle_vote_request().times(1).returning(|_, _, _, _| {
+        Ok(StateUpdate {
+            new_voted_for: None,
+            term_update: None,
+        })
+    });
+
+    let mut replication_handler = MockReplicationCore::new();
+    replication_handler
+        .expect_handle_append_entries()
+        .times(2) // ← 2 runs × 1 merged dispatch each = 2 total (not 4)
+        .returning(|_, _, _| Ok(ok_append_response()));
+
+    let mut raft = MockBuilder::new(graceful_rx)
+        .with_election_handler(election_handler)
+        .with_replication_handler(replication_handler)
+        .build_raft();
+
+    let mut queue = VecDeque::new();
+    // first run: entries 10-19
+    queue.push_back(make_ae_event(make_request(5, 9, 5, 1)));
+    queue.push_back(make_ae_event(make_request(5, 14, 5, 1)));
+    queue.push_back(make_vr_event());
+    // second run: entries 20-29 (new entries, continues after first run)
+    queue.push_back(make_ae_event(make_request(5, 19, 5, 1)));
+    queue.push_back(make_ae_event(make_request(5, 24, 5, 1)));
+    raft.buffered_raft_event = queue;
+
+    raft.process_raft_events().await.unwrap();
+}
+
+/// Buffer with only non-AE events drains without calling merge and without panicking.
+///
+/// Buffer before: [VoteRequest, VoteRequest]
+///
+/// Expected dispatches:
+/// - handle_vote_request × 2
+#[tokio::test]
+async fn test_all_non_ae_drain_no_merge_no_panic() {
+    let (_graceful_tx, graceful_rx) = watch::channel(());
+
+    let mut election_handler = default_election_mock();
+    election_handler.expect_handle_vote_request().times(2).returning(|_, _, _, _| {
+        Ok(StateUpdate {
+            new_voted_for: None,
+            term_update: None,
+        })
+    });
+
+    let mut replication_handler = MockReplicationCore::new();
+    replication_handler
+        .expect_handle_append_entries()
+        .times(0) //  ← no AE events, merge guard never triggers
+        .returning(|_, _, _| Ok(ok_append_response()));
+
+    let mut raft = MockBuilder::new(graceful_rx)
+        .with_election_handler(election_handler)
+        .with_replication_handler(replication_handler)
+        .build_raft();
+
+    let mut queue = VecDeque::new();
+    queue.push_back(make_vr_event());
+    queue.push_back(make_vr_event());
+    raft.buffered_raft_event = queue;
+
+    raft.process_raft_events().await.unwrap();
+}
+
+/// A single AE sandwiched between non-AE events is dispatched alone.
+/// Merge fires but finds no contiguous AE neighbour to coalesce.
+///
+/// Buffer before: [AE(prev=9, 5 entries), VoteRequest, AE(prev=14, 5 entries)]
+///
+/// Expected dispatches:
+/// - handle_append_entries × 2   ← each AE alone, no merge opportunity
+/// - handle_vote_request × 1
+#[tokio::test]
+async fn test_isolated_ae_between_non_ae_dispatched_alone() {
+    let (_graceful_tx, graceful_rx) = watch::channel(());
+
+    let mut election_handler = default_election_mock();
+    election_handler.expect_handle_vote_request().times(1).returning(|_, _, _, _| {
+        Ok(StateUpdate {
+            new_voted_for: None,
+            term_update: None,
+        })
+    });
+
+    let mut replication_handler = MockReplicationCore::new();
+    replication_handler
+        .expect_handle_append_entries()
+        .times(2)
+        .returning(|_, _, _| Ok(ok_append_response()));
+
+    let mut raft = MockBuilder::new(graceful_rx)
+        .with_election_handler(election_handler)
+        .with_replication_handler(replication_handler)
+        .build_raft();
+
+    let mut queue = VecDeque::new();
+    // isolated AE: entries 10-14
+    queue.push_back(make_ae_event(make_request(5, 9, 5, 1)));
+    queue.push_back(make_vr_event());
+    // isolated AE: entries 15-19
+    queue.push_back(make_ae_event(make_request(5, 14, 5, 1)));
+    raft.buffered_raft_event = queue;
+
+    raft.process_raft_events().await.unwrap();
+}
+
+// -----------------------------------------------------------------------
+// process_role_events error propagation
+// -----------------------------------------------------------------------
+
+/// A RoleEvent::FatalError in the buffer causes process_role_events to
+/// return Err immediately rather than swallowing the error and continuing.
+///
+/// This guards against the node silently continuing after a fatal signal
+/// (e.g. state machine worker crash).
+///
+/// Buffer before: [RoleEvent::FatalError { source: "sm_worker", error: "disk full" }]
+///
+/// Expected:
+/// - returns Err(e) where e.is_fatal() == true
+/// - does NOT return Ok(())
+#[tokio::test]
+async fn test_process_role_events_propagates_fatal_error() {
+    let (_graceful_tx, graceful_rx) = watch::channel(());
+
+    let mut raft = MockBuilder::new(graceful_rx).build_raft();
+
+    let mut queue = VecDeque::new();
+    queue.push_back(make_fatal_event());
+    raft.buffered_role_event = queue;
+
+    let err = raft.process_role_events().await.unwrap_err();
+    assert!(err.is_fatal());
+}
diff --git a/d-engine-core/src/storage/buffered_raft_log.rs b/d-engine-core/src/storage/buffered_raft_log.rs
index 62a54773..ec986ceb 100644
--- a/d-engine-core/src/storage/buffered_raft_log.rs
+++ b/d-engine-core/src/storage/buffered_raft_log.rs
@@ -1147,10 +1147,12 @@ where
         }
     }
 
-    /// Advance durable_index to `max_index` after data has been written to OS page cache (Level 2).
+    /// Advance durable_index to `max_index` after ensuring WAL durability.
     ///
-    /// For Level 2 (current): `is_write_durable=true` → skip flush_wal, advance immediately.
-    /// For Level 3 (future):  `is_write_durable=false` → call flush_wal(sync=true) first.
+    /// Batched Level 3 (current): `is_write_durable=false` → flush_wal(sync=true) first,
+    ///   coalescing multiple db.write() calls into a single fdatasync before notifying Raft.
+    /// Per-write durable mode:    `is_write_durable=true`  → skip flush_wal, advance immediately.
+    ///   (backend guarantees each write() is already durable, e.g. write_options.sync=true)
     async fn advance_durable_after_write(
         &self,
         max_index: u64,
diff --git a/d-engine-server/src/network/grpc/grpc_raft_service.rs b/d-engine-server/src/network/grpc/grpc_raft_service.rs
index 3e673064..b34cad59 100644
--- a/d-engine-server/src/network/grpc/grpc_raft_service.rs
+++ b/d-engine-server/src/network/grpc/grpc_raft_service.rs
@@ -116,7 +116,10 @@ where
 
         let (resp_tx, resp_rx) = MaybeCloneOneshot::new();
         self.event_tx
-            .send(RaftEvent::AppendEntries(request.into_inner(), resp_tx))
+            .send(RaftEvent::AppendEntries(
+                request.into_inner(),
+                vec![resp_tx],
+            ))
             .await
             .map_err(|_| Status::internal("Event channel closed"))?;
 
@@ -177,7 +180,7 @@ where
                         match result {
                             Some(Ok(req)) => {
                                 let (resp_tx, resp_rx) = MaybeCloneOneshot::new();
-                                if event_tx.send(RaftEvent::AppendEntries(req, resp_tx)).await.is_err() {
+                                if event_tx.send(RaftEvent::AppendEntries(req, vec![resp_tx])).await.is_err() {
                                     warn!("[stream_append_entries|recv] event_tx closed");
                                     break;
                                 }
diff --git a/d-engine-server/src/storage/adaptors/rocksdb/rocksdb_storage_engine.rs b/d-engine-server/src/storage/adaptors/rocksdb/rocksdb_storage_engine.rs
index b5ce36ad..5bcbd42d 100644
--- a/d-engine-server/src/storage/adaptors/rocksdb/rocksdb_storage_engine.rs
+++ b/d-engine-server/src/storage/adaptors/rocksdb/rocksdb_storage_engine.rs
@@ -404,14 +404,17 @@ impl LogStore for RocksDBLogStore {
     }
 
     fn is_write_durable(&self) -> bool {
-        // Level 2 semantics: db.write() lands in OS page cache (not RocksDB internal buffer).
+        // Batched Level 3: db.write() uses sync=false (lands in OS page cache),
+        // but durable_index does NOT advance until flush_wal(true) (fdatasync) completes.
+        //
         // Process crash: OS retains page cache → WAL replay on restart → full recovery. ✅
-        // Power loss:    OS page cache lost → data unrecoverable. ❌  (intentional trade-off)
+        // Power loss:    committed data (durable_index advanced) has been fdatasync'd → safe. ✅
+        //                Only entries in the current unflushed batch are at risk — a small window
+        //                bounded by idle_flush_interval_ms, and not yet counted toward quorum.
         //
-        // Returns true → advance_durable_and_notify fires immediately after persist_entries,
-        // without waiting for an explicit flush_wal(true) (Level 3 / fdatasync).
-        // Level 3 support (sync thread + flush_wal) is tracked as a future feature.
-        true
+        // Returns false → advance_durable_after_write calls flush_wal(true) before notifying,
+        // coalescing multiple db.write() calls into a single fdatasync (batched Level 3).
+        false
     }
 
     fn load_purge_boundary(&self) -> Result<Option<LogId>, Error> {
diff --git a/d-engine/src/docs/performance/throughput-optimization-guide.md b/d-engine/src/docs/performance/throughput-optimization-guide.md
index 6cc9f83c..fac639db 100644
--- a/d-engine/src/docs/performance/throughput-optimization-guide.md
+++ b/d-engine/src/docs/performance/throughput-optimization-guide.md
@@ -23,7 +23,7 @@ pub(crate) enum ConnectionType {
 
 ## Persistence Strategy & Throughput/Latency Trade-offs
 
-`MemFirst` is the only persistence strategy in v0.2.4+. It writes to OS page cache (process-crash safe, not power-loss safe) and flushes asynchronously.
+`MemFirst` is the only persistence strategy in v0.2.4+. It batches writes to OS page cache and flushes with fsync asynchronously — committing data to disk before notifying Raft.
 
 ### Strategy Configuration
 
@@ -35,17 +35,28 @@ flush_policy = { Batch = { idle_flush_interval_ms = 1000 } }
 
 ### `MemFirst` Strategy
 
-- **Write Path**: Entry written to OS page cache immediately; IO thread flushes asynchronously.
-- **Durability**: Process crash safe (OS page cache survives). Power loss may drop the most recent unflushed entries.
-- **Throughput**: High. Writes batch naturally; no per-write fsync.
+- **Write Path**: Entries are written to OS page cache via `db.write()` / `file.write()`; the IO
+  thread batches them and calls fsync (`flush_wal(true)` / `sync_all()`) before advancing
+  `durable_index`. Raft only counts an entry toward quorum after fsync completes.
+- **Durability**:
+  - *Process crash*: OS page cache survives a process restart → full recovery via WAL replay. ✅
+  - *Power loss (single node)*: In a multi-node cluster, Raft quorum ensures committed data
+    survives a single-node power failure — the other quorum members retain the data. ✅
+  - *Power loss (majority of nodes simultaneously)*: Entries in the current unflushed batch
+    (written since the last fsync) may be lost. This window is bounded by
+    `idle_flush_interval_ms`. Raft has not yet counted these entries toward quorum, so no
+    client-acknowledged write is lost — the leader will re-replicate after re-election. ⚠️
+- **Throughput**: High. Multiple writes share a single fsync; no per-write fsync overhead.
 
 ### `FlushPolicy` Tuning
 
-- **`Batch { idle_flush_interval_ms }`**: Flush after this many milliseconds of idle time.
-  - Lower values reduce data loss window but increase IO pressure.
+- **`Batch { idle_flush_interval_ms }`**: Flush (fsync) after this many milliseconds of idle time.
+  - Lower values reduce the unflushed batch window but increase IO pressure.
   - Default `1000` ms is suitable for most workloads.
 
-> **Note**: `DiskFirst` strategy was removed in v0.2.4. `MemFirst` writes to OS page cache and is process-crash safe but **not power-loss safe** — `idle_flush_interval_ms` controls flush frequency but does not provide fsync-level durability.
+> **Note**: `DiskFirst` strategy was removed in v0.2.4. `MemFirst` replaced it with batched
+> fsync — multiple writes share one fsync call, reducing IO overhead while still providing
+> disk-level durability for all client-acknowledged (committed) writes.
 
 ## Batching Configuration