feat: support dev_pack_tx_immediately in espace

Author: darwintree

Cargo.lock

@@ -725,6 +725,7 @@ pub fn initialize_not_light_node_modules(
             consensus.clone(),
             sync.clone(),
             txpool.clone(),
+            blockgen.clone(),
             notifications.clone(),
             task_executor.clone(),
             conf,

crates/client/src/common/mod.rs

@@ -2,6 +2,7 @@
 // Conflux is free software and distributed under GNU General Public License.
 // See http://www.gnu.org/licenses/
 
+use blockgen::BlockGenerator;
 use cfx_rpc_builder::{
     RpcModuleBuilder, RpcServerConfig, RpcServerHandle,
     TransportRpcModuleConfig,
@@ -406,8 +407,9 @@ where
 // start espace rpc server v2(async)
 pub async fn launch_async_rpc_servers(
     consensus: SharedConsensusGraph, sync: SharedSynchronizationService,
-    tx_pool: SharedTransactionPool, notifications: Arc<Notifications>,
-    executor: TaskExecutor, conf: &Configuration,
+    tx_pool: SharedTransactionPool, block_gen: Arc<BlockGenerator>,
+    notifications: Arc<Notifications>, executor: TaskExecutor,
+    conf: &Configuration,
 ) -> Result<Option<RpcServerHandle>, String> {
     let http_config = conf.eth_http_config();
     let ws_config = conf.eth_ws_config();
@@ -455,6 +457,7 @@ pub async fn launch_async_rpc_servers(
         consensus,
         sync,
         tx_pool,
+        block_gen,
         executor,
         notifications,
     );

crates/client/src/rpc/mod.rs

@@ -31,5 +31,6 @@ log = { workspace = true }
 cfx-rpc-middlewares = { workspace = true }
 cfx-tasks = { workspace = true }
 tokio = { workspace = true }
+blockgen = { workspace = true }
 
 [dev-dependencies]

crates/rpc/rpc-builder/Cargo.toml

@@ -37,6 +37,7 @@ pub use id_provider::EthSubscriptionIdProvider;
 use log::debug;
 pub use module::{EthRpcModule, RpcModuleSelection};
 
+use blockgen::BlockGenerator;
 use cfx_rpc::{helpers::ChainInfo, *};
 use cfx_rpc_cfx_types::RpcImplConfiguration;
 use cfx_rpc_eth_api::*;
@@ -75,6 +76,7 @@ pub struct RpcModuleBuilder {
     consensus: SharedConsensusGraph,
     sync: SharedSynchronizationService,
     tx_pool: SharedTransactionPool,
+    block_gen: Arc<BlockGenerator>,
     executor: TaskExecutor,
     notifications: Arc<Notifications>,
 }
@@ -83,13 +85,15 @@ impl RpcModuleBuilder {
     pub fn new(
         config: RpcImplConfiguration, consensus: SharedConsensusGraph,
         sync: SharedSynchronizationService, tx_pool: SharedTransactionPool,
-        executor: TaskExecutor, notifications: Arc<Notifications>,
+        block_gen: Arc<BlockGenerator>, executor: TaskExecutor,
+        notifications: Arc<Notifications>,
     ) -> Self {
         Self {
             config,
             consensus,
             sync,
             tx_pool,
+            block_gen,
             executor,
             notifications,
         }
@@ -111,6 +115,7 @@ impl RpcModuleBuilder {
                 consensus,
                 sync,
                 tx_pool,
+                block_gen,
                 executor,
                 notifications,
             } = self;
@@ -120,6 +125,7 @@ impl RpcModuleBuilder {
                 consensus,
                 sync,
                 tx_pool,
+                block_gen,
                 executor,
                 notifications,
             );
@@ -140,6 +146,7 @@ pub struct RpcRegistryInner {
     config: RpcImplConfiguration,
     sync: SharedSynchronizationService,
     tx_pool: SharedTransactionPool,
+    block_gen: Arc<BlockGenerator>,
     modules: HashMap<EthRpcModule, Methods>,
     executor: TaskExecutor,
     notifications: Arc<Notifications>,
@@ -149,13 +156,15 @@ impl RpcRegistryInner {
     pub fn new(
         config: RpcImplConfiguration, consensus: SharedConsensusGraph,
         sync: SharedSynchronizationService, tx_pool: SharedTransactionPool,
-        executor: TaskExecutor, notifications: Arc<Notifications>,
+        block_gen: Arc<BlockGenerator>, executor: TaskExecutor,
+        notifications: Arc<Notifications>,
     ) -> Self {
         Self {
             consensus,
             config,
             sync,
             tx_pool,
+            block_gen,
             modules: Default::default(),
             executor,
             notifications,
@@ -248,6 +257,7 @@ impl RpcRegistryInner {
                             self.consensus.clone(),
                             self.sync.clone(),
                             self.tx_pool.clone(),
+                            self.block_gen.clone(),
                             self.executor.clone(),
                         )
                         .into_rpc();
@@ -286,6 +296,7 @@ impl RpcRegistryInner {
                             self.consensus.clone(),
                             self.sync.clone(),
                             self.tx_pool.clone(),
+                            self.block_gen.clone(),
                             self.executor.clone(),
                         );
                         ParityApi::new(eth_api).into_rpc().into()

crates/rpc/rpc-builder/src/lib.rs

@@ -19,6 +19,7 @@ cfx-types = { workspace = true }
 cfx-rpc-eth-types = { workspace = true }
 cfx-rpc-primitives = { workspace = true }
 cfx-rpc-cfx-impl = { workspace = true }
+blockgen = { workspace = true }
 async-trait = { workspace = true }
 keccak-hash = { workspace = true }
 alloy-rpc-types-trace = { workspace = true }
@@ -49,4 +50,4 @@ solidity-abi = { workspace = true }
 cfx-rpc-common-impl = { workspace = true }
 cfx-tasks = { workspace = true }
 cfx-parity-trace-types = { workspace = true }
-cfxcore-errors = { workspace = true }
+cfxcore-errors = { workspace = true }

crates/rpc/rpc-eth-impl/Cargo.toml

@@ -1,5 +1,6 @@
 use crate::helpers::{FeeHistoryCache, MAX_FEE_HISTORY_CACHE_BLOCK_COUNT};
 use async_trait::async_trait;
+use blockgen::BlockGenerator;
 use cfx_execute_helper::estimation::EstimateRequest;
 use cfx_executor::executive::{
     Executed, ExecutionError, ExecutionOutcome, TxDropError,
@@ -32,6 +33,7 @@ use cfx_types::{
 use cfx_util_macros::bail;
 use cfx_vm_types::Error as VmError;
 use cfxcore::{
+    consensus_parameters::DEFERRED_STATE_EPOCH_COUNT,
     errors::{Error as CoreError, Result as CoreResult},
     ConsensusGraph, SharedConsensusGraph, SharedSynchronizationService,
     SharedTransactionPool,
@@ -45,7 +47,9 @@ use primitives::{
 };
 use rustc_hex::ToHex;
 use solidity_abi::string_revert_reason_decode;
-use std::{collections::HashMap, future::Future};
+use std::{
+    collections::HashMap, future::Future, sync::Arc, thread, time::Duration,
+};
 
 type BlockNumber = BlockId;
 type BlockNumberOrTag = BlockId;
@@ -59,6 +63,7 @@ pub struct EthApi {
     consensus: SharedConsensusGraph,
     sync: SharedSynchronizationService,
     tx_pool: SharedTransactionPool,
+    block_gen: Arc<BlockGenerator>,
     fee_history_cache: FeeHistoryCache,
     executor: TaskExecutor,
 }
@@ -67,13 +72,14 @@ impl EthApi {
     pub fn new(
         config: RpcImplConfiguration, consensus: SharedConsensusGraph,
         sync: SharedSynchronizationService, tx_pool: SharedTransactionPool,
-        executor: TaskExecutor,
+        block_gen: Arc<BlockGenerator>, executor: TaskExecutor,
     ) -> Self {
         EthApi {
             config,
             consensus,
             sync,
             tx_pool,
+            block_gen,
             fee_history_cache: FeeHistoryCache::new(),
             executor,
         }
@@ -83,6 +89,17 @@ impl EthApi {
 
     pub fn tx_pool(&self) -> &SharedTransactionPool { &self.tx_pool }
 
+    fn generate_one_block(
+        &self, num_txs: usize, block_size_limit: usize,
+    ) -> CoreResult<H256> {
+        let hash = self.block_gen.test_api().generate_block(
+            num_txs,
+            block_size_limit,
+            vec![],
+        );
+        Ok(hash)
+    }
+
     pub fn fetch_block_by_height(
         &self, height: u64,
     ) -> Result<PhantomBlock, ProviderBlockError> {
@@ -1491,6 +1508,28 @@ impl EthApiServer for EthApi {
         }
 
         let r = self.send_transaction_with_signature(tx)?;
+        if self.config.dev_pack_tx_immediately {
+            // Try to pack and execute this new tx.
+            for _ in 0..DEFERRED_STATE_EPOCH_COUNT {
+                let generated = self.generate_one_block(
+                    1, /* num_txs */
+                    self.sync
+                        .get_synchronization_graph()
+                        .verification_config
+                        .max_block_size_in_bytes,
+                )?;
+                loop {
+                    // Wait for the new block to be fully processed, so all
+                    // generated blocks form a chain for
+                    // `tx` to be executed.
+                    if self.consensus.best_block_hash() == generated {
+                        break;
+                    } else {
+                        thread::sleep(Duration::from_millis(10));
+                    }
+                }
+            }
+        }
         Ok(r)
     }