feat(storage) L1BlockManager

Cargo.toml

@@ -37,6 +37,7 @@ members = [
   "crates/state",
   "crates/status",
   "crates/storage",
+  "crates/macros",
   "crates/sync",
   "crates/tasks",
   "crates/test-utils",
@@ -134,10 +135,13 @@ alloy-rpc-types-eth = { version = "0.4.2", default-features = false, features =
 ] }
 alloy-serde = { version = "0.4.2", default-features = false }
 alloy-sol-types = "0.8.0"
+proc-macro2 = "1.0"
+quote = "1.0"
 revm = { version = "14.0.3", features = ["std"], default-features = false }
 revm-primitives = { version = "10.0.0", features = [
   "std",
 ], default-features = false }
+syn = "2.0"
 
 # reth itself:
 reth = { git = "https://github.com/paradigmxyz/reth.git", rev = "v1.1.0" }

crates/db/Cargo.toml

@@ -20,6 +20,7 @@ parking_lot.workspace = true
 serde.workspace = true
 thiserror.workspace = true
 tracing.workspace = true
+strata-macros = { version = "0.1.0", path = "../macros" }
 
 [features]
 default = []

crates/db/src/traits.rs

@@ -4,6 +4,7 @@
 use std::sync::Arc;
 
 use borsh::{BorshDeserialize, BorshSerialize};
