feat: add mempool integration with execution sync and error isolation

Author: qj0r9j0vc2

crates/execution/src/engine.rs

@@ -5,7 +5,7 @@
 
 use crate::{
     database::{CipherBftDatabase, Provider},
-    error::{ExecutionError, Result},
+    error::{ExecutionError, Result, TxErrorCategory},
     evm::CipherBftEvmConfig,
     precompiles::{GenesisValidatorData, StakingPrecompile},
     receipts::{
@@ -353,7 +353,26 @@ impl<P: Provider + Clone> ExecutionEngine<P> {
                 let tx_start = Instant::now();
 
                 // Execute transaction
-                let tx_result = self.evm_config.execute_transaction(&mut evm, tx_bytes)?;
+                let tx_result = match self.evm_config.execute_transaction(&mut evm, tx_bytes) {
+                    Ok(result) => result,
+                    Err(e) => match e.category() {
+                        TxErrorCategory::Skip { reason } => {
+                            tracing::warn!(
+                                tx_index,
+                                ?reason,
+                                "Skipping invalid transaction (mempool should catch this)"
+                            );
+                            continue;
+                        }
+                        TxErrorCategory::FailedReceipt => {
+                            tracing::warn!(tx_index, error = %e, "Transaction reverted, skipping");
+                            continue;
+                        }
+                        TxErrorCategory::Fatal => {
+                            return Err(e);
+                        }
+                    },
+                };
 
                 // Record per-transaction metrics
                 let tx_duration = tx_start.elapsed();
@@ -947,4 +966,42 @@ mod tests {
         // Verify beneficiary is set in sealed block header
         assert_eq!(sealed.header.beneficiary, beneficiary);
     }
+
+    #[test]
+    fn test_error_isolation_skips_invalid_transactions() {
+        // This test documents the expected behavior of error isolation.
+        //
+        // When a block contains transactions with invalid nonces (NonceTooHigh, NonceTooLow),
+        // insufficient balance, or other validation errors, the execution engine should:
+        //
+        // 1. Skip the invalid transaction (no receipt generated)
+        // 2. Continue processing remaining transactions in the block
+        // 3. Return success with results from valid transactions only
+        //
+        // This prevents a single bad transaction from failing an entire block,
+        // which would cause execution-consensus divergence.
+        //
+        // Implementation note: The actual error isolation is implemented in
+        // execute_block() which handles TxErrorCategory::Skip by continuing
+        // the transaction loop instead of returning an error.
+        //
+        // Testing this end-to-end requires creating signed transactions with
+        // specific nonces, which requires test infrastructure for:
+        // - Generating valid ECDSA signatures
+        // - Setting up account state with specific nonces
+        // - Creating transactions that will trigger NonceTooHigh
+        //
+        // For now, this test documents the expected behavior.
+        // Full integration testing is done via devnet MassMint scenarios.
+
+        let engine = create_test_engine();
+
+        // Verify engine is created correctly
+        assert_eq!(engine.chain_config.chain_id, 85300);
+
+        // The actual error isolation behavior is tested by:
+        // 1. Running the devnet with MassMint script
+        // 2. Verifying blocks execute even when some transactions have wrong nonces
+        // 3. Checking that valid transactions are included and executed
+    }
 }

crates/execution/src/error.rs

@@ -143,3 +143,59 @@ impl DatabaseError {
 }
 
 impl DBErrorMarker for DatabaseError {}
+
+/// Categorizes transaction execution errors for handling decisions.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum TxErrorCategory {
+    /// Skip transaction, continue block execution.
+    Skip {
+        /// The reason for skipping this transaction.
+        reason: SkipReason,
+    },
+    /// Include transaction with failed receipt (EVM revert).
+    FailedReceipt,
+    /// Fatal error - halt block execution.
+    Fatal,
+}
+
+/// Reason for skipping a transaction.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum SkipReason {
+    /// Transaction nonce is higher than expected.
+    NonceTooHigh,
+    /// Transaction nonce is lower than expected (already executed).
+    NonceTooLow,
+    /// Account has insufficient balance for transaction.
+    InsufficientBalance,
+    /// Transaction failed basic validation.
+    InvalidTransaction,
+}
+
+impl ExecutionError {
+    /// Categorize this error for handling decision.
+    pub fn category(&self) -> TxErrorCategory {
+        let error_str = format!("{:?}", self);
+
+        if error_str.contains("NonceTooHigh") {
+            TxErrorCategory::Skip {
+                reason: SkipReason::NonceTooHigh,
+            }
+        } else if error_str.contains("NonceTooLow") {
+            TxErrorCategory::Skip {
+                reason: SkipReason::NonceTooLow,
+            }
+        } else if error_str.contains("InsufficientFunds") || error_str.contains("insufficient") {
+            TxErrorCategory::Skip {
+                reason: SkipReason::InsufficientBalance,
+            }
+        } else if matches!(self, ExecutionError::InvalidTransaction(_)) {
+            TxErrorCategory::Skip {
+                reason: SkipReason::InvalidTransaction,
+            }
+        } else if matches!(self, ExecutionError::Evm(_)) {
+            TxErrorCategory::FailedReceipt
+        } else {
+            TxErrorCategory::Fatal
+        }
+    }
+}

