feat: JournalDb trait (#13)

* feat: JournalDb trait

* feat: ingestor task

* refactor: introduce a chill type alias

Author: prestwich

Cargo.toml

@@ -53,7 +53,7 @@ signet-zenith = "0.9.0"
 ajj = { version = "0.3.4" }
 
 # trevm
-trevm = { version = "0.27.0", features = ["full_env_cfg"] }
+trevm = { version = "0.27.1", features = ["full_env_cfg"] }
 
 # Alloy periphery crates
 alloy = { version = "1.0.19", features = [

crates/db/Cargo.toml

@@ -11,11 +11,12 @@ repository.workspace = true
 [dependencies]
 signet-node-types.workspace = true
 
-
 signet-evm.workspace = true
 signet-types.workspace = true
 signet-zenith.workspace = true
 
+trevm.workspace = true
+
 alloy.workspace = true
 
 reth.workspace = true
@@ -24,4 +25,7 @@ reth-prune-types.workspace = true
 
 itertools.workspace = true
 serde.workspace = true
-tracing.workspace = true
+tracing.workspace = true
+futures-util = "0.3.31"
+tokio.workspace = true
+auto_impl = "1.3.0"

crates/db/src/aliases.rs

@@ -0,0 +1,5 @@
+use reth::providers::DatabaseProviderRW;
+use signet_node_types::SignetNodeTypes;
+
+/// A Convenience alias for a [`DatabaseProviderRW`] using [`SignetNodeTypes`].
+pub type SignetDbRw<Db> = DatabaseProviderRW<Db, SignetNodeTypes<Db>>;

crates/db/src/chain.rs

@@ -3,7 +3,14 @@ use reth::providers::Chain;
 use signet_zenith::{Passage, Transactor, Zenith};
 use std::collections::BTreeMap;
 
-/// The host extraction contents for a block.
+/// Host extraction contrents for a single block.
+///
+/// This is a container type for DB operations. It is held by the [`RuChain`]
+/// struct, and constructed during the [`RuWriter::get_extraction_results`]
+/// method.
+///
+/// [`RuWriter::get_extraction_results`]:
+/// crate::traits::RuWriter::get_extraction_results
 #[derive(Debug, Clone, Default, PartialEq, Eq)]
 pub struct DbExtractionResults {
     /// The zenith header for the block.
@@ -16,7 +23,8 @@ pub struct DbExtractionResults {
     pub enter_tokens: Vec<Passage::EnterToken>,
 }
 
-/// Equivalent of [`Chain`] but also containing zenith events and information.
+/// Equivalent of [`Chain`] but also containing [`DbExtractionResults`] for
+/// each block.
 #[derive(Debug, Clone, Default, PartialEq, Eq)]
 pub struct RuChain {
     /// Inner chain of RU blocks.

crates/db/src/journal/mod.rs

@@ -0,0 +1,78 @@
+//! Utilities for working with Signet journals in a reth database.
+
+mod r#trait;
+pub use r#trait::JournalDb;
+
+use crate::SignetDbRw;
+use futures_util::{Stream, StreamExt};
+use reth::providers::ProviderResult;
+use signet_node_types::NodeTypesDbTrait;
+use std::sync::Arc;
+use tokio::task::JoinHandle;
+use trevm::journal::BlockUpdate;
+
+/// A task that ingests journals into a reth database.
+#[derive(Debug)]
+pub struct JournalIngestor<Db: NodeTypesDbTrait> {
+    db: Arc<SignetDbRw<Db>>,
+}
+
+impl<Db: NodeTypesDbTrait> From<SignetDbRw<Db>> for JournalIngestor<Db> {
+    fn from(value: SignetDbRw<Db>) -> Self {
+        Self::new(value.into())
+    }
+}
+
+impl<Db: NodeTypesDbTrait> From<Arc<SignetDbRw<Db>>> for JournalIngestor<Db> {
+    fn from(value: Arc<SignetDbRw<Db>>) -> Self {
+        Self::new(value)
+    }
+}
+
+impl<Db: NodeTypesDbTrait> JournalIngestor<Db> {
+    /// Create a new `JournalIngestor` with the given database provider.
+    pub const fn new(db: Arc<SignetDbRw<Db>>) -> Self {
+        Self { db }
+    }
+
+    async fn task_future<S>(self, mut stream: S) -> ProviderResult<()>
+    where
+        S: Stream<Item = (alloy::consensus::Header, BlockUpdate<'static>)> + Send + Unpin + 'static,
+    {
+        while let Some(item) = stream.next().await {
+            // FUTURE: Sanity check that the header height matches the update
+            // height. Sanity check that both heights are 1 greater than the
+            // last height in the database.
+
+            let db = self.db.clone();
+            let (header, block_update) = item;
+
+            // DB interaction is sync, so we spawn a blocking task for it. We
+            // immediately await that task. This prevents blocking the worker
+            // thread
+            tokio::task::spawn_blocking(move || db.ingest(&header, block_update))
+                .await
+                .expect("ingestion should not panic")?;
+        }
+        // Stream has ended, return Ok
+        Ok(())
+    }
+
+    /// Spawn a task to ingest journals from the provided stream.
+    pub fn spawn<S>(self, stream: S) -> JoinHandle<ProviderResult<()>>
+    where
+        S: Stream<Item = (alloy::consensus::Header, BlockUpdate<'static>)> + Send + Unpin + 'static,
+    {
+        tokio::spawn(self.task_future(stream))
+    }
+}
+
+/// Ingest journals from a stream into a reth database.
+pub async fn ingest_journals<Db, S>(db: Arc<SignetDbRw<Db>>, stream: S) -> ProviderResult<()>
+where
+    Db: NodeTypesDbTrait,
+    S: Stream<Item = (alloy::consensus::Header, BlockUpdate<'static>)> + Send + Unpin + 'static,
+{
+    let ingestor = JournalIngestor::new(db);
+    ingestor.task_future(stream).await
+}

crates/db/src/journal/trait.rs

@@ -0,0 +1,47 @@
+use crate::RuWriter;
+use alloy::consensus::{BlockHeader, Header};
+use reth::{providers::ProviderResult, revm::db::BundleState};
+use signet_evm::{BlockResult, ExecutionOutcome};
+use signet_types::primitives::{RecoveredBlock, SealedBlock, SealedHeader, TransactionSigned};
+use trevm::journal::BlockUpdate;
+
+/// A database that can be updated with journals.
+pub trait JournalDb: RuWriter {
+    /// Ingest a journal into the database.
+    ///
+    /// This will create a [`BlockResult`] from the provided header and update,
+    /// and append it to the database using [`RuWriter::append_host_block`].
+    ///
+    /// This DOES NOT update tables containing historical transactions,
+    /// receipts, events, etc. It only updates tables related to headers,
+    /// and state.
+    ///
+    /// This is intended to be used for tx simulation, and other purposes that
+    /// need fast state access WITHTOUT needing to retrieve historical data.
+    fn ingest(&self, header: &Header, update: BlockUpdate<'_>) -> ProviderResult<()> {
+        let journal_hash = update.journal_hash();
+
+        // TODO: remove the clone in future versions. This can be achieved by
+        // _NOT_ making a `BlockResult` and instead manually updating relevan
+        // tables. However, this means diverging more fro the underlying reth
+        // logic that we are currently re-using.
+        let bundle_state: BundleState = update.journal().clone().into();
+        let execution_outcome = ExecutionOutcome::new(bundle_state, vec![], header.number());
+
+        let block: SealedBlock<TransactionSigned, Header> =
+            SealedBlock { header: SealedHeader::new(header.to_owned()), body: Default::default() };
+        let block_result =
+            BlockResult { sealed_block: RecoveredBlock::new(block, vec![]), execution_outcome };
+
+        self.append_host_block(
+            None,
+            std::iter::empty(),
+            std::iter::empty(),
+            std::iter::empty(),
+            &block_result,
+            journal_hash,
+        )
+    }
+}
+
+impl<T> JournalDb for T where T: RuWriter {}

crates/db/src/lib.rs

@@ -11,12 +11,17 @@
 #![deny(unused_must_use, rust_2018_idioms)]
 #![cfg_attr(docsrs, feature(doc_cfg, doc_auto_cfg))]
 
+mod aliases;
+pub use aliases::SignetDbRw;
+
 mod chain;
 pub use chain::{DbExtractionResults, RuChain};
 
 mod convert;
 pub use convert::DataCompat;
 
+pub mod journal;
+
 mod provider;
 
 mod tables;

crates/db/src/provider.rs

@@ -1,5 +1,5 @@
 use crate::{
-    DataCompat, DbZenithHeader, RuChain, ZenithHeaders,
+    DataCompat, DbZenithHeader, RuChain, SignetDbRw, ZenithHeaders,
     tables::{DbSignetEvent, JournalHashes, SignetEvents},
     traits::RuWriter,
 };
@@ -11,9 +11,9 @@ use reth::{
     primitives::{Account, StaticFileSegment},
     providers::{
         AccountReader, BlockBodyIndicesProvider, BlockNumReader, BlockReader, BlockWriter, Chain,
-        DBProvider, DatabaseProviderRW, HistoryWriter, OriginalValuesKnown, ProviderError,
-        ProviderResult, StageCheckpointWriter, StateWriter, StaticFileProviderFactory,
-        StaticFileWriter, StorageLocation,
+        DBProvider, HistoryWriter, OriginalValuesKnown, ProviderError, ProviderResult,
+        StageCheckpointWriter, StateWriter, StaticFileProviderFactory, StaticFileWriter,
+        StorageLocation,
     },
 };
 use reth_db::{
@@ -25,7 +25,7 @@ use reth_db::{
 };
 use reth_prune_types::{MINIMUM_PRUNING_DISTANCE, PruneMode};
 use signet_evm::BlockResult;
-use signet_node_types::{NodeTypesDbTrait, SignetNodeTypes};
+use signet_node_types::NodeTypesDbTrait;
 use signet_types::primitives::RecoveredBlock;
 use signet_zenith::{
     Passage::{self, Enter, EnterToken},
@@ -35,7 +35,7 @@ use signet_zenith::{
 use std::ops::RangeInclusive;
 use tracing::{debug, instrument, trace, warn};
 
-impl<Db> RuWriter for DatabaseProviderRW<Db, SignetNodeTypes<Db>>
+impl<Db> RuWriter for SignetDbRw<Db>
 where
     Db: NodeTypesDbTrait,
 {
@@ -430,7 +430,6 @@ where
     /// see the documentation for each function.
     fn append_host_block(
         &self,
-        host_height: u64,
         header: Option<Zenith::BlockHeader>,
         transacts: impl IntoIterator<Item = Transact>,
         enters: impl IntoIterator<Item = Passage::Enter>,
@@ -480,7 +479,7 @@ where
 
         self.update_pipeline_stages(ru_height, false)?;
 
-        debug!(target: "signet_db_lifecycle", host_height, ru_height, "Appended blocks");
+        debug!(target: "signet_db_lifecycle", ru_height, "Appended blocks");
 
         Ok(())
     }

crates/db/src/traits.rs

@@ -1,19 +1,20 @@
-use crate::{DbExtractionResults, DbSignetEvent, RuChain};
+use crate::{DbExtractionResults, DbSignetEvent, RuChain, SignetDbRw};
 use alloy::primitives::{Address, B256, BlockNumber, U256};
 use itertools::Itertools;
 use reth::{
     primitives::Account,
-    providers::{DatabaseProviderRW, OriginalValuesKnown, ProviderResult, StorageLocation},
+    providers::{OriginalValuesKnown, ProviderResult, StorageLocation},
 };
 use reth_db::models::StoredBlockBodyIndices;
 use signet_evm::BlockResult;
-use signet_node_types::{NodeTypesDbTrait, SignetNodeTypes};
+use signet_node_types::NodeTypesDbTrait;
 use signet_types::primitives::RecoveredBlock;
 use signet_zenith::{Passage, Transactor, Zenith};
 use std::{collections::BTreeMap, ops::RangeInclusive};
 use tracing::trace;
 
 /// Writer for [`Passage::Enter`] events.
+#[auto_impl::auto_impl(&, Arc, Box)]
 pub trait RuWriter {
     /// Get the last block number
     fn last_block_number(&self) -> ProviderResult<BlockNumber>;
@@ -254,7 +255,6 @@ pub trait RuWriter {
     #[allow(clippy::too_many_arguments)]
     fn append_host_block(
         &self,
-        host_height: u64,
         header: Option<Zenith::BlockHeader>,
         transacts: impl IntoIterator<Item = Transactor::Transact>,
         enters: impl IntoIterator<Item = Passage::Enter>,
@@ -293,7 +293,7 @@ pub trait RuWriter {
 }
 
 /// Extend the [`DatabaseProviderRW`] with a guarded commit function.
-pub trait DbProviderExt<Db>: Into<DatabaseProviderRW<Db, SignetNodeTypes<Db>>>
+pub trait DbProviderExt<Db>: Into<SignetDbRw<Db>>
 where
     Db: NodeTypesDbTrait,
 {
@@ -302,18 +302,18 @@ where
     /// transaction is rolled back.
     fn update(
         self,
-        f: impl FnOnce(&mut DatabaseProviderRW<Db, SignetNodeTypes<Db>>) -> ProviderResult<()>,
+        f: impl FnOnce(&mut SignetDbRw<Db>) -> ProviderResult<()>,
     ) -> ProviderResult<()>;
 }
 
 impl<T, Db> DbProviderExt<Db> for T
 where
     Db: NodeTypesDbTrait,
-    T: Into<DatabaseProviderRW<Db, SignetNodeTypes<Db>>>,
+    T: Into<SignetDbRw<Db>>,
 {
     fn update(
         self,
-        f: impl FnOnce(&mut DatabaseProviderRW<Db, SignetNodeTypes<Db>>) -> ProviderResult<()>,
+        f: impl FnOnce(&mut SignetDbRw<Db>) -> ProviderResult<()>,
     ) -> ProviderResult<()> {
         let mut this = self.into();
         f(&mut this)?;