+use strata_macros::gen_async_version;
 use strata_mmr::CompactMmr;
 use strata_primitives::{
     l1::*,
@@ -43,6 +44,7 @@ pub trait Database {
 }
 
 /// Database interface to control our view of L1 data.
+#[gen_async_version]
 pub trait L1Database {
     /// Atomically extends the chain with a new block, providing the manifest
     /// and a list of transactions we find relevant.  Returns error if
@@ -282,7 +284,7 @@ pub trait ProofDatabase {
     /// Deletes a proof by its key.
     ///
     /// Tries to delete a proof by its key, returning if it really
-    /// existed or not.  
+    /// existed or not.
     fn del_proof(&self, proof_key: ProofKey) -> DbResult<bool>;
 
     /// Inserts dependencies for a given [`ProofContext`] into the database.
@@ -298,7 +300,7 @@ pub trait ProofDatabase {
     /// Deletes dependencies for a given [`ProofContext`].
     ///
     /// Tries to delete dependencies of by its context, returning if it really
-    /// existed or not.  
+    /// existed or not.
     fn del_proof_deps(&self, proof_context: ProofContext) -> DbResult<bool>;
 }
 

crates/macros/Cargo.toml

@@ -0,0 +1,12 @@
+[package]
+name = "strata-macros"
+version = "0.1.0"
+edition = "2024"
+
+[dependencies]
+proc-macro2.workspace = true
+quote.workspace = true
+syn.workspace = true
+
+[lib]
+proc-macro = true

crates/macros/src/lib.rs

@@ -0,0 +1,49 @@
+use proc_macro::TokenStream;
+use quote::quote;
+use syn::{Attribute, ItemTrait, parse_macro_input};
+
+#[proc_macro_attribute]
+pub fn gen_async_version(_attr: TokenStream, item: TokenStream) -> TokenStream {
+    // Parse the input tokens into a syntax tree
+    let input = parse_macro_input!(item as ItemTrait);
+
+    // Extract the trait name
+    let trait_name = &input.ident;
+    let async_trait_name = syn::Ident::new(&format!("{}Async", trait_name), trait_name.span());
+
+    // Generate async methods
+    let async_methods = input.items.iter().filter_map(|item| {
+        if let syn::TraitItem::Fn(method) = item {
+            let sig = &method.sig;
+            let async_sig = syn::Signature {
+                asyncness: Some(Default::default()), // Add async keyword
+                ..sig.clone()
+            };
+
+            let docs: Vec<&Attribute> = method
+                .attrs
+                .iter()
+                .filter(|attr| attr.path().is_ident("doc"))
+                .collect();
+
+            Some(quote! {
+                #(#docs)*
+                #async_sig;
+            })
+        } else {
+            None
+        }
+    });
+
+    // Generate the new async trait
+    let expanded = quote! {
+        #input
+
+        pub trait #async_trait_name {
+            #(#async_methods)*
+        }
+    };
+
+    // Convert the expanded code into a TokenStream and return it
+    TokenStream::from(expanded)
+}

crates/storage/Cargo.toml

@@ -17,3 +17,4 @@ paste.workspace = true
 threadpool.workspace = true
 tokio.workspace = true
 tracing.workspace = true
+strata-mmr.workspace = true

crates/storage/src/managers/l1.rs

@@ -0,0 +1,124 @@
+use std::sync::Arc;
+
+use strata_db::{
+    traits::{BlockStatus, Database, L1Database, L1DatabaseAsync},
+    DbResult,
+};
+use strata_mmr::CompactMmr;
+use strata_primitives::{
+    buf::Buf32,
+    l1::{L1BlockManifest, L1TxRef},
+};
+use strata_state::{block::L2BlockBundle, header::L2Header, id::L2BlockId, l1::L1Tx};
+use threadpool::ThreadPool;
+use tokio::sync::oneshot::{self, error::RecvError};
+use tracing::warn;
+
+use crate::cache::{self, CacheTable};
+
+/// Caching manager of L1 blocks in the block database.
+pub struct L1BlockManager<DB>
+where
+    DB: L1Database + Sync + Send + 'static,
+{
+    pool: ThreadPool,
+    db: Arc<DB>,
+    block_cache: CacheTable<L2BlockId, Option<L2BlockBundle>>,
+}
+
+impl<DB> L1BlockManager<DB>
+where
+    DB: L1Database + Sync + Send + 'static,
+{
+    pub fn new(pool: ThreadPool, db: Arc<DB>) -> Self {
+        Self {
+            pool,
+            db,
+            block_cache: CacheTable::new(64.try_into().unwrap()),
+        }
+    }
+}
+
+impl<DB> L1DatabaseAsync for L1BlockManager<DB>
+where
+    DB: L1Database + Sync + Send + 'static,
+{
+    async fn put_block_data(&self, idx: u64, mf: L1BlockManifest, txs: Vec<L1Tx>) -> DbResult<()> {
+        let db = self.db.clone();
+        self.pool
+            .spawn(move || db.put_block_data(idx, mf, txs))
+            .await
+    }
+
+    async fn put_mmr_checkpoint(&self, idx: u64, mmr: CompactMmr) -> DbResult<()> {
+        let db = self.db.clone();
+        self.pool
+            .spawn(move || db.put_mmr_checkpoint(idx, mmr))
+            .await
+    }
+
+    async fn revert_to_height(&self, idx: u64) -> DbResult<()> {
+        let db = self.db.clone();
+        self.pool.spawn(move || db.revert_to_height(idx)).await
+    }
+
+    async fn get_chain_tip(&self) -> DbResult<Option<u64>> {
+        let db = self.db.clone();
+        self.pool.spawn(move || db.get_chain_tip()).await
+    }
+
+    async fn get_block_manifest(&self, idx: u64) -> DbResult<Option<L1BlockManifest>> {
+        let db = self.db.clone();
+        self.pool.spawn(move || db.get_block_manifest(idx)).await
+    }
+
+    async fn get_blockid_range(&self, start_idx: u64, end_idx: u64) -> DbResult<Vec<Buf32>> {
+        let db = self.db.clone();
+        self.pool
+            .spawn(move || db.get_blockid_range(start_idx, end_idx))
+            .await
+    }
+
+    async fn get_block_txs(&self, idx: u64) -> DbResult<Option<Vec<L1TxRef>>> {
+        let db = self.db.clone();
+        self.pool.spawn(move || db.get_block_txs(idx)).await
+    }
+
+    async fn get_tx(&self, tx_ref: L1TxRef) -> DbResult<Option<L1Tx>> {
+        let db = self.db.clone();
+        self.pool.spawn(move || db.get_tx(tx_ref)).await
+    }
+
+    async fn get_last_mmr_to(&self, idx: u64) -> DbResult<Option<CompactMmr>> {
+        let db = self.db.clone();
+        self.pool.spawn(move || db.get_last_mmr_to(idx)).await
+    }
+
+    async fn get_txs_from(&self, start_idx: u64) -> DbResult<(Vec<L1Tx>, u64)> {
+        let db = self.db.clone();
+        self.pool.spawn(move || db.get_txs_from(start_idx)).await
+    }
+}
+
+trait ThreadPoolSpawn {
+    async fn spawn<T, F>(&self, func: F) -> T
+    where
+        F: FnOnce() -> T + Send + 'static,
+        T: Send + 'static;
+}
+
+impl ThreadPoolSpawn for ThreadPool {
+    async fn spawn<T, F>(&self, func: F) -> T
+    where
+        F: FnOnce() -> T + Send + 'static,
+        T: Send + 'static,
+    {
+        let (tx, rx) = oneshot::channel();
+        self.execute(move || {
+            if tx.send(func()).is_err() {
+                warn!("failed to send response")
+            }
+        });
+        rx.await.expect("Sender was dropped without sending")
+    }
+}

crates/storage/src/managers/mod.rs

@@ -1,2 +1,3 @@
 pub mod checkpoint;
+pub mod l1;
 pub mod l2;