feat: preexecute transactions on followers and send access list to leader

e2e/test/automine/e2e-json-rpc.test.ts

@@ -104,8 +104,10 @@ describe("JSON-RPC", () => {
                 return;
             }
 
-            await send("stratus_disableUnknownClients");
-            await send("stratus_enableUnknownClients");
+            // client identification is required to toggle unknown clients once they are disabled
+            const identifiedHeaders = { "x-app": "test-client" };
+            await send("stratus_disableUnknownClients", [], identifiedHeaders);
+            await send("stratus_enableUnknownClients", [], identifiedHeaders);
 
             // Request without client identification should now succeed
             const blockNumber = await send("eth_blockNumber");

src/eth/executor/evm/types/input/call_execution.rs

@@ -8,12 +8,10 @@ use crate::eth::executor::evm::types::GAS_MAX_LIMIT;
 use crate::eth::executor::evm::types::GeneralRevm;
 use crate::eth::storage::ExecutionKind;
 use crate::eth::types::Address;
-use crate::eth::types::BlockHeader;
+use crate::eth::types::BlockInfo;
 use crate::eth::types::BlockNumber;
 use crate::eth::types::Bytes;
 use crate::eth::types::CallInput;
-use crate::eth::types::PendingBlockHeader;
-use crate::eth::types::PointInTime;
 use crate::eth::types::UnixTime;
 use crate::eth::types::Wei;
 use crate::ext::OptionExt;
@@ -61,32 +59,14 @@ pub struct CallExecutionInput {
 }
 
 impl CallExecutionInput {
-    /// Creates from a call that was sent directly to Stratus with `eth_call` or `eth_estimateGas` for a pending block.
-    pub fn from_pending_block(input: CallInput, block: PendingBlockHeader) -> Self {
+    pub fn create(input: CallInput, block_info: BlockInfo, kind: ExecutionKind) -> Self {
         Self {
             from: input.from.unwrap_or(Address::ZERO),
             to: input.to.map_into(),
             value: input.value,
             data: input.data,
-            block_number: block.number,
-            block_timestamp: *block.timestamp,
-            kind: ExecutionKind::CallLatest(block.number.prev().unwrap_or_default()),
-        }
-    }
-
-    /// Creates from a call that was sent directly to Stratus with `eth_call` or `eth_estimateGas` for a mined block.
-    pub fn from_mined_block(input: CallInput, block: BlockHeader, point_in_time: PointInTime) -> Self {
-        let kind = match point_in_time {
-            PointInTime::Latest | PointInTime::Pending => ExecutionKind::CallLatest(block.number),
-            PointInTime::Past(number) => ExecutionKind::CallPast(number),
-        };
-        Self {
-            from: input.from.unwrap_or(Address::ZERO),
-            to: input.to.map_into(),
-            value: input.value,
-            data: input.data,
-            block_number: block.number,
-            block_timestamp: block.timestamp,
+            block_number: block_info.number,
+            block_timestamp: *block_info.timestamp,
             kind,
         }
     }

src/eth/executor/evm/types/input/transaction_execution.rs

@@ -8,12 +8,12 @@ use crate::eth::executor::evm::types::GAS_MAX_LIMIT;
 use crate::eth::executor::evm::types::GeneralRevm;
 use crate::eth::storage::ExecutionKind;
 use crate::eth::types::Address;
+use crate::eth::types::BlockInfo;
 use crate::eth::types::BlockNumber;
 use crate::eth::types::Bytes;
 use crate::eth::types::ChainId;
 use crate::eth::types::Gas;
 use crate::eth::types::Nonce;
-use crate::eth::types::PendingBlockHeader;
 use crate::eth::types::TransactionInput;
 use crate::eth::types::UnixTime;
 use crate::eth::types::Wei;
@@ -81,7 +81,7 @@ pub struct TransactionExecutionInput {
 
 impl TransactionExecutionInput {
     /// Creates from a transaction that was sent to Stratus with `eth_sendRawTransaction` or during Importing.
-    pub fn from_eth_transaction(input: &TransactionInput, block_number: BlockNumber, block_timestamp: UnixTime) -> Self {
+    pub fn create(input: &TransactionInput, block_info: BlockInfo) -> Self {
         Self {
             from: input.signer(),
             to: input.execution_info.to,
@@ -90,16 +90,16 @@ impl TransactionExecutionInput {
             gas_limit: input.execution_info.gas_limit,
             gas_price: input.execution_info.gas_price,
             nonce: input.execution_info.nonce,
-            block_number,
-            block_timestamp,
+            block_number: block_info.number,
+            block_timestamp: *block_info.timestamp,
             chain_id: input.execution_info.chain_id,
             kind: ExecutionKind::Transaction,
         }
     }
 }
 
-impl PartialEq<&PendingBlockHeader> for TransactionExecutionInput {
-    fn eq(&self, other: &&PendingBlockHeader) -> bool {
+impl PartialEq<BlockInfo> for TransactionExecutionInput {
+    fn eq(&self, other: &BlockInfo) -> bool {
         self.block_number == other.number && self.block_timestamp == *other.timestamp
     }
 }

src/eth/executor/evm/types/output/access_list.rs

@@ -1,3 +1,4 @@
+use derive_more::IntoIterator;
 use display_json::DebugAsJson;
 use revm_state::EvmState;
 
@@ -6,8 +7,9 @@ use crate::eth::types::Address;
 use crate::eth::types::SlotIndex;
 use crate::eth::types::StratusError;
 
-#[derive(serde::Serialize, DebugAsJson)]
+#[derive(serde::Serialize, serde::Deserialize, DebugAsJson, Clone, IntoIterator)]
 pub struct AccessListOutput {
+    #[into_iterator(owned, ref, ref_mut)]
     access_list: Vec<(Address, Vec<SlotIndex>)>,
 }
 

src/eth/executor/mod.rs

@@ -40,7 +40,6 @@ use crate::eth::executor::evm::types::SlotAccessMetrics;
 use crate::eth::executor::evm_worker_pool::EvmWorkerPool;
 use crate::eth::executor::types::EvmRoute;
 use crate::eth::miner::Miner;
-use crate::eth::rpc::RpcError;
 use crate::eth::storage::ExecutionKind;
 use crate::eth::storage::StorageError;
 use crate::eth::storage::StratusStorage;
@@ -61,6 +60,7 @@ use crate::ext::to_json_string;
 use crate::infra::metrics;
 use crate::infra::metrics::timed;
 use crate::infra::tracing::SpanExt;
+use crate::utils::Semaphore;
 
 // -----------------------------------------------------------------------------
 // Executor
@@ -70,6 +70,7 @@ use crate::infra::tracing::SpanExt;
 #[derive(Default)]
 pub struct ExecutorLocks {
     transaction: Mutex<()>,
+    transaction_warmup: Semaphore,
 }
 
 pub struct Executor {
@@ -95,7 +96,10 @@ impl Executor {
         let reject_not_contract = config.executor_reject_not_contract;
         let evms = EvmWorkerPool::spawn(Arc::clone(&storage), &config);
         Self {
-            locks: ExecutorLocks::default(),
+            locks: ExecutorLocks {
+                transaction_warmup: Semaphore::new(100),
+                ..Default::default()
+            },
             evms,
             miner,
             storage,
@@ -172,8 +176,8 @@ impl Executor {
         tracing::info!(%block_number, tx_hash = %tx.hash(), "reexecuting external transaction");
 
         let tx_input: TransactionInput = tx.try_into()?;
-        let pending_block = self.storage.read_pending_block_header();
-        let mut evm_input = TransactionExecutionInput::from_eth_transaction(&tx_input, pending_block.number, *pending_block.timestamp);
+        let pending_header = self.storage.read_pending_block_header();
+        let mut evm_input = TransactionExecutionInput::create(&tx_input, pending_header);
 
         // when transaction externally failed, create fake transaction instead of reexecuting
         let (tx_execution, state) = match receipt.is_success() {
@@ -273,7 +277,7 @@ impl Executor {
 
     /// Executes a transaction persisting state changes.
     #[tracing::instrument(name = "executor::local_transaction", skip_all, fields(tx_hash, tx_from, tx_to, tx_nonce))]
-    pub fn execute_local_transaction(&self, tx: TransactionInput) -> Result<(), StratusError> {
+    pub fn execute_local_transaction(&self, tx: TransactionInput, access_list: Option<AccessListOutput>) -> Result<(), StratusError> {
         #[cfg(feature = "metrics")]
         let function = codegen::function_sig(&tx.execution_info.input);
         #[cfg(feature = "metrics")]
@@ -289,20 +293,25 @@ impl Executor {
             s.rec_str("tx_nonce", &tx.execution_info.nonce);
         });
 
+        let permit = self.locks.transaction_warmup.acquire();
+
+        if let Some(access_list) = access_list {
+            self.storage.load_access_list(access_list);
+        }
+
         // execute according to the strategy
         const INFINITE_ATTEMPTS: usize = usize::MAX;
 
         // Executes transactions serially:
         // * Uses a Mutex, so a new transactions starts executing only after the previous one is executed and persisted.
         // * Without a Mutex, conflict can happen because the next transactions starts executing before the previous one is saved.
         #[cfg(feature = "metrics")]
-        let lock_wait_start = metrics::now();
+        metrics::inc_executor_local_transaction_lock_waiting(1);
         let transaction_lock = self.locks.transaction.lock();
+        let start = metrics::now();
 
         #[cfg(feature = "metrics")]
-        let lock_wait = lock_wait_start.elapsed();
-        #[cfg(feature = "metrics")]
-        let start = metrics::now();
+        metrics::dec_executor_local_transaction_lock_waiting(1);
 
         // execute transaction
         let tx_execution = self.execute_local_transaction_attempts(tx, INFINITE_ATTEMPTS);
@@ -311,9 +320,8 @@ impl Executor {
         let execution_elapsed = start.elapsed();
 
         drop(transaction_lock);
+        drop(permit);
 
-        #[cfg(feature = "metrics")]
-        metrics::inc_executor_local_transaction_lock_wait(lock_wait);
         #[cfg(feature = "metrics")]
         metrics::inc_executor_local_transaction(execution_elapsed, tx_execution.is_ok(), contract, function);
 
@@ -348,7 +356,7 @@ impl Executor {
 
             // prepare evm input
             let pending_header = self.storage.read_pending_block_header();
-            let evm_input = TransactionExecutionInput::from_eth_transaction(&tx_input, pending_header.number, *pending_header.timestamp);
+            let evm_input = TransactionExecutionInput::create(&tx_input, pending_header);
 
             // execute transaction in evm (retry only in case of conflict, but do not retry on other failures)
             tracing::debug!(
@@ -407,15 +415,14 @@ impl Executor {
         }
     }
 
-    /// Executes a transaction without persisting state changes.
+    /// Executes a read-only call in the local EVM, without persisting state changes.
     #[tracing::instrument(name = "executor::local_call", skip_all, fields(from, to))]
-    pub fn execute_local_call<Output>(&self, call_input: CallInput, point_in_time: PointInTime) -> Result<Output, StratusError>
+    pub fn execute_local_call<Output>(&self, call_input: CallInput, kind: ExecutionKind) -> Result<Output, StratusError>
     where
         Output: TryFrom<RevmResultAndState, Error = StratusError>,
     {
         #[cfg(feature = "metrics")]
         let start = metrics::now();
-
         Span::with(|s| {
             s.rec_opt("from", &call_input.from);
             s.rec_opt("to", &call_input.to);
@@ -425,31 +432,25 @@ impl Executor {
             to = ?call_input.to,
             data_len = call_input.data.len(),
             data = %call_input.data,
-            %point_in_time,
+            ?kind,
             "executing read-only local transaction"
         );
 
         #[cfg(feature = "metrics")]
         let (function, contract) = { (codegen::function_sig(&call_input.data), codegen::contract_name(&call_input.to)) };
 
-        // execute
-        let evm_input = match point_in_time {
-            PointInTime::Pending => {
-                let pending_header = self.storage.read_pending_block_header();
-                CallExecutionInput::from_pending_block(call_input, pending_header)
-            }
-            _ => {
-                let Some(block) = self.storage.read_block(point_in_time.into())? else {
-                    return Err(RpcError::BlockFilterInvalid { filter: point_in_time.into() }.into());
-                };
-                CallExecutionInput::from_mined_block(call_input, block.header, point_in_time)
-            }
+        let filter = kind.into();
+        let Some(block_info) = self.storage.read_block_info(filter)? else {
+            return Err(StorageError::BlockNotFound { filter }.into());
         };
 
-        let evm_route = match point_in_time {
+        let evm_input = CallExecutionInput::create(call_input, block_info, kind);
+
+        let evm_route = match kind.point_in_time() {
             PointInTime::Pending | PointInTime::Latest => EvmRoute::CallPresent(evm_input),
             PointInTime::Past(_) => EvmRoute::CallPast(evm_input),
         };
+
         let evm_result = self.evms.execute::<Output>(evm_route);
 
         // track metrics

src/eth/follower/consensus.rs

@@ -2,10 +2,14 @@ use std::sync::Arc;
 
 use strum::AsRefStr;
 
+use crate::eth::executor::AccessListOutput;
+use crate::eth::executor::Executor;
 use crate::eth::follower::importer::BlockchainClient;
 use crate::eth::types::Bytes;
+use crate::eth::types::ExecutionKind;
 use crate::eth::types::Hash;
 use crate::eth::types::StratusError;
+use crate::eth::types::TransactionInput;
 #[cfg(feature = "metrics")]
 use crate::infra::metrics;
 
@@ -57,25 +61,39 @@ pub trait Consensus: Send + Sync {
         !(lag.is_far_behind() || lag.is_ahead())
     }
 
+    /// Whether transactions forwarded to the leader should carry a pre-computed access list.
+    fn forward_access_list(&self) -> bool;
+
     /// Forwards a transaction to leader.
     ///
     /// The current machine name is sent as the `x-client` header by `BlockchainClient`, so the leader
     /// attributes the transaction to this node automatically.
-    async fn forward_to_leader(&self, tx_hash: Hash, tx_data: Bytes) -> Result<Hash, StratusError> {
+    async fn forward_to_leader(&self, tx: TransactionInput, tx_hash: Hash, tx_data: Bytes) -> Result<Hash, StratusError> {
         #[cfg(feature = "metrics")]
         let start = metrics::now();
 
         tracing::info!(%tx_hash, "forwarding transaction to leader");
 
-        let hash = self.get_chain()?.send_raw_transaction_to_leader(tx_data.into()).await?;
+        let access_list = if self.forward_access_list() {
+            Some(
+                self.get_executor()
+                    .execute_local_call::<AccessListOutput>(tx.into(), ExecutionKind::AccessList)?,
+            )
+        } else {
+            None
+        };
+
+        let hash = self.get_client().send_raw_transaction_to_leader(tx_data.into(), access_list).await?;
 
         #[cfg(feature = "metrics")]
         metrics::inc_consensus_forward(start.elapsed());
 
         Ok(hash)
     }
 
-    fn get_chain(&self) -> anyhow::Result<&Arc<BlockchainClient>>;
+    fn get_client(&self) -> &Arc<BlockchainClient>;
+
+    fn get_executor(&self) -> &Arc<Executor>;
 
     /// Get the lag status between this node and the leader.
     async fn lag(&self) -> anyhow::Result<LagStatus>;

src/eth/follower/importer/importer_config.rs → src/eth/follower/importer/config.rs

@@ -12,8 +12,8 @@ use crate::eth::follower::ConsensusError;
 use crate::eth::follower::ImporterError;
 use crate::eth::follower::importer::BlockchainClient;
 use crate::eth::follower::importer::ImporterMode;
-use crate::eth::follower::importer::importer_supervisor::ImporterConsensus;
-use crate::eth::follower::importer::importer_supervisor::start_importer;
+use crate::eth::follower::importer::supervisor::ImporterConsensus;
+use crate::eth::follower::importer::supervisor::start_importer;
 use crate::eth::miner::Miner;
 use crate::eth::rpc::RpcContext;
 use crate::eth::storage::StratusStorage;
@@ -47,6 +47,10 @@ pub struct ImporterConfig {
     #[arg(long = "enable-block-changes-replication", env = "ENABLE_BLOCK_CHANGES_REPLICATION", default_value = "false")]
     pub enable_block_changes_replication: bool,
 
+    /// Compute an access list for transactions before forwarding them to the leader.
+    #[arg(long = "forward-access-list", env = "FORWARD_ACCESS_LIST", default_value = "true", required = false)]
+    pub forward_access_list: bool,
+
     /// Specify the block to stop importing. (useful for validating a follower db against a fake leader)
     #[arg(long = "stop-at-block", env = "STOP_AT_BLOCK")]
     pub stop_at_block: Option<BlockNumber>,
@@ -94,6 +98,8 @@ impl ImporterConfig {
         let consensus = Arc::new(ImporterConsensus {
             storage: Arc::clone(&storage),
             chain: Arc::clone(&chain),
+            executor: Arc::clone(&executor),
+            forward_access_list: self.forward_access_list,
         });
 
         spawn(

src/eth/follower/importer/importers/blockchain_client.rs

@@ -20,6 +20,7 @@ use crate::GlobalState;
 use crate::alias::AlloyBytes;
 use crate::alias::AlloyTransaction;
 use crate::alias::JsonValue;
+use crate::eth::executor::AccessListOutput;
 use crate::eth::executor::ExecutorError;
 use crate::eth::rpc::pagination;
 use crate::eth::storage::permanent::rocks::types::BlockChangesRocksdb;
@@ -343,11 +344,12 @@ impl BlockchainClient {
     ///
     /// The current machine name is sent as the `x-client` header on every request (see `client_headers`),
     /// so the leader attributes the transaction to this node automatically.
-    pub async fn send_raw_transaction_to_leader(&self, tx: AlloyBytes) -> Result<Hash, StratusError> {
+    pub async fn send_raw_transaction_to_leader(&self, tx: AlloyBytes, access_list: Option<AccessListOutput>) -> Result<Hash, StratusError> {
         tracing::debug!("sending raw transaction to leader");
 
         let tx = to_json_value(tx);
-        let result = self.http.request::<Hash, _>("eth_sendRawTransaction", [tx]).await;
+        let access_list = to_json_value(access_list);
+        let result = self.http.request::<Hash, _>("eth_sendRawTransaction", [tx, access_list]).await;
 
         match result {
             Ok(hash) => Ok(hash),