crates/node/Cargo.toml

@@ -26,11 +26,15 @@ cipherbft-metrics = { path = "../metrics" }
 alloy-primitives = { version = "1", features = ["serde"] }
 alloy-rlp = { workspace = true }
 alloy-consensus = { workspace = true }
+alloy-eips = { version = "1" }
 
 # Reth primitives (for transaction parsing in block execution)
 reth-primitives = { workspace = true }
 reth-primitives-traits = { workspace = true }
 
+# Reth transaction pool (for WorkerPoolAdapter)
+reth-transaction-pool = { git = "https://github.com/paradigmxyz/reth", tag = "v1.10.0" }
+
 # Async runtime
 tokio = { workspace = true, features = ["full", "signal"] }
 tokio-util = { workspace = true }

crates/node/src/execution_bridge.rs

@@ -787,7 +787,7 @@ mod tests {
 
     #[tokio::test]
     async fn test_set_genesis_block_hash() {
-        let bridge = create_default_bridge().unwrap();
+        let (bridge, _temp_dir) = create_default_bridge().unwrap();
 
         // Initially should be B256::ZERO
         let initial_hash = bridge.last_block_hash.read().map(|guard| *guard).unwrap();

crates/node/src/execution_sync.rs

@@ -0,0 +1,137 @@
+//! Execution-Consensus synchronization tracking.
+
+use std::sync::atomic::{AtomicU32, AtomicU64, Ordering};
+use tracing::error;
+
+/// Configuration for execution sync tracking.
+///
+/// Note: Named `ExecutionSyncConfig` to avoid collision with
+/// `cipherbft_consensus::ExecutionSyncConfig`.
+#[derive(Clone, Debug)]
+pub struct ExecutionSyncConfig {
+    /// Maximum blocks execution can fall behind before halting.
+    pub max_divergence: u64,
+    /// Maximum consecutive failures before halting.
+    pub max_consecutive_failures: u32,
+}
+
+impl Default for ExecutionSyncConfig {
+    fn default() -> Self {
+        Self {
+            max_divergence: 10,
+            max_consecutive_failures: 5,
+        }
+    }
+}
+
+/// Action to take after execution failure.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum SyncAction {
+    Continue,
+    Halt { reason: String },
+}
+
+/// Tracks execution progress relative to consensus.
+pub struct ExecutionSyncTracker {
+    last_executed: AtomicU64,
+    consecutive_failures: AtomicU32,
+    config: ExecutionSyncConfig,
+}
+
+impl ExecutionSyncTracker {
+    pub fn new(config: ExecutionSyncConfig) -> Self {
+        Self {
+            last_executed: AtomicU64::new(0),
+            consecutive_failures: AtomicU32::new(0),
+            config,
+        }
+    }
+
+    pub fn on_success(&self, height: u64) {
+        self.last_executed.store(height, Ordering::SeqCst);
+        self.consecutive_failures.store(0, Ordering::SeqCst);
+    }
+
+    pub fn on_failure(&self, consensus_height: u64, error: &str) -> SyncAction {
+        let failures = self.consecutive_failures.fetch_add(1, Ordering::SeqCst) + 1;
+        let last_executed = self.last_executed.load(Ordering::SeqCst);
+        let divergence = consensus_height.saturating_sub(last_executed);
+
+        error!(
+            last_executed,
+            consensus_height,
+            divergence,
+            consecutive_failures = failures,
+            error,
+            "Execution failed"
+        );
+
+        if divergence > self.config.max_divergence {
+            return SyncAction::Halt {
+                reason: format!(
+                    "Divergence {} exceeds max {}. Last executed: {}, consensus: {}",
+                    divergence, self.config.max_divergence, last_executed, consensus_height
+                ),
+            };
+        }
+
+        if failures > self.config.max_consecutive_failures {
+            return SyncAction::Halt {
+                reason: format!(
+                    "Consecutive failures {} exceeds max {}",
+                    failures, self.config.max_consecutive_failures
+                ),
+            };
+        }
+
+        SyncAction::Continue
+    }
+
+    pub fn last_executed(&self) -> u64 {
+        self.last_executed.load(Ordering::SeqCst)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_success_resets_failures() {
+        let tracker = ExecutionSyncTracker::new(ExecutionSyncConfig::default());
+        tracker.on_failure(5, "test");
+        tracker.on_failure(6, "test");
+        tracker.on_success(7);
+        assert_eq!(tracker.consecutive_failures.load(Ordering::SeqCst), 0);
+        assert_eq!(tracker.last_executed(), 7);
+    }
+
+    #[test]
+    fn test_divergence_triggers_halt() {
+        let config = ExecutionSyncConfig {
+            max_divergence: 5,
+            max_consecutive_failures: 100,
+        };
+        let tracker = ExecutionSyncTracker::new(config);
+        tracker.on_success(10);
+        let action = tracker.on_failure(16, "test");
+        assert!(matches!(action, SyncAction::Halt { .. }));
+    }
+
+    #[test]
+    fn test_consecutive_failures_triggers_halt() {
+        let config = ExecutionSyncConfig {
+            max_divergence: 100,
+            max_consecutive_failures: 3,
+        };
+        let tracker = ExecutionSyncTracker::new(config);
+        tracker.on_success(10);
+        assert_eq!(tracker.on_failure(11, "e"), SyncAction::Continue);
+        assert_eq!(tracker.on_failure(11, "e"), SyncAction::Continue);
+        assert_eq!(tracker.on_failure(11, "e"), SyncAction::Continue);
+        assert!(matches!(
+            tracker.on_failure(11, "e"),
+            SyncAction::Halt { .. }
+        ));
+    }
+}

crates/node/src/lib.rs

@@ -7,13 +7,16 @@
 pub mod client_config;
 pub mod config;
 pub mod execution_bridge;
+pub mod execution_sync;
 pub mod genesis_bootstrap;
 pub mod key_cli;
+pub mod mempool_state;
 pub mod network;
 pub mod network_api;
 pub mod node;
 pub mod supervisor;
 pub mod util;
+pub mod worker_pool_adapter;
 
 pub use client_config::ClientConfig;
 pub use config::{
@@ -23,11 +26,14 @@ pub use config::{
     DEFAULT_RPC_HTTP_PORT, DEFAULT_RPC_WS_PORT,
 };
 pub use execution_bridge::ExecutionBridge;
+pub use execution_sync::{ExecutionSyncConfig, ExecutionSyncTracker, SyncAction};
 pub use genesis_bootstrap::{
     GeneratedValidator, GenesisGenerationResult, GenesisGenerator, GenesisGeneratorConfig,
     GenesisLoader, ValidatorKeyFile,
 };
 pub use key_cli::{execute_keys_command, KeysCommand};
+pub use mempool_state::ExecutionStateValidator;
 pub use network_api::{NodeNetworkApi, TcpNetworkApi};
 pub use node::Node;
 pub use supervisor::{NodeSupervisor, ShutdownError};
+pub use worker_pool_adapter::WorkerPoolAdapter;