refactor: make it depend on signet-journal

Author: prestwich

Cargo.toml

@@ -45,6 +45,7 @@ signet-bundle = "0.10.0"
 signet-constants = "0.10.0"
 signet-evm = "0.10.0"
 signet-extract = "0.10.0"
+signet-journal = "0.10.0"
 signet-tx-cache = "0.10.0"
 signet-types = "0.10.0"
 signet-zenith = "0.10.0"

crates/db/Cargo.toml

@@ -12,11 +12,10 @@ repository.workspace = true
 signet-node-types.workspace = true
 
 signet-evm.workspace = true
+signet-journal.workspace = true
 signet-types.workspace = true
 signet-zenith.workspace = true
 
-trevm.workspace = true
-
 alloy.workspace = true
 
 reth.workspace = true

crates/db/src/convert.rs

@@ -10,7 +10,7 @@
 use alloy::consensus::TxReceipt;
 
 /// Trait for types that can be converted into other types as they're already compatible.
-/// Uswed for converting between alloy/reth/signet types.
+/// Used for converting between alloy/reth/signet types.
 pub trait DataCompat<Other: DataCompat<Self>>: Sized {
     /// Convert `self` into the target type.
     fn convert(self) -> Other;

crates/db/src/journal/ingestor.rs

@@ -1,11 +1,10 @@
 use crate::{SignetDbRw, journal::JournalDb};
-use futures_util::{Stream, StreamExt};
+use futures_util::StreamExt;
 use reth::providers::ProviderResult;
+use signet_journal::{Journal, JournalStream};
 use signet_node_types::NodeTypesDbTrait;
-use signet_types::primitives::SealedHeader;
 use std::sync::Arc;
 use tokio::task::JoinHandle;
-use trevm::journal::BlockUpdate;
 
 /// A task that ingests journals into a reth database.
 #[derive(Debug)]
@@ -33,20 +32,19 @@ impl<Db: NodeTypesDbTrait> JournalIngestor<Db> {
 
     async fn task_future<S>(self, mut stream: S) -> ProviderResult<()>
     where
-        S: Stream<Item = (SealedHeader, BlockUpdate<'static>)> + Send + Unpin + 'static,
+        S: JournalStream<'static> + Send + Unpin + 'static,
     {
-        while let Some(item) = stream.next().await {
+        while let Some(Journal::V1(journal)) = 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))
+            tokio::task::spawn_blocking(move || db.ingest(journal))
                 .await
                 .expect("ingestion should not panic")?;
         }
@@ -57,7 +55,7 @@ impl<Db: NodeTypesDbTrait> JournalIngestor<Db> {
     /// Spawn a task to ingest journals from the provided stream.
     pub fn spawn<S>(self, stream: S) -> JoinHandle<ProviderResult<()>>
     where
-        S: Stream<Item = (SealedHeader, BlockUpdate<'static>)> + Send + Unpin + 'static,
+        S: JournalStream<'static> + Send + Unpin + 'static,
     {
         tokio::spawn(self.task_future(stream))
     }
@@ -67,7 +65,7 @@ impl<Db: NodeTypesDbTrait> JournalIngestor<Db> {
 pub async fn ingest_journals<Db, S>(db: Arc<SignetDbRw<Db>>, stream: S) -> ProviderResult<()>
 where
     Db: NodeTypesDbTrait,
-    S: Stream<Item = (SealedHeader, BlockUpdate<'static>)> + Send + Unpin + 'static,
+    S: JournalStream<'static> + Send + Unpin + 'static,
 {
     let ingestor = JournalIngestor::new(db);
     ingestor.task_future(stream).await

crates/db/src/journal/provider.rs

@@ -1,16 +1,16 @@
-use crate::{DataCompat, journal::JournalDb};
-use futures_util::{Stream, StreamExt};
+use crate::journal::JournalDb;
+use futures_util::StreamExt;
 use reth::{
+    primitives::SealedHeader,
     providers::{
         CanonChainTracker, DatabaseProviderFactory, DatabaseProviderRW, ProviderResult,
         providers::BlockchainProvider,
     },
     rpc::types::engine::ForkchoiceState,
 };
+use signet_journal::{Journal, JournalStream};
 use signet_node_types::{NodeTypesDbTrait, SignetNodeTypes};
-use signet_types::primitives::SealedHeader;
 use tokio::task::JoinHandle;
-use trevm::journal::BlockUpdate;
 
 /// A task that processes journal updates for a specific database, and calls
 /// the appropriate methods on a [`BlockchainProvider`] to update the in-memory
@@ -40,21 +40,20 @@ impl<Db: NodeTypesDbTrait> JournalProviderTask<Db> {
     /// task-spawning system.
     pub async fn task_future<S>(self, mut journals: S) -> ProviderResult<()>
     where
-        S: Stream<Item = (SealedHeader, BlockUpdate<'static>)> + Send + Unpin + 'static,
+        S: JournalStream<'static> + Send + Unpin + 'static,
     {
         loop {
-            let Some((header, block_update)) = journals.next().await else { break };
-
-            let block_hash = header.hash();
+            let Some(Journal::V1(journal)) = journals.next().await else { break };
 
             let rw = self.provider.database_provider_rw().map(DatabaseProviderRW);
 
-            let r_header = header.clone_convert();
+            let r_header = SealedHeader::new_unhashed(journal.header().clone());
+            let block_hash = r_header.hash();
 
             // 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 || rw?.ingest(header, block_update))
+            tokio::task::spawn_blocking(move || rw?.ingest(journal))
                 .await
                 .expect("ingestion should not panic")?;
 
@@ -74,7 +73,7 @@ impl<Db: NodeTypesDbTrait> JournalProviderTask<Db> {
     /// Spawn the journal provider task.
     pub fn spawn<S>(self, journals: S) -> JoinHandle<ProviderResult<()>>
     where
-        S: Stream<Item = (SealedHeader, BlockUpdate<'static>)> + Send + Unpin + 'static,
+        S: JournalStream<'static> + Send + Unpin + 'static,
     {
         tokio::spawn(self.task_future(journals))
     }

crates/db/src/journal/trait.rs

@@ -2,8 +2,8 @@ use crate::RuWriter;
 use alloy::consensus::{BlockHeader, Header};
 use reth::{providers::ProviderResult, revm::db::BundleState};
 use signet_evm::{BlockResult, ExecutionOutcome};
+use signet_journal::HostJournal;
 use signet_types::primitives::{RecoveredBlock, SealedBlock, SealedHeader, TransactionSigned};
-use trevm::journal::BlockUpdate;
 
 /// A database that can be updated with journals.
 pub trait JournalDb: RuWriter {
@@ -18,18 +18,21 @@ pub trait JournalDb: RuWriter {
     ///
     /// 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: SealedHeader, update: BlockUpdate<'_>) -> ProviderResult<()> {
-        let journal_hash = update.journal_hash();
+    fn ingest(&self, journal: HostJournal<'static>) -> ProviderResult<()> {
+        let journal_hash = journal.journal_hash();
+
+        let (meta, bsi) = journal.into_parts();
+        let (_, _, header) = meta.into_parts();
 
         // 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 bundle_state: BundleState = bsi.into();
         let execution_outcome = ExecutionOutcome::new(bundle_state, vec![], header.number());
 
         let block: SealedBlock<TransactionSigned, Header> =
-            SealedBlock { header, body: Default::default() };
+            SealedBlock { header: SealedHeader::new(header), body: Default::default() };
         let block_result =
             BlockResult { sealed_block: RecoveredBlock::new(block, vec![]), execution_outcome };