fix #413: move internal events from bounded P4 event_tx to unbounded P2 role_tx — eliminate deadlock risk

CHANGELOG.md

@@ -33,7 +33,7 @@ All notable changes to this project will be documented in this file.
 - **`scan_prefix` API** (#378): `ClientApi::scan_prefix(prefix)` returns all KV pairs matching a prefix in a single read — intended for zero-race-window state re-sync after watch reconnection.
 
 - **Async IO architecture — pipeline replication** (#334, #341, #342, #343, #345, #349, #350, #351):
-  The Raft event loop is now fully non-blocking under write load
+  The Inbound event loop is now fully non-blocking under write load
   - `BufferedRaftLog` runs on a dedicated OS thread — WAL writes never steal tokio worker threads
   - `StateMachineWorker::apply_batch` is fully async — RocksDB apply no longer blocks the event loop
   - AppendEntries uses a **persistent bidirectional gRPC stream per peer** — eliminates per-batch connection overhead
@@ -78,7 +78,7 @@ All notable changes to this project will be documented in this file.
   - Eliminates election churn when recovering or newly-joined nodes have lagged logs
   - Aligns with Raft §5.2 (term check takes priority over log matching)
 
-- **fix(snapshot) #315**: Snapshot disk I/O isolated from Raft event loop via `spawn_blocking`
+- **fix(snapshot) #315**: Snapshot disk I/O isolated from Inbound event loop via `spawn_blocking`
   - Previously, snapshot transfer competed with consensus and caused `ProposeFailed` (4006) under load
 
 - **fix(snapshot) #308**: Snapshot install success is now driven by the follower's apply confirmation, not the transfer ACK

README.md

@@ -22,7 +22,7 @@ and pluggable storage backends. Start with one node, scale to a cluster when nee
 
 ## New in v0.2.4 🎉
 
-- **Async IO Architecture**: Raft event loop is fully non-blocking — WAL writes, state machine apply, and replication all run off the hot path. AppendEntries uses a persistent bidirectional stream per peer; replication is pipelined across followers.
+- **Async IO Architecture**: Inbound event loop is fully non-blocking — WAL writes, state machine apply, and replication all run off the hot path. AppendEntries uses a persistent bidirectional stream per peer; replication is pipelined across followers.
 - **Cluster Membership Streaming**: `EmbeddedEngine::watch_membership()` / `GrpcClient::watch_membership()` — subscribe to real-time membership changes in both embedded and standalone modes
 - **Simpler Startup**: `EmbeddedEngine::start(data_dir)` and `StandaloneEngine::run(data_dir, shutdown_rx)` — no config file required for common cases
 - **Jepsen Validated**: 5 workloads + 6-hour soak test under combined kill/partition/pause faults — see [Correctness Guarantees](https://github.com/deventlab/d-engine-jepsen/blob/main/GUARANTEES.md)

d-engine-core/benches/leader_state_bench.rs

@@ -112,6 +112,7 @@ use std::time::Duration;
 
 use criterion::{BenchmarkId, Criterion, black_box, criterion_group, criterion_main};
 use d_engine_core::leader_state::{LeaderState, PendingPromotion};
+use d_engine_core::role_state::RaftRoleState;
 use d_engine_core::{
     MockElectionCore, MockMembership, MockPurgeExecutor, MockRaftLog, MockReplicationCore,
     MockStateMachine, MockStateMachineHandler, MockTransport, MockTypeConfig, RaftContext,
@@ -149,7 +150,7 @@ struct BenchFixture {
     leader_state: LeaderState<MockTypeConfig>,
     raft_context: RaftContext<MockTypeConfig>,
     #[allow(dead_code)]
-    role_tx: mpsc::UnboundedSender<d_engine_core::RoleEvent>,
+    internal_event_tx: mpsc::UnboundedSender<d_engine_core::InternalEvent>,
 }
 
 impl BenchFixture {
@@ -256,13 +257,13 @@ impl BenchFixture {
             node_config: Arc::new(node_config.clone()),
         };
 
-        let (role_tx, _role_rx) = mpsc::unbounded_channel();
+        let (internal_event_tx, _internal_event_rx) = mpsc::unbounded_channel();
         let leader_state = LeaderState::new(1, Arc::new(node_config));
 
         BenchFixture {
             leader_state,
             raft_context,
-            role_tx,
+            internal_event_tx,
         }
     }
 }
@@ -286,7 +287,10 @@ fn bench_process_promotions_2_nodes(c: &mut Criterion) {
             black_box(
                 fixture
                     .leader_state
-                    .process_pending_promotions(&fixture.raft_context, &fixture.role_tx)
+                    .handle_promote_ready_learners(
+                        &fixture.raft_context,
+                        &fixture.internal_event_tx,
+                    )
                     .await,
             )
         })
@@ -319,7 +323,10 @@ fn bench_batch_promotion_scaling(c: &mut Criterion) {
                     black_box(
                         fixture
                             .leader_state
-                            .process_pending_promotions(&fixture.raft_context, &fixture.role_tx)
+                            .handle_promote_ready_learners(
+                                &fixture.raft_context,
+                                &fixture.internal_event_tx,
+                            )
                             .await,
                     )
                 })

d-engine-core/src/commit_handler/default_commit_handler.rs

@@ -11,9 +11,9 @@ use tracing::trace;
 use tracing::warn;
 
 use super::CommitHandler;
+use crate::InternalEvent;
 use crate::Membership;
 use crate::NewCommitData;
-use crate::RaftEvent;
 use crate::RaftLog;
 use crate::Result;
 use crate::StateMachineHandler;
@@ -28,7 +28,7 @@ pub struct CommitHandlerDependencies<T: TypeConfig> {
     pub state_machine_handler: Arc<SMHOF<T>>,
     pub raft_log: Arc<ROF<T>>,
     pub membership: Arc<MOF<T>>,
-    pub event_tx: mpsc::Sender<RaftEvent>,
+    pub internal_event_tx: mpsc::UnboundedSender<InternalEvent>,
     pub sm_apply_tx: mpsc::UnboundedSender<Vec<Entry>>,
     pub shutdown_signal: watch::Receiver<()>,
     pub max_batch_size: usize,
@@ -47,8 +47,8 @@ where
     new_commit_rx: Option<mpsc::UnboundedReceiver<NewCommitData>>,
     membership: Arc<MOF<T>>,
 
-    event_tx: mpsc::Sender<RaftEvent>, // Cloned from Raft
-    sm_apply_tx: mpsc::UnboundedSender<Vec<Entry>>, // Send entries to SM Worker
+    internal_event_tx: mpsc::UnboundedSender<InternalEvent>, // Cloned from Raft
+    sm_apply_tx: mpsc::UnboundedSender<Vec<Entry>>,          // Send entries to SM Worker
 
     // Shutdown signal
     shutdown_signal: watch::Receiver<()>,
@@ -130,7 +130,7 @@ where
             raft_log: deps.raft_log,
             membership: deps.membership,
             new_commit_rx: Some(new_commit_rx),
-            event_tx: deps.event_tx,
+            internal_event_tx: deps.internal_event_tx,
             sm_apply_tx: deps.sm_apply_tx,
             shutdown_signal: deps.shutdown_signal,
             max_batch_size: deps.max_batch_size,
@@ -198,7 +198,7 @@ where
             self.send_to_sm_worker(&mut command_batch).await?;
         }
 
-        // Snapshot check moved to ApplyCompleted handler in Raft event loop.
+        // Snapshot check moved to ApplyCompleted handler in inbound event loop.
         // SM Worker applies entries asynchronously, so last_applied is stale here.
 
         Ok(())
@@ -254,7 +254,7 @@ where
 
             // 2.5. Notify leader to refresh cluster metadata cache
             // This must happen AFTER membership is applied
-            if let Err(e) = self.event_tx.send(RaftEvent::MembershipApplied).await {
+            if let Err(e) = self.internal_event_tx.send(InternalEvent::MembershipApplied) {
                 warn!("Failed to send MembershipApplied event: {:?}", e);
             }
 
@@ -265,7 +265,7 @@ where
                     self.my_id, entry.index
                 );
                 // Signal step down - error is non-fatal as removal is already committed
-                if let Err(e) = self.event_tx.send(RaftEvent::StepDownSelfRemoved).await {
+                if let Err(e) = self.internal_event_tx.send(InternalEvent::StepDownSelfRemoved) {
                     error!(
                         "[{}] Failed to send StepDownSelfRemoved event: {:?}",
                         self.my_id, e
@@ -299,3 +299,7 @@ where
         Ok(())
     }
 }
+
+#[cfg(test)]
+#[path = "default_commit_handler_test.rs"]
+mod default_commit_handler_test;

d-engine-core/src/commit_handler/default_commit_handler_test.rs

@@ -1,11 +1,10 @@
-use std::sync::Arc;
-use std::time::Duration;
-
 use bytes::Bytes;
 use d_engine_proto::common::Entry;
 use d_engine_proto::common::EntryPayload;
 use d_engine_proto::common::NodeRole::Leader;
 use d_engine_proto::common::membership_change::Change;
+use std::sync::Arc;
+use std::time::Duration;
 use tokio::sync::mpsc::{self};
 use tokio::sync::watch;
 use tokio::task::JoinHandle;
@@ -15,12 +14,12 @@ use super::CommitHandler;
 use super::CommitHandlerDependencies;
 use super::DefaultCommitHandler;
 use crate::Error;
+use crate::InternalEvent;
 use crate::MockMembership;
 use crate::MockRaftLog;
 use crate::MockStateMachineHandler;
 use crate::MockTypeConfig;
 use crate::NewCommitData;
-use crate::RaftEvent;
 use crate::Result;
 
 const TEST_TERM: u64 = 1;
@@ -69,8 +68,8 @@ pub struct TestHarness {
     mock_membership: Arc<MockMembership<MockTypeConfig>>,
     commit_tx: mpsc::UnboundedSender<NewCommitData>,
     commit_rx: Option<mpsc::UnboundedReceiver<NewCommitData>>,
-    event_tx: mpsc::Sender<RaftEvent>,
-    event_rx: mpsc::Receiver<RaftEvent>,
+    internal_event_tx: mpsc::UnboundedSender<InternalEvent>,
+    internal_event_rx: mpsc::UnboundedReceiver<InternalEvent>,
     shutdown_tx: watch::Sender<()>,
     shutdown_rx: Option<watch::Receiver<()>>,
     handle: Option<JoinHandle<()>>,
@@ -91,7 +90,7 @@ where
 {
     let (commit_tx, commit_rx) = mpsc::unbounded_channel();
     let (shutdown_tx, shutdown_rx) = watch::channel(());
-    let (event_tx, event_rx) = mpsc::channel(1000); // Large capacity to prevent blocking in tests
+    let (internal_event_tx, internal_event_rx) = mpsc::unbounded_channel(); // Large capacity to prevent blocking in tests
 
     // Mock state machine
     let mut mock_smh = MockStateMachineHandler::new();
@@ -135,8 +134,8 @@ where
         mock_membership: Arc::new(mock_membership),
         commit_rx: Some(commit_rx),
         commit_tx,
-        event_tx,
-        event_rx,
+        internal_event_tx,
+        internal_event_rx,
         shutdown_tx,
         shutdown_rx: Some(shutdown_rx),
         handle: None,
@@ -151,7 +150,7 @@ impl TestHarness {
             state_machine_handler: self.mock_smh.clone(),
             raft_log: self.mock_log.clone(),
             membership: self.mock_membership.clone(),
-            event_tx: self.event_tx.clone(),
+            internal_event_tx: self.internal_event_tx.clone(),
             sm_apply_tx,
             shutdown_signal: self.shutdown_rx.take().unwrap(),
             max_batch_size: 10,
@@ -186,7 +185,7 @@ impl TestHarness {
             state_machine_handler: self.mock_smh.clone(),
             raft_log: self.mock_log.clone(),
             membership: self.mock_membership.clone(),
-            event_tx: self.event_tx.clone(),
+            internal_event_tx: self.internal_event_tx.clone(),
             sm_apply_tx,
             shutdown_signal: self.shutdown_rx.take().unwrap(),
             max_batch_size: 10,
@@ -225,7 +224,7 @@ impl TestHarness {
     }
 
     async fn expect_snapshot_trigger(&mut self) -> bool {
-        match time::timeout(Duration::from_millis(50), self.event_rx.recv()).await {
+        match time::timeout(Duration::from_millis(50), self.internal_event_rx.recv()).await {
             Ok(Some(_)) => true, // Event received normally
             Ok(None) => false,   // Channel closed
             Err(_) => false,     // Timeout
@@ -651,7 +650,7 @@ mod process_batch_test {
     //     snapshot_condition: Option<u64>,
     // ) -> (
     //     DefaultCommitHandler<MockTypeConfig>,
-    //     mpsc::Receiver<RaftEvent>,
+    //     mpsc::Receiver<InternalEvent>,
     //     watch::Sender<()>,
     // ) where
     //     F: Fn() -> bool + 'static + Send + Sync,
@@ -911,7 +910,7 @@ mod process_batch_test {
     /// Test 1: MembershipApplied event MUST be sent after successful config change
     ///
     /// Verifies that when apply_config_change() succeeds, the commit handler
-    /// sends RaftEvent::MembershipApplied to notify Leader to refresh cache.
+    /// sends InternalEvent::MembershipApplied to notify Leader to refresh cache.
     ///
     /// Related: Bug fix #209 - cluster metadata cache timing issue
     #[tokio::test]
@@ -940,8 +939,10 @@ mod process_batch_test {
         assert!(result.is_ok(), "Config change should succeed");
 
         // Verify MembershipApplied event was sent
-        match tokio::time::timeout(Duration::from_millis(50), harness.event_rx.recv()).await {
-            Ok(Some(RaftEvent::MembershipApplied)) => {
+        match tokio::time::timeout(Duration::from_millis(50), harness.internal_event_rx.recv())
+            .await
+        {
+            Ok(Some(InternalEvent::MembershipApplied)) => {
                 // Success - event received
             }
             Ok(Some(other)) => panic!("Expected MembershipApplied, got {other:?}"),
@@ -994,8 +995,10 @@ mod process_batch_test {
         harness.process_batch_handler().await.unwrap();
 
         // Verify event received
-        match tokio::time::timeout(Duration::from_millis(50), harness.event_rx.recv()).await {
-            Ok(Some(RaftEvent::MembershipApplied)) => {
+        match tokio::time::timeout(Duration::from_millis(50), harness.internal_event_rx.recv())
+            .await
+        {
+            Ok(Some(InternalEvent::MembershipApplied)) => {
                 // Record event reception
                 apply_order.lock().push("MembershipApplied_event");
             }
@@ -1043,7 +1046,9 @@ mod process_batch_test {
         assert!(result.is_err(), "Config change should fail");
 
         // Verify NO event was sent
-        match tokio::time::timeout(Duration::from_millis(50), harness.event_rx.recv()).await {
+        match tokio::time::timeout(Duration::from_millis(50), harness.internal_event_rx.recv())
+            .await
+        {
             Err(_) => {
                 // Timeout is expected - no event sent
             }
@@ -1077,8 +1082,9 @@ mod process_batch_test {
         harness.process_batch_handler().await.unwrap();
         assert!(
             matches!(
-                tokio::time::timeout(Duration::from_millis(50), harness.event_rx.recv()).await,
-                Ok(Some(RaftEvent::MembershipApplied))
+                tokio::time::timeout(Duration::from_millis(50), harness.internal_event_rx.recv())
+                    .await,
+                Ok(Some(InternalEvent::MembershipApplied))
             ),
             "AddNode should send MembershipApplied"
         );
@@ -1094,8 +1100,9 @@ mod process_batch_test {
         harness.process_batch_handler().await.unwrap();
         assert!(
             matches!(
-                tokio::time::timeout(Duration::from_millis(50), harness.event_rx.recv()).await,
-                Ok(Some(RaftEvent::MembershipApplied))
+                tokio::time::timeout(Duration::from_millis(50), harness.internal_event_rx.recv())
+                    .await,
+                Ok(Some(InternalEvent::MembershipApplied))
             ),
             "RemoveNode should send MembershipApplied"
         );
@@ -1114,8 +1121,9 @@ mod process_batch_test {
         harness.process_batch_handler().await.unwrap();
         assert!(
             matches!(
-                tokio::time::timeout(Duration::from_millis(50), harness.event_rx.recv()).await,
-                Ok(Some(RaftEvent::MembershipApplied))
+                tokio::time::timeout(Duration::from_millis(50), harness.internal_event_rx.recv())
+                    .await,
+                Ok(Some(InternalEvent::MembershipApplied))
             ),
             "PromoteLearner should send MembershipApplied"
         );
@@ -1147,8 +1155,8 @@ mod process_batch_test {
 
         // Replace the receiver with a dummy so the original event_tx sends will fail.
         // Swap old_rx out of the harness so harness is still usable, then drop old_rx.
-        let (_dummy_tx, dummy_rx) = mpsc::channel::<RaftEvent>(1);
-        let old_rx = std::mem::replace(&mut harness.event_rx, dummy_rx);
+        let (_dummy_tx, dummy_rx) = mpsc::unbounded_channel::<InternalEvent>();
+        let old_rx = std::mem::replace(&mut harness.internal_event_rx, dummy_rx);
         drop(old_rx); // Now harness.event_tx's peer receiver is gone → sends fail.
 
         let result = harness.process_batch_handler().await;
@@ -1182,8 +1190,8 @@ mod process_batch_test {
         );
 
         // Replace receiver with a dummy so both sends fail.
-        let (_dummy_tx, dummy_rx) = mpsc::channel::<RaftEvent>(1);
-        let old_rx = std::mem::replace(&mut harness.event_rx, dummy_rx);
+        let (_dummy_tx, dummy_rx) = mpsc::unbounded_channel::<InternalEvent>();
+        let old_rx = std::mem::replace(&mut harness.internal_event_rx, dummy_rx);
         drop(old_rx);
 
         let result = harness.process_batch_handler().await;
@@ -1226,17 +1234,21 @@ mod process_batch_test {
         harness.process_batch_handler().await.unwrap();
 
         // First event should be MembershipApplied
-        match tokio::time::timeout(Duration::from_millis(50), harness.event_rx.recv()).await {
-            Ok(Some(RaftEvent::MembershipApplied)) => {
+        match tokio::time::timeout(Duration::from_millis(50), harness.internal_event_rx.recv())
+            .await
+        {
+            Ok(Some(InternalEvent::MembershipApplied)) => {
                 // Correct first event
             }
             Ok(Some(other)) => panic!("Expected MembershipApplied first, got {other:?}"),
             _ => panic!("Expected MembershipApplied event"),
         }
 
         // Second event should be StepDownSelfRemoved
-        match tokio::time::timeout(Duration::from_millis(50), harness.event_rx.recv()).await {
-            Ok(Some(RaftEvent::StepDownSelfRemoved)) => {
+        match tokio::time::timeout(Duration::from_millis(50), harness.internal_event_rx.recv())
+            .await
+        {
+            Ok(Some(InternalEvent::StepDownSelfRemoved)) => {
                 // Correct second event
             }
             Ok(Some(other)) => panic!("Expected StepDownSelfRemoved second, got {other:?}"),