feat(l1,l2): make write path APIs async (#2336)

**Motivation**

Some of our sync APIs can produce starving when running on Tokio due to
taking a long time to reach the next `await`-point.
Specifically, writing to the DB tends to take a long time, which blocks
other tasks, sometimes the whole runtime due to how the scheduler in
Tokio works.
Thus, we need a way to inform the runtime we're going to be working for
a while, and give it control while we wait for stuff.

**Description**

Take the mutable APIs for the DB and mark them `async`. Then bubble that
up to their users. Then make the functions non-blocking by using
`spawn_blocking` to run on the blocking thread, releasing the runtime to
handle more work.
The DB writing APIs had to change to pass-by-value to satisfy the
borrow-checker in the blocking task context. I think I can use proper
lifetime bounds with a helper crate, if that's preferred. The values
were already being discarded after passing to the DB, so passing by
value should not be a problem either way.

Special considerations:
- For some work performed before benchmarks and EF tests which are
inherently synchronous I opted for calling with an ad-hoc runtime
instance and `block_on`, as that might reduce the changes needed by
localizing the async work. If desired, that can be changed up to making
a `tokio::main`. The same is true for some setup functions for tests.
- For the DBs I had to separate the Tokio import. This is because they
need to compile with L2, which means provers' custom compilers, and
those don't support the networking functions in the stdlib, which Tokio
with full features (as the workspace dep declares) brings them in.
- The InMemoryDB was left untouched other than updating the interfaces,
given hashmap access should be quick enough.
- I need to comment on [this
hack](https://github.com/lambdaclass/ethrex/pull/2336/files#diff-264636d3ee6ee67bd6e136b8c98f74152de6a8e2a07f597cfb5f622d4e0d815aR143-R146):
`and_then` can't be used on futures and everything became a mess without
that little helper.
- I'm unsure about whether or not we also want to cover the read APIs,
at least for consistency I would think so, but for now I left them out.

closes #2402

Author: Oppen

CHANGELOG.md

@@ -2,6 +2,10 @@
 
 ## Perf
 
+#### 2025-04-01
+
+- Asyncify DB write APIs, as well as its users [#2336](https://github.com/lambdaclass/ethrex/pull/2336)
+
 #### 2025-03-30
 
 - Faster block import, use a slice instead of copy

Cargo.lock

@@ -51,6 +51,7 @@ ethrex-prover = { path = "./crates/l2/prover" }
 tracing = { version = "0.1", features = ["log"] }
 tracing-subscriber = { version = "0.3.0", features = ["env-filter"] }
 
+async-trait = "0.1.88"
 ethereum-types = { version = "0.15.1", features = ["serialize"] }
 serde = { version = "1.0.203", features = ["derive"] }
 serde_json = "1.0.117"

Cargo.toml

@@ -15,19 +15,19 @@ use tracing_subscriber::{filter::Directive, EnvFilter, FmtSubscriber};
 fn block_import() {
     let data_dir = DEFAULT_DATADIR;
     set_datadir(data_dir);
-
     remove_db(data_dir);
 
     let evm_engine = "revm".to_owned().try_into().unwrap();
 
     let network = "../../test_data/genesis-l2-ci.json";
 
-    import_blocks(
+    let rt = tokio::runtime::Runtime::new().unwrap();
+    rt.block_on(import_blocks(
         "../../test_data/l2-1k-erc20.rlp",
         data_dir,
         network,
         evm_engine,
-    );
+    ));
 }
 
 pub fn criterion_benchmark(c: &mut Criterion) {

bench/criterion_benchmark.rs

@@ -15,6 +15,7 @@ serde_json.workspace = true
 bytes.workspace = true
 hex.workspace = true
 lazy_static.workspace = true
+tokio.workspace = true
 
 [dev-dependencies]
 datatest-stable = "0.2.9"

cmd/ef_tests/blockchain/Cargo.toml

@@ -8,14 +8,14 @@ use ethrex_common::types::{
 use ethrex_rlp::decode::RLPDecode;
 use ethrex_storage::{EngineType, Store};
 
-pub fn run_ef_test(test_key: &str, test: &TestUnit) {
+pub async fn run_ef_test(test_key: &str, test: &TestUnit) {
     // check that the decoded genesis block header matches the deserialized one
     let genesis_rlp = test.genesis_rlp.clone();
     let decoded_block = CoreBlock::decode(&genesis_rlp).unwrap();
     let genesis_block_header = CoreBlockHeader::from(test.genesis_block_header.clone());
     assert_eq!(decoded_block.header, genesis_block_header);
 
-    let store = build_store_for_test(test);
+    let store = build_store_for_test(test).await;
 
     // Check world_state
     check_prestate_against_db(test_key, test, &store);
@@ -33,7 +33,7 @@ pub fn run_ef_test(test_key: &str, test: &TestUnit) {
         let hash = block.hash();
 
         // Attempt to add the block as the head of the chain
-        let chain_result = blockchain.add_block(block);
+        let chain_result = blockchain.add_block(block).await;
         match chain_result {
             Err(error) => {
                 assert!(
@@ -50,7 +50,7 @@ pub fn run_ef_test(test_key: &str, test: &TestUnit) {
                     test_key,
                     block_fixture.expect_exception.clone().unwrap()
                 );
-                apply_fork_choice(&store, hash, hash, hash).unwrap();
+                apply_fork_choice(&store, hash, hash, hash).await.unwrap();
             }
         }
     }
@@ -82,12 +82,13 @@ pub fn parse_test_file(path: &Path) -> HashMap<String, TestUnit> {
 }
 
 /// Creats a new in-memory store and adds the genesis state
-pub fn build_store_for_test(test: &TestUnit) -> Store {
+pub async fn build_store_for_test(test: &TestUnit) -> Store {
     let store =
         Store::new("store.db", EngineType::InMemory).expect("Failed to build DB for testing");
     let genesis = test.get_genesis();
     store
         .add_initial_state(genesis)
+        .await
         .expect("Failed to add genesis state");
     store
 }

cmd/ef_tests/blockchain/test_runner.rs

@@ -14,6 +14,7 @@ use std::path::Path;
 // test or set of tests
 
 fn parse_and_execute_until_cancun(path: &Path) -> datatest_stable::Result<()> {
+    let rt = tokio::runtime::Runtime::new().unwrap();
     let tests = parse_test_file(path);
 
     for (test_key, test) in tests {
@@ -22,22 +23,23 @@ fn parse_and_execute_until_cancun(path: &Path) -> datatest_stable::Result<()> {
             // them. This produces false positives
             continue;
         }
-        run_ef_test(&test_key, &test);
+        rt.block_on(run_ef_test(&test_key, &test));
     }
 
     Ok(())
 }
 
 #[allow(dead_code)]
 fn parse_and_execute_all(path: &Path) -> datatest_stable::Result<()> {
+    let rt = tokio::runtime::Runtime::new().unwrap();
     let tests = parse_test_file(path);
 
     for (test_key, test) in tests {
         if test.network < Network::Merge {
             // These tests fall into the not supported forks. This produces false positives
             continue;
         }
-        run_ef_test(&test_key, &test);
+        rt.block_on(run_ef_test(&test_key, &test));
     }
 
     Ok(())

cmd/ef_tests/blockchain/tests/cancun.rs

@@ -13,6 +13,7 @@ const SKIPPED_TEST: [&str; 2] = [
 
 #[allow(dead_code)]
 fn parse_and_execute(path: &Path) -> datatest_stable::Result<()> {
+    let rt = tokio::runtime::Runtime::new().unwrap();
     let tests = parse_test_file(path);
 
     for (test_key, test) in tests {
@@ -21,7 +22,7 @@ fn parse_and_execute(path: &Path) -> datatest_stable::Result<()> {
             continue;
         }
 
-        run_ef_test(&test_key, &test);
+        rt.block_on(run_ef_test(&test_key, &test));
     }
     Ok(())
 }

cmd/ef_tests/blockchain/tests/prague.rs

@@ -6,6 +6,7 @@ use ef_tests_blockchain::{
 };
 
 fn parse_and_execute(path: &Path) -> datatest_stable::Result<()> {
+    let rt = tokio::runtime::Runtime::new().unwrap();
     let tests = parse_test_file(path);
 
     for (test_key, test) in tests {
@@ -14,7 +15,7 @@ fn parse_and_execute(path: &Path) -> datatest_stable::Result<()> {
             continue;
         }
 
-        run_ef_test(&test_key, &test);
+        rt.block_on(run_ef_test(&test_key, &test));
     }
     Ok(())
 }

cmd/ef_tests/blockchain/tests/shanghai.rs

@@ -29,6 +29,7 @@ revm = { version = "19.0.0", features = [
     "optional_no_base_fee",
     "optional_block_gas_limit",
 ], default-features = false }
+tokio.workspace = true
 
 [dev-dependencies]
 hex = "0.4.3"

cmd/ef_tests/state/Cargo.toml

@@ -18,7 +18,7 @@ use ethrex_vm::backends;
 use keccak_hash::keccak;
 use std::collections::HashMap;
 
-pub fn run_ef_test(test: &EFTest) -> Result<EFTestReport, EFTestRunnerError> {
+pub async fn run_ef_test(test: &EFTest) -> Result<EFTestReport, EFTestRunnerError> {
     // There are some tests that don't have a hash, unwrap will panic
     let hash = test
         ._info
@@ -35,7 +35,7 @@ pub fn run_ef_test(test: &EFTest) -> Result<EFTestReport, EFTestRunnerError> {
             if !test.post.has_vector_for_fork(vector, *fork) {
                 continue;
             }
-            match run_ef_test_tx(vector, test, fork) {
+            match run_ef_test_tx(vector, test, fork).await {
                 Ok(_) => continue,
                 Err(EFTestRunnerError::VMInitializationFailed(reason)) => {
                     ef_test_report_fork.register_vm_initialization_failure(reason, *vector);
@@ -78,16 +78,16 @@ pub fn run_ef_test(test: &EFTest) -> Result<EFTestReport, EFTestRunnerError> {
     Ok(ef_test_report)
 }
 
-pub fn run_ef_test_tx(
+pub async fn run_ef_test_tx(
     vector: &TestVector,
     test: &EFTest,
     fork: &Fork,
 ) -> Result<(), EFTestRunnerError> {
-    let mut db = utils::load_initial_state_levm(test);
+    let mut db = utils::load_initial_state_levm(test).await;
     let mut levm = prepare_vm_for_tx(vector, test, fork, &mut db)?;
     ensure_pre_state(&levm, test)?;
     let levm_execution_result = levm.execute();
-    ensure_post_state(&levm_execution_result, vector, test, fork, &mut db)?;
+    ensure_post_state(&levm_execution_result, vector, test, fork, &mut db).await?;
     Ok(())
 }
 
@@ -282,7 +282,7 @@ fn exception_is_expected(
     })
 }
 
-pub fn ensure_post_state(
+pub async fn ensure_post_state(
     levm_execution_result: &Result<ExecutionReport, VMError>,
     vector: &TestVector,
     test: &EFTest,
@@ -326,7 +326,7 @@ pub fn ensure_post_state(
                                     .to_owned(),
                             )
                         })?;
-                    let pos_state_root = post_state_root(&levm_account_updates, test);
+                    let pos_state_root = post_state_root(&levm_account_updates, test).await;
                     let expected_post_state_root_hash =
                         test.post.vector_post_value(vector, *fork).hash;
                     if expected_post_state_root_hash != pos_state_root {
@@ -381,12 +381,13 @@ pub fn ensure_post_state(
     Ok(())
 }
 
-pub fn post_state_root(account_updates: &[AccountUpdate], test: &EFTest) -> H256 {
-    let (initial_state, block_hash) = utils::load_initial_state(test);
+pub async fn post_state_root(account_updates: &[AccountUpdate], test: &EFTest) -> H256 {
+    let (initial_state, block_hash) = utils::load_initial_state(test).await;
     initial_state
         .database()
         .unwrap()
         .apply_account_updates(block_hash, account_updates)
+        .await
         .unwrap()
         .unwrap()
 }

cmd/ef_tests/state/runner/levm_runner.rs

@@ -70,27 +70,27 @@ pub struct EFTestRunnerOptions {
     pub revm: bool,
 }
 
-pub fn run_ef_tests(
+pub async fn run_ef_tests(
     ef_tests: Vec<EFTest>,
     opts: &EFTestRunnerOptions,
 ) -> Result<(), EFTestRunnerError> {
     let mut reports = report::load()?;
     if reports.is_empty() {
         if opts.revm {
-            run_with_revm(&mut reports, &ef_tests, opts)?;
+            run_with_revm(&mut reports, &ef_tests, opts).await?;
             return Ok(());
         } else {
-            run_with_levm(&mut reports, &ef_tests, opts)?;
+            run_with_levm(&mut reports, &ef_tests, opts).await?;
         }
     }
     if opts.summary {
         return Ok(());
     }
-    re_run_with_revm(&mut reports, &ef_tests, opts)?;
+    re_run_with_revm(&mut reports, &ef_tests, opts).await?;
     write_report(&reports)
 }
 
-fn run_with_levm(
+async fn run_with_levm(
     reports: &mut Vec<EFTestReport>,
     ef_tests: &[EFTest],
     opts: &EFTestRunnerOptions,
@@ -113,7 +113,7 @@ fn run_with_levm(
         if !opts.spinner && opts.verbose {
             println!("Running test: {:?}", test.name);
         }
-        let ef_test_report = match levm_runner::run_ef_test(test) {
+        let ef_test_report = match levm_runner::run_ef_test(test).await {
             Ok(ef_test_report) => ef_test_report,
             Err(EFTestRunnerError::Internal(err)) => return Err(EFTestRunnerError::Internal(err)),
             non_internal_errors => {
@@ -154,7 +154,7 @@ fn run_with_levm(
 }
 
 /// ### Runs all tests with REVM
-fn run_with_revm(
+async fn run_with_revm(
     reports: &mut Vec<EFTestReport>,
     ef_tests: &[EFTest],
     opts: &EFTestRunnerOptions,
@@ -183,7 +183,7 @@ fn run_with_revm(
             ),
             opts.spinner,
         );
-        let ef_test_report = match revm_runner::_run_ef_test_revm(test) {
+        let ef_test_report = match revm_runner::_run_ef_test_revm(test).await {
             Ok(ef_test_report) => ef_test_report,
             Err(EFTestRunnerError::Internal(err)) => return Err(EFTestRunnerError::Internal(err)),
             non_internal_errors => {
@@ -210,7 +210,7 @@ fn run_with_revm(
     Ok(())
 }
 
-fn re_run_with_revm(
+async fn re_run_with_revm(
     reports: &mut [EFTestReport],
     ef_tests: &[EFTest],
     opts: &EFTestRunnerOptions,
@@ -262,7 +262,7 @@ fn re_run_with_revm(
                 })
                 .unwrap(),
             failed_test_report,
-        ) {
+        ).await {
             Ok(re_run_report) => {
                 failed_test_report.register_re_run_report(re_run_report.clone());
             }