Merge branch 'main' into importer-component

.clippy.toml

@@ -1 +1,2 @@
+allow-unwrap-in-tests = true
 disallowed-names = ["lock"]

Cargo.toml

@@ -180,10 +180,12 @@ raft = []
 [lints.clippy]
 clone_on_ref_ptr = "warn"
 cloned_instead_of_copied = "warn"
+default_trait_access = "warn"
 disallowed_names = "warn"
 manual_let_else = "warn"
 semicolon_if_nothing_returned = "warn"
 unused_async = "warn"
 unused_self = "warn"
 used_underscore_binding = "warn"
-wildcard_imports = "warn"
+wildcard_imports = "warn"
+unwrap_used = "allow"

build.rs

@@ -130,8 +130,7 @@ fn generate_signature_module_content() -> String {
     let mut signatures_4_bytes = phf_codegen::Map::<[u8; 4]>::new();
     let mut signatures_32_bytes = phf_codegen::Map::<[u8; 32]>::new();
     for signature_file in signature_files {
-        let prefix = signature_file.filename.file_name().unwrap().to_str().unwrap().split('.').next().unwrap();
-        populate_signature_maps(&signature_file.content, &mut seen, &mut signatures_4_bytes, &mut signatures_32_bytes, prefix);
+        populate_signature_maps(&signature_file.content, &mut seen, &mut signatures_4_bytes, &mut signatures_32_bytes);
     }
 
     format!(
@@ -152,7 +151,6 @@ fn populate_signature_maps(
     seen: &mut HashSet<Vec<u8>>,
     signatures_4_bytes: &mut phf_codegen::Map<[u8; 4]>,
     signatures_32_bytes: &mut phf_codegen::Map<[u8; 32]>,
-    prefix: &str,
 ) {
     for line in file_content.lines() {
         if let Ok((_, (id, signature))) = parse_signature(line) {
@@ -161,7 +159,7 @@ fn populate_signature_maps(
             }
             seen.insert(id.clone());
 
-            let signature = format!("\"{}::{}\"", prefix, signature);
+            let signature = format!("\"{}\"", signature);
             match id.len() {
                 4 => {
                     signatures_4_bytes.entry(id.try_into().unwrap(), &signature);
@@ -188,7 +186,7 @@ fn parse_signature(input: &str) -> IResult<&str, (SolidityId, &SoliditySignature
 // -----------------------------------------------------------------------------
 
 struct InputFile {
-    filename: PathBuf,
+    _filename: PathBuf,
     content: String,
 }
 
@@ -202,15 +200,15 @@ fn list_files(pattern: &'static str) -> Vec<InputFile> {
 
     // ensure at least one exists
     if filenames.is_empty() {
-        panic!("No signature files found in \"{}\"", pattern);
+        panic!("No files found in \"{}\"", pattern);
     }
 
     // read file contents
     let mut files = Vec::with_capacity(filenames.len());
     for filename in filenames {
         files.push(InputFile {
             content: fs::read_to_string(&filename).expect("Reading file should not fail"),
-            filename,
+            _filename: filename,
         });
     }
 

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

@@ -21,6 +21,7 @@ import {
     send,
     sendEvmMine,
     sendExpect,
+    sendGetBlockNumber,
     sendGetNonce,
     sendRawTransaction,
     sendReset,
@@ -136,14 +137,87 @@ describe("JSON-RPC", () => {
     });
 
     describe("Logs", () => {
+        describe("eth_getLogs", () => {
+            it("returns the expected amount of logs for different block ranges", async () => {
+                await sendReset();
+
+                const contract = await deployTestContractBalances();
+                const filter = { address: contract.target };
+                await sendEvmMine();
+                await contract.waitForDeployment();
+
+                const contractOps = contract.connect(ALICE.signer());
+                await sendEvmMine();
+                const block = await sendGetBlockNumber();
+
+                async function sendNTransactions(n: number) {
+                    for (let i = 0; i < n; i++) {
+                        await contractOps.add(ALICE.address, 10);
+                    }
+                    await sendEvmMine();
+                }
+
+                await sendNTransactions(8);
+                await sendNTransactions(4);
+                await sendNTransactions(2);
+                await sendNTransactions(1);
+
+                // single block
+                expect(
+                    await send("eth_getLogs", [{ ...filter, fromBlock: toHex(block + 1), toBlock: toHex(block + 1) }]),
+                ).length(8);
+                expect(
+                    await send("eth_getLogs", [{ ...filter, fromBlock: toHex(block + 2), toBlock: toHex(block + 2) }]),
+                ).length(4);
+                expect(
+                    await send("eth_getLogs", [{ ...filter, fromBlock: toHex(block + 3), toBlock: toHex(block + 3) }]),
+                ).length(2);
+                expect(
+                    await send("eth_getLogs", [{ ...filter, fromBlock: toHex(block + 4), toBlock: toHex(block + 4) }]),
+                ).length(1);
+                expect(
+                    await send("eth_getLogs", [{ ...filter, fromBlock: toHex(block + 5), toBlock: toHex(block + 5) }]),
+                ).length(0);
+
+                // block range (offset fromBlock)
+                expect(await send("eth_getLogs", [{ ...filter, fromBlock: toHex(block + 1) }])).length(15);
+                expect(await send("eth_getLogs", [{ ...filter, fromBlock: toHex(block + 2) }])).length(7);
+                expect(await send("eth_getLogs", [{ ...filter, fromBlock: toHex(block + 3) }])).length(3);
+                expect(await send("eth_getLogs", [{ ...filter, fromBlock: toHex(block + 4) }])).length(1);
+                expect(await send("eth_getLogs", [{ ...filter, fromBlock: toHex(block + 5) }])).length(0);
+
+                // block range (offset toBlock)
+                expect(
+                    await send("eth_getLogs", [{ ...filter, fromBlock: toHex(block), toBlock: toHex(block + 1) }]),
+                ).length(8);
+                expect(
+                    await send("eth_getLogs", [{ ...filter, fromBlock: toHex(block), toBlock: toHex(block + 2) }]),
+                ).length(12);
+                expect(
+                    await send("eth_getLogs", [{ ...filter, fromBlock: toHex(block), toBlock: toHex(block + 3) }]),
+                ).length(14);
+                expect(
+                    await send("eth_getLogs", [{ ...filter, fromBlock: toHex(block), toBlock: toHex(block + 4) }]),
+                ).length(15);
+                expect(
+                    await send("eth_getLogs", [{ ...filter, fromBlock: toHex(block), toBlock: toHex(block + 4) }]),
+                ).length(15);
+
+                // block range (middle blocks)
+                expect(
+                    await send("eth_getLogs", [{ ...filter, fromBlock: toHex(block + 2), toBlock: toHex(block + 3) }]),
+                ).length(6);
+            });
+        });
+
         describe("eth_getLogs", () => {
             it("returns no logs for queries after last mined block", async () => {
                 const contract = await deployTestContractBalances();
-                sendEvmMine();
+                await sendEvmMine();
                 await contract.waitForDeployment();
 
                 const txResponse = await contract.connect(ALICE.signer()).add(ALICE.address, 10);
-                sendEvmMine();
+                await sendEvmMine();
                 const txReceipt = await ETHERJS.getTransactionReceipt(txResponse.hash);
                 expect(txReceipt).to.not.be.null;
 
@@ -155,10 +229,10 @@ describe("JSON-RPC", () => {
                 const filter = { address: contract.target };
                 expect(await send("eth_getLogs", [{ ...filter, fromBlock: toHex(txBlockNumber) }])).length(1); // last mined block
 
-                sendEvmMine();
+                await sendEvmMine();
                 expect(await send("eth_getLogs", [{ ...filter, fromBlock: toHex(txBlockNumber + 1) }])).length(0); // 1 after mined block
 
-                sendEvmMine();
+                await sendEvmMine();
                 expect(await send("eth_getLogs", [{ ...filter, fromBlock: toHex(txBlockNumber + 2) }])).length(0); // 2 after mined block
             });
         });

src/bin/importer_offline.rs

@@ -119,19 +119,32 @@ fn execute_block_importer(
             return Ok(());
         };
 
-        // receive new tasks to execute, or exit
+        // receive blocks to execute
         let Some((blocks, receipts)) = backlog_rx.blocking_recv() else {
-            tracing::info!("{} has no more blocks to process", TASK_NAME);
+            tracing::info!("{} has no more blocks to reexecute", TASK_NAME);
             return Ok(());
         };
 
-        // imports block transactions
-        let block_start = blocks.first().unwrap().number();
-        let block_end = blocks.last().unwrap().number();
+        // ensure range is not empty
+        let (Some(block_start), Some(block_end)) = (blocks.first(), blocks.last()) else {
+            let message = GlobalState::shutdown_from(TASK_NAME, "received empty block range to reexecute");
+            return log_and_err!(message);
+        };
+
+        // track operation
+        let block_start = block_start.number();
+        let block_end = block_end.number();
         let blocks_len = blocks.len();
-        let receipts = ExternalReceipts::from(receipts);
+        tracing::info!(%block_start, %block_end, receipts = %receipts.len(), "reexecuting blocks");
 
-        tracing::info!(%block_start, %block_end, receipts = %receipts.len(), "reexecuting (and importing) blocks");
+        // ensure block range have no gaps
+        if block_start.count_to(&block_end) != blocks_len as u64 {
+            let message = GlobalState::shutdown_from(TASK_NAME, "received block range with gaps to reexecute");
+            return log_and_err!(message);
+        }
+
+        // imports block transactions
+        let receipts = ExternalReceipts::from(receipts);
         let mut transaction_count = 0;
         let instant_before_execution = Instant::now();
 
@@ -230,10 +243,10 @@ async fn execute_external_rpc_storage_loader(
     }
 }
 
-async fn load_blocks_and_receipts(rpc_storage: Arc<dyn ExternalRpcStorage>, start: BlockNumber, end: BlockNumber) -> anyhow::Result<BacklogTask> {
-    tracing::info!(%start, %end, "loading blocks and receipts");
-    let blocks_task = rpc_storage.read_blocks_in_range(start, end);
-    let receipts_task = rpc_storage.read_receipts_in_range(start, end);
+async fn load_blocks_and_receipts(rpc_storage: Arc<dyn ExternalRpcStorage>, block_start: BlockNumber, block_end: BlockNumber) -> anyhow::Result<BacklogTask> {
+    tracing::info!(%block_start, %block_end, "loading blocks and receipts");
+    let blocks_task = rpc_storage.read_blocks_in_range(block_start, block_end);
+    let receipts_task = rpc_storage.read_receipts_in_range(block_start, block_end);
     try_join!(blocks_task, receipts_task)
 }
 

src/eth/codegen.rs

@@ -1,16 +1,29 @@
 //! Auto-generated code.
 
 use crate::eth::primitives::Address;
+use crate::infra::metrics;
 
 include!(concat!(env!("OUT_DIR"), "/contracts.rs"));
 include!(concat!(env!("OUT_DIR"), "/signatures.rs"));
 
+pub type SoliditySignature = &'static str;
+
 pub type ContractName = &'static str;
 
-/// Returns the contract name for a given address.
-pub fn get_contract_name(address: &Address) -> ContractName {
+/// Returns the contract name to be used in observability tasks.
+pub fn contract_name_for_o11y(address: &Option<Address>) -> ContractName {
+    let Some(address) = address else { return metrics::LABEL_MISSING };
     match CONTRACTS.get(address.as_bytes()) {
         Some(contract_name) => contract_name,
-        None => "unknown",
+        None => metrics::LABEL_UNKNOWN,
+    }
+}
+
+/// Returns the function name to be used in observability tasks.
+pub fn function_sig_for_o11y(bytes: impl AsRef<[u8]>) -> SoliditySignature {
+    let Some(id) = bytes.as_ref().get(..4) else { return metrics::LABEL_MISSING };
+    match SIGNATURES_4_BYTES.get(id) {
+        Some(signature) => signature,
+        None => metrics::LABEL_UNKNOWN,
     }
 }

src/eth/consensus/mod.rs

@@ -36,7 +36,7 @@ use crate::infra::BlockchainClient;
 use crate::GlobalState;
 
 pub mod append_entry {
-    #![allow(clippy::wildcard_imports)]
+    #![allow(clippy::default_trait_access, clippy::wildcard_imports)]
     tonic::include_proto!("append_entry");
 }
 #[allow(unused_imports)]

src/eth/executor/evm.rs

@@ -1,4 +1,5 @@
 use std::cmp::min;
+use std::collections::HashMap;
 use std::sync::Arc;
 
 use anyhow::anyhow;
@@ -199,17 +200,17 @@ impl RevmSession {
     pub fn new(storage: Arc<StratusStorage>) -> Self {
         Self {
             storage,
-            input: Default::default(),
-            storage_changes: Default::default(),
-            metrics: Default::default(),
+            input: EvmInput::default(),
+            storage_changes: HashMap::default(),
+            metrics: EvmExecutionMetrics::default(),
         }
     }
 
     /// Resets the session to be used with a new transaction.
     pub fn reset(&mut self, input: EvmInput) {
         self.input = input;
-        self.storage_changes = Default::default();
-        self.metrics = Default::default();
+        self.storage_changes = HashMap::default();
+        self.metrics = EvmExecutionMetrics::default();
     }
 }
 

src/eth/executor/executor.rs

@@ -7,6 +7,7 @@ use anyhow::anyhow;
 use tracing::info_span;
 use tracing::Span;
 
+use crate::eth::codegen;
 use crate::eth::executor::Evm;
 use crate::eth::executor::EvmExecutionResult;
 use crate::eth::executor::EvmInput;
@@ -333,16 +334,17 @@ impl Executor {
         self.miner.save_execution(tx_execution.clone())?;
 
         // track metrics
+
         #[cfg(feature = "metrics")]
         {
             let evm_execution = tx_execution.execution();
             let evm_metrics = tx_execution.metrics();
             *block_metrics += *evm_metrics;
 
-            let function = tx.solidity_signature();
-            metrics::inc_executor_external_transaction(start.elapsed(), function.clone());
-            metrics::inc_executor_external_transaction_account_reads(evm_metrics.account_reads, function.clone());
-            metrics::inc_executor_external_transaction_slot_reads(evm_metrics.slot_reads, function.clone());
+            let function = codegen::function_sig_for_o11y(&tx.0.input);
+            metrics::inc_executor_external_transaction(start.elapsed(), function);
+            metrics::inc_executor_external_transaction_account_reads(evm_metrics.account_reads, function);
+            metrics::inc_executor_external_transaction_slot_reads(evm_metrics.slot_reads, function);
             metrics::inc_executor_external_transaction_gas(evm_execution.gas.as_u64() as usize, function);
         }
 
@@ -422,12 +424,12 @@ impl Executor {
         // track metrics
         #[cfg(feature = "metrics")]
         {
-            let function = tx_input.solidity_signature();
+            let function = codegen::function_sig_for_o11y(&tx_input.input);
             match &tx_execution {
                 Ok(tx_execution) => {
-                    metrics::inc_executor_local_transaction(start.elapsed(), true, function.clone());
-                    metrics::inc_executor_local_transaction_account_reads(tx_execution.metrics().account_reads, function.clone());
-                    metrics::inc_executor_local_transaction_slot_reads(tx_execution.metrics().slot_reads, function.clone());
+                    metrics::inc_executor_local_transaction(start.elapsed(), true, function);
+                    metrics::inc_executor_local_transaction_account_reads(tx_execution.metrics().account_reads, function);
+                    metrics::inc_executor_local_transaction_slot_reads(tx_execution.metrics().slot_reads, function);
                     metrics::inc_executor_local_transaction_gas(tx_execution.execution().gas.as_u64() as usize, true, function);
                 }
                 Err(_) => {
@@ -548,18 +550,19 @@ impl Executor {
         };
         let evm_result = self.evms.execute(evm_input, evm_route);
 
+        // track metrics
         #[cfg(feature = "metrics")]
         {
-            let function = call_input.solidity_signature();
+            let function = codegen::function_sig_for_o11y(&call_input.data);
             match &evm_result {
                 Ok(evm_result) => {
-                    metrics::inc_executor_local_call(start.elapsed(), true, function.clone());
-                    metrics::inc_executor_local_call_account_reads(evm_result.metrics.account_reads, function.clone());
-                    metrics::inc_executor_local_call_slot_reads(evm_result.metrics.slot_reads, function.clone());
+                    metrics::inc_executor_local_call(start.elapsed(), true, function);
+                    metrics::inc_executor_local_call_account_reads(evm_result.metrics.account_reads, function);
+                    metrics::inc_executor_local_call_slot_reads(evm_result.metrics.slot_reads, function);
                     metrics::inc_executor_local_call_gas(evm_result.execution.gas.as_u64() as usize, function);
                 }
                 Err(_) => {
-                    metrics::inc_executor_local_call(start.elapsed(), false, function.clone());
+                    metrics::inc_executor_local_call(start.elapsed(), false, function);
                 }
             }
         }