feat: message queue hash (#59)

* feat: add message queue hash to db schema.

* feat: update L1MessageWithBlockNumber to L1MessageEnvelope

* feat: fetch block timestamp block for l1 message in watcher

* feat: add chainspec to rollup node

* feat: compute rolling hash in indexer

* fix: rename batch to block to derived block

* fix: reorder args

* fix: lints

* fix: missing ts

* fix: set queue hash column as null

* feat: implement hash cursor for all `L1MessageProvider` instances

* fix: merge main

Author: greged93

Cargo.lock

@@ -131,6 +131,7 @@ alloy-transport = { version = "0.14.0", default-features = false }
 
 # scroll-alloy
 scroll-alloy-consensus = { git = "https://github.com/scroll-tech/reth.git", default-features = false }
+scroll-alloy-hardforks = { git = "https://github.com/scroll-tech/reth.git", default-features = false }
 scroll-alloy-network = { git = "https://github.com/scroll-tech/reth.git", default-features = false }
 scroll-alloy-provider = { git = "https://github.com/scroll-tech/reth.git", default-features = false }
 scroll-alloy-rpc-types-engine = { git = "https://github.com/scroll-tech/reth.git", default-features = false }
@@ -149,6 +150,7 @@ reth-tokio-util = { git = "https://github.com/scroll-tech/reth.git", default-fea
 # reth-scroll
 reth-scroll-chainspec = { git = "https://github.com/scroll-tech/reth.git", default-features = false }
 reth-scroll-engine-primitives = { git = "https://github.com/scroll-tech/reth.git", default-features = false }
+reth-scroll-forks = { git = "https://github.com/scroll-tech/reth.git", default-features = false }
 reth-scroll-node = { git = "https://github.com/scroll-tech/reth.git", default-features = false }
 reth-scroll-primitives = { git = "https://github.com/scroll-tech/reth.git", default-features = false }
 

Cargo.toml

@@ -133,6 +133,9 @@ where
         // Wrap the database in an Arc
         let db = Arc::new(db);
 
+        // Get the chain specification
+        let chain_spec = ctx.chain_spec();
+
         // Spawn the L1Watcher
         let l1_provider_args = self.config.l1_provider_args;
         let l1_notification_rx = if let Some(l1_rpc_url) = l1_provider_args.l1_rpc_url {
@@ -183,6 +186,7 @@ where
             db,
             l1_notification_rx,
             consensus,
+            chain_spec,
             block_rx,
             sequencer,
             block_time,

bin/rollup/src/network.rs

@@ -55,7 +55,7 @@ mod test {
     use arbitrary::{Arbitrary, Unstructured};
     use futures::StreamExt;
     use rand::Rng;
-    use rollup_node_primitives::{BatchCommitData, BatchInfo, BlockInfo, L1MessageWithBlockNumber};
+    use rollup_node_primitives::{BatchCommitData, BatchInfo, BlockInfo, L1MessageEnvelope};
     use sea_orm::{ColumnTrait, EntityTrait, QueryFilter};
 
     #[tokio::test]
@@ -119,13 +119,16 @@ mod test {
         let mut u = Unstructured::new(&bytes);
 
         // Generate a random L1Message.
-        let l1_message = L1MessageWithBlockNumber::arbitrary(&mut u).unwrap();
+        let l1_message = L1MessageEnvelope::arbitrary(&mut u).unwrap();
 
         // Round trip the L1Message through the database.
         db.insert_l1_message(l1_message.clone()).await.unwrap();
-        let l1_message_from_db =
-            db.get_l1_message(l1_message.transaction.queue_index).await.unwrap().unwrap();
-        assert_eq!(l1_message, l1_message_from_db);
+        let l1_message_from_db_index =
+            db.get_l1_message_by_index(l1_message.transaction.queue_index).await.unwrap().unwrap();
+        let l1_message_from_db_hash =
+            db.get_l1_message_by_hash(l1_message.queue_hash.unwrap()).await.unwrap().unwrap();
+        assert_eq!(l1_message, l1_message_from_db_index);
+        assert_eq!(l1_message, l1_message_from_db_hash);
     }
 
     #[tokio::test]
@@ -139,7 +142,7 @@ mod test {
     }
 
     #[tokio::test]
-    async fn test_database_batch_to_block_exists() {
+    async fn test_derived_block_exists() {
         // Set up the test database.
         let db = setup_test_db().await;
 
@@ -157,7 +160,7 @@ mod test {
         for _ in 0..10 {
             let block_info =
                 BlockInfo { number: block_number, hash: B256::arbitrary(&mut u).unwrap() };
-            db.insert_batch_to_block(batch_info, block_info).await.unwrap();
+            db.insert_derived_block(block_info, batch_info).await.unwrap();
             block_number += 1;
         }
 
@@ -168,7 +171,7 @@ mod test {
     }
 
     #[tokio::test]
-    async fn test_database_batch_to_block_missing() {
+    async fn test_derived_block_missing() {
         // Set up the test database.
         let db = setup_test_db().await;
 
@@ -191,7 +194,7 @@ mod test {
         for _ in 0..10 {
             let block_info =
                 BlockInfo { number: block_number, hash: B256::arbitrary(&mut u).unwrap() };
-            db.insert_batch_to_block(first_batch_info, block_info).await.unwrap();
+            db.insert_derived_block(block_info, first_batch_info).await.unwrap();
             block_number += 1;
         }
 
@@ -256,8 +259,8 @@ mod test {
         let mut u = Unstructured::new(&bytes);
 
         // Generate 2 random L1Messages.
-        let l1_message_1 = L1MessageWithBlockNumber::arbitrary(&mut u).unwrap();
-        let l1_message_2 = L1MessageWithBlockNumber::arbitrary(&mut u).unwrap();
+        let l1_message_1 = L1MessageEnvelope::arbitrary(&mut u).unwrap();
+        let l1_message_2 = L1MessageEnvelope::arbitrary(&mut u).unwrap();
 
         // Insert the L1Messages into the database in a transaction.
         let tx = db.tx().await.unwrap();
@@ -266,11 +269,17 @@ mod test {
         tx.commit().await.unwrap();
 
         // Check that the L1Messages are in the database.
-        let l1_message_1_from_db =
-            db.get_l1_message(l1_message_1.transaction.queue_index).await.unwrap().unwrap();
+        let l1_message_1_from_db = db
+            .get_l1_message_by_index(l1_message_1.transaction.queue_index)
+            .await
+            .unwrap()
+            .unwrap();
         assert_eq!(l1_message_1, l1_message_1_from_db);
-        let l1_message_2_from_db =
-            db.get_l1_message(l1_message_2.transaction.queue_index).await.unwrap().unwrap();
+        let l1_message_2_from_db = db
+            .get_l1_message_by_index(l1_message_2.transaction.queue_index)
+            .await
+            .unwrap()
+            .unwrap();
         assert_eq!(l1_message_2, l1_message_2_from_db);
     }
 
@@ -285,8 +294,8 @@ mod test {
         let mut u = Unstructured::new(&bytes);
 
         // Generate 2 random L1Messages.
-        let l1_message_1 = L1MessageWithBlockNumber::arbitrary(&mut u).unwrap();
-        let l1_message_2 = L1MessageWithBlockNumber::arbitrary(&mut u).unwrap();
+        let l1_message_1 = L1MessageEnvelope::arbitrary(&mut u).unwrap();
+        let l1_message_2 = L1MessageEnvelope::arbitrary(&mut u).unwrap();
 
         // Insert the L1Messages into the database.
         db.insert_l1_message(l1_message_1.clone()).await.unwrap();

crates/database/db/src/db.rs

@@ -13,4 +13,7 @@ pub enum DatabaseError {
     /// The block was not found in database.
     #[error("no block for id {0}")]
     BlockNotFound(BlockId),
+    /// The L1 message was not found in database.
+    #[error("L1 message at index [{0}] not found in database")]
+    L1MessageNotFound(u64),
 }

crates/database/db/src/error.rs

@@ -22,12 +22,12 @@ pub struct Model {
 pub enum Relation {
     /// A one-to-many relation with the derived block table.
     #[sea_orm(has_many = "super::derived_block::Entity")]
-    BatchToBlock,
+    DerivedBlock,
 }
 
 impl Related<super::derived_block::Entity> for Entity {
     fn to() -> RelationDef {
-        Relation::BatchToBlock.def()
+        Relation::DerivedBlock.def()
     }
 }
 

crates/database/db/src/models/batch_commit.rs

@@ -42,17 +42,17 @@ impl Related<super::batch_commit::Entity> for Entity {
 /// The active model behavior for the batch input model.
 impl ActiveModelBehavior for ActiveModel {}
 
-impl From<(BatchInfo, BlockInfo)> for ActiveModel {
-    fn from((batch_info, block_info): (BatchInfo, BlockInfo)) -> Self {
+impl From<(BlockInfo, BatchInfo)> for ActiveModel {
+    fn from((block_info, batch_info): (BlockInfo, BatchInfo)) -> Self {
         Self {
-            batch_index: ActiveValue::Set(
-                batch_info.index.try_into().expect("index should fit in i64"),
-            ),
-            batch_hash: ActiveValue::Set(batch_info.hash.to_vec()),
             block_number: ActiveValue::Set(
                 block_info.number.try_into().expect("block number should fit in i64"),
             ),
             block_hash: ActiveValue::Set(block_info.hash.to_vec()),
+            batch_index: ActiveValue::Set(
+                batch_info.index.try_into().expect("index should fit in i64"),
+            ),
+            batch_hash: ActiveValue::Set(batch_info.hash.to_vec()),
         }
     }
 }

crates/database/db/src/models/derived_block.rs

@@ -1,5 +1,5 @@
-use alloy_primitives::{Address, U256};
-use rollup_node_primitives::L1MessageWithBlockNumber;
+use alloy_primitives::{Address, B256, U256};
+use rollup_node_primitives::L1MessageEnvelope;
 use scroll_alloy_consensus::TxL1Message;
 use sea_orm::{entity::prelude::*, ActiveValue};
 
@@ -9,6 +9,7 @@ use sea_orm::{entity::prelude::*, ActiveValue};
 pub struct Model {
     #[sea_orm(primary_key)]
     queue_index: i64,
+    queue_hash: Option<Vec<u8>>,
     block_number: i64,
     gas_limit: String,
     to: Vec<u8>,
@@ -24,10 +25,11 @@ pub enum Relation {}
 /// The active model behavior for the L1 message model.
 impl ActiveModelBehavior for ActiveModel {}
 
-impl From<L1MessageWithBlockNumber> for ActiveModel {
-    fn from(value: L1MessageWithBlockNumber) -> Self {
+impl From<L1MessageEnvelope> for ActiveModel {
+    fn from(value: L1MessageEnvelope) -> Self {
         Self {
             queue_index: ActiveValue::Set(value.transaction.queue_index as i64),
+            queue_hash: ActiveValue::Set(value.queue_hash.map(|q| q.to_vec())),
             block_number: ActiveValue::Set(value.block_number as i64),
             gas_limit: ActiveValue::Set(value.transaction.gas_limit.to_string()),
             to: ActiveValue::Set(value.transaction.to.to_vec()),
@@ -38,10 +40,11 @@ impl From<L1MessageWithBlockNumber> for ActiveModel {
     }
 }
 
-impl From<Model> for L1MessageWithBlockNumber {
+impl From<Model> for L1MessageEnvelope {
     fn from(value: Model) -> Self {
         Self {
             block_number: value.block_number as u64,
+            queue_hash: value.queue_hash.map(|q| B256::from_slice(&q)),
             transaction: TxL1Message {
                 queue_index: value.queue_index as u64,
                 gas_limit: value.gas_limit.parse().expect("gas limit is valid"),

crates/database/db/src/models/l1_message.rs

@@ -4,7 +4,7 @@ use crate::DatabaseConnectionProvider;
 use alloy_eips::{BlockId, BlockNumberOrTag};
 use alloy_primitives::B256;
 use futures::{Stream, StreamExt};
-use rollup_node_primitives::{BatchCommitData, BatchInfo, BlockInfo, L1MessageWithBlockNumber};
+use rollup_node_primitives::{BatchCommitData, BatchInfo, BlockInfo, L1MessageEnvelope};
 use scroll_alloy_rpc_types_engine::BlockDataHint;
 use sea_orm::{
     ActiveModelTrait, ColumnTrait, Condition, DbErr, EntityTrait, QueryFilter, QueryOrder,
@@ -107,18 +107,15 @@ pub trait DatabaseOperations: DatabaseConnectionProvider {
             .map(|res| res.map(Into::into)))
     }
 
-    /// Insert an [`L1MessageWithBlockNumber`] into the database.
-    async fn insert_l1_message(
-        &self,
-        l1_message: L1MessageWithBlockNumber,
-    ) -> Result<(), DatabaseError> {
+    /// Insert an [`L1MessageEnvelope`] into the database.
+    async fn insert_l1_message(&self, l1_message: L1MessageEnvelope) -> Result<(), DatabaseError> {
         tracing::trace!(target: "scroll::db", queue_index = l1_message.transaction.queue_index, "Inserting L1 message into database.");
         let l1_message: models::l1_message::ActiveModel = l1_message.into();
         l1_message.insert(self.get_connection()).await?;
         Ok(())
     }
 
-    /// Delete all [`L1MessageWithBlockNumber`]s with a block number greater than the provided block
+    /// Delete all [`L1MessageEnvelope`]s with a block number greater than the provided block
     /// number.
     async fn delete_l1_messages_gt(&self, block_number: u64) -> Result<(), DatabaseError> {
         tracing::trace!(target: "scroll::db", block_number, "Deleting L1 messages greater than block number.");
@@ -129,22 +126,37 @@ pub trait DatabaseOperations: DatabaseConnectionProvider {
             .map(|_| ())?)
     }
 
-    /// Get a [`L1MessageWithBlockNumber`] from the database by its message queue index.
-    async fn get_l1_message(
+    /// Get a [`L1MessageEnvelope`] from the database by its message queue index.
+    async fn get_l1_message_by_index(
         &self,
         queue_index: u64,
-    ) -> Result<Option<L1MessageWithBlockNumber>, DatabaseError> {
+    ) -> Result<Option<L1MessageEnvelope>, DatabaseError> {
         Ok(models::l1_message::Entity::find_by_id(queue_index as i64)
             .one(self.get_connection())
             .await
             .map(|x| x.map(Into::into))?)
     }
 
-    /// Gets an iterator over all [`L1MessageWithBlockNumber`]s in the database.
+    /// Get a [`L1MessageEnvelope`] from the database by its message queue hash.
+    async fn get_l1_message_by_hash(
+        &self,
+        queue_hash: B256,
+    ) -> Result<Option<L1MessageEnvelope>, DatabaseError> {
+        Ok(models::l1_message::Entity::find()
+            .filter(
+                Condition::all()
+                    .add(models::l1_message::Column::QueueHash.is_not_null())
+                    .add(models::l1_message::Column::QueueHash.eq(queue_hash.to_vec())),
+            )
+            .one(self.get_connection())
+            .await
+            .map(|x| x.map(Into::into))?)
+    }
+
+    /// Gets an iterator over all [`L1MessageEnvelope`]s in the database.
     async fn get_l1_messages<'a>(
         &'a self,
-    ) -> Result<impl Stream<Item = Result<L1MessageWithBlockNumber, DbErr>> + 'a, DatabaseError>
-    {
+    ) -> Result<impl Stream<Item = Result<L1MessageEnvelope, DbErr>> + 'a, DatabaseError> {
         Ok(models::l1_message::Entity::find()
             .stream(self.get_connection())
             .await?
@@ -170,21 +182,21 @@ pub trait DatabaseOperations: DatabaseConnectionProvider {
             .map(|x| x.map(Into::into))?)
     }
 
-    /// Insert a new batch to block line in the database.
-    async fn insert_batch_to_block(
+    /// Insert a new derived block line in the database.
+    async fn insert_derived_block(
         &self,
-        batch_info: BatchInfo,
         block_info: BlockInfo,
+        batch_info: BatchInfo,
     ) -> Result<(), DatabaseError> {
         tracing::trace!(
             target: "scroll::db",
             batch_hash = ?batch_info.hash,
             batch_index = batch_info.index,
             block_number = block_info.number,
             block_hash = ?block_info.hash,
-            "Inserting batch to block into database."
+            "Inserting derived block into database."
         );
-        let derived_block: models::derived_block::ActiveModel = (batch_info, block_info).into();
+        let derived_block: models::derived_block::ActiveModel = (block_info, batch_info).into();
         derived_block.insert(self.get_connection()).await?;
 
         Ok(())

crates/database/db/src/operations.rs

@@ -12,6 +12,7 @@ impl MigrationTrait for Migration {
                     .table(L1Message::Table)
                     .if_not_exists()
                     .col(pk_auto(L1Message::QueueIndex))
+                    .col(binary_len_null(L1Message::QueueHash, 32))
                     .col(unsigned(L1Message::BlockNumber))
                     .col(text(L1Message::GasLimit))
                     .col(binary_len(L1Message::To, 20))
@@ -32,6 +33,7 @@ impl MigrationTrait for Migration {
 enum L1Message {
     Table,
     QueueIndex,
+    QueueHash,
     BlockNumber,
     GasLimit,
     To,