feat: integrate derivation pipeline

Cargo.toml

@@ -167,6 +167,7 @@ scroll-migration = { path = "crates/database/migration" }
 # misc
 arbitrary = { version = "1.4", default-features = false }
 async-trait = "0.1"
+auto_impl = "1.2"
 derive_more = { version = "2.0", default-features = false }
 eyre = "0.6"
 futures = { version = "0.3", default-features = false }

bin/rollup/Cargo.toml

@@ -39,15 +39,16 @@ reth-scroll-cli = { git = "https://github.com/scroll-tech/reth.git" }
 reth-scroll-engine-primitives = { git = "https://github.com/scroll-tech/reth.git" }
 reth-scroll-node = { workspace = true, features = ["skip-state-root-validation"] }
 reth-scroll-primitives = { workspace = true }
-scroll-wire.workspace = true
-scroll-network.workspace = true
-scroll-engine = { workspace = true, features = ["test-utils"] }
 scroll-db = { workspace = true }
+scroll-engine = { workspace = true, features = ["test-utils"] }
 scroll-migration.workspace = true
+scroll-network.workspace = true
+scroll-wire.workspace = true
 
 # rollup-node
-rollup-node-manager.workspace = true
 rollup-node-indexer.workspace = true
+rollup-node-manager.workspace = true
+rollup-node-providers.workspace = true
 rollup-node-watcher.workspace = true
 
 # misc

bin/rollup/src/args.rs

@@ -12,6 +12,8 @@ pub struct ScrollRollupNodeArgs {
     pub database_path: Option<PathBuf>,
     /// The URL for the L1 RPC URL.
     pub l1_rpc_url: Option<reqwest::Url>,
+    /// The URL for the Beacon RPC URL.
+    pub beacon_rpc_url: reqwest::Url,
     /// The EngineAPI URL.
     pub engine_api_url: Option<reqwest::Url>,
 }

bin/rollup/src/network.rs

@@ -10,6 +10,7 @@ use reth_scroll_primitives::ScrollPrimitives;
 use reth_transaction_pool::{PoolTransaction, TransactionPool};
 use rollup_node_indexer::Indexer;
 use rollup_node_manager::{PoAConsensus, RollupNodeManager};
+use rollup_node_providers::{DatabaseL1MessageProvider, OnlineBeaconClient, OnlineL1Provider};
 use rollup_node_watcher::L1Watcher;
 use scroll_alloy_provider::ScrollAuthEngineApiProvider;
 use scroll_db::{Database, DatabaseConnectionProvider};
@@ -121,10 +122,15 @@ where
             None
         };
 
+        let beacon_client = OnlineBeaconClient::new_http(self.config.beacon_rpc_url.to_string());
+        let l1_messages_provider = DatabaseL1MessageProvider::new(db.clone());
+        let l1_provider = OnlineL1Provider::new(beacon_client, 100, l1_messages_provider).await;
+
         // Spawn the rollup node manager
         let rollup_node_manager = RollupNodeManager::new(
             scroll_network_manager,
             engine,
+            l1_provider,
             l1_notification_rx,
             indexer,
             ForkchoiceState::genesis(

bin/rollup/tests/e2e.rs

@@ -137,6 +137,8 @@ pub async fn build_bridge_node(
         enable_scroll_wire: true,
         database_path: Some(PathBuf::from("sqlite::memory:")),
         l1_rpc_url: None,
+        // <https://docs.arbitrum.io/run-arbitrum-node/l1-ethereum-beacon-chain-rpc-providers>
+        beacon_rpc_url: reqwest::Url::parse("https://eth-beacon-chain.drpc.org/rest/")?,
         engine_api_url: None,
     };
     let node = ScrollNode;

crates/database/db/Cargo.toml

@@ -22,10 +22,9 @@ rollup-node-primitives.workspace = true
 
 # misc
 async-trait.workspace = true
+auto_impl.workspace = true
 futures.workspace = true
 sea-orm = { version = "1.1.0", features = ["sqlx-sqlite", "runtime-tokio-native-tls", "macros"] }
-serde = { version = "1.0", features = ["derive"] }
-serde_json = "1.0"
 thiserror.workspace = true
 tokio = { workspace = true, features = ["macros", "sync"] }
 tracing.workspace = true

crates/database/db/src/connection.rs

@@ -1,5 +1,6 @@
 /// The [`DatabaseConnectionProvider`] trait provides a way to get a connection to the database.
 /// This is implemented by the [`crate::Database`] and [`crate::DatabaseTransaction`] types.
+#[auto_impl::auto_impl(Arc)]
 pub trait DatabaseConnectionProvider {
     /// The type of the database connection.
     type Connection: sea_orm::ConnectionTrait + sea_orm::StreamTrait;

crates/database/db/src/db.rs

@@ -45,12 +45,16 @@ impl From<DatabaseConnection> for Database {
 
 #[cfg(test)]
 mod test {
-    use crate::{operations::DatabaseOperations, test_utils::setup_test_db};
+    use crate::{
+        models, operations::DatabaseOperations, test_utils::setup_test_db,
+        DatabaseConnectionProvider,
+    };
 
     use arbitrary::{Arbitrary, Unstructured};
     use futures::StreamExt;
     use rand::Rng;
     use rollup_node_primitives::{BatchCommitData, L1MessageWithBlockNumber};
+    use sea_orm::{ColumnTrait, EntityTrait, QueryFilter};
 
     #[tokio::test]
     async fn test_database_round_trip_batch_commit() {
@@ -62,16 +66,46 @@ mod test {
         rand::rng().fill(bytes.as_mut_slice());
         let mut u = Unstructured::new(&bytes);
 
-        // Generate a random BatchInputV1.
+        // Generate a random BatchCommitData.
         let batch_commit = BatchCommitData::arbitrary(&mut u).unwrap();
 
-        // Round trip the BatchInput through the database.
+        // Round trip the BatchCommitData through the database.
         db.insert_batch(batch_commit.clone()).await.unwrap();
         let batch_commit_from_db =
             db.get_batch_by_index(batch_commit.index).await.unwrap().unwrap();
         assert_eq!(batch_commit, batch_commit_from_db);
     }
 
+    #[tokio::test]
+    async fn test_database_finalize_batch_commit() {
+        // Set up the test database.
+        let db = setup_test_db().await;
+
+        // Generate unstructured bytes.
+        let mut bytes = [0u8; 1024];
+        rand::rng().fill(bytes.as_mut_slice());
+        let mut u = Unstructured::new(&bytes);
+
+        // Generate a random BatchCommitData.
+        let batch_commit = BatchCommitData::arbitrary(&mut u).unwrap();
+
+        // Store the batch and finalize it.
+        let finalized_block_number = u64::arbitrary(&mut u).unwrap();
+        db.insert_batch(batch_commit.clone()).await.unwrap();
+        db.finalize_batch(batch_commit.hash, finalized_block_number).await.unwrap();
+
+        // Verify the finalized_block_number is correctly updated.
+        let finalized_block_number_from_db = models::batch_commit::Entity::find()
+            .filter(models::batch_commit::Column::Hash.eq(batch_commit.hash.to_vec()))
+            .one(db.get_connection())
+            .await
+            .unwrap()
+            .unwrap()
+            .finalized_block_number
+            .unwrap();
+        assert_eq!(finalized_block_number, finalized_block_number_from_db as u64);
+    }
+
     #[tokio::test]
     async fn test_database_round_trip_l1_message() {
         // Set up the test database.

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

@@ -14,7 +14,7 @@ pub struct Model {
     block_timestamp: i64,
     calldata: Vec<u8>,
     blob_hash: Option<Vec<u8>>,
-    finalized_block_number: Option<i64>,
+    pub(crate) finalized_block_number: Option<i64>,
 }
 
 /// The relation for the batch input model.

crates/database/db/src/operations.rs

@@ -32,7 +32,7 @@ pub trait DatabaseOperations: DatabaseConnectionProvider {
             .one(self.get_connection())
             .await?
         {
-            tracing::trace!(target: "scroll::db", batch_hash = ?batch_hash, block_number, "Finalizing batch input in database.");
+            tracing::trace!(target: "scroll::db", batch_hash = ?batch_hash, block_number, "Finalizing batch commit in database.");
             let mut batch: models::batch_commit::ActiveModel = batch.into();
             batch.finalized_block_number = Set(Some(block_number as i64));
             batch.update(self.get_connection()).await?;

crates/database/migration/src/m20220101_000001_create_batch_commit_table.rs

@@ -20,7 +20,7 @@ impl MigrationTrait for Migration {
                     .col(big_unsigned(BatchCommit::BlockTimestamp))
                     .col(binary(BatchCommit::Calldata))
                     .col(binary_len_null(BatchCommit::BlobHash, HASH_LENGTH))
-                    .col(boolean_null(BatchCommit::FinalizedBlockNumber))
+                    .col(big_unsigned_null(BatchCommit::FinalizedBlockNumber))
                     .to_owned(),
             )
             .await

crates/derivation-pipeline/Cargo.toml

@@ -24,13 +24,17 @@ scroll-alloy-rpc-types-engine.workspace = true
 rollup-node-primitives.workspace = true
 rollup-node-providers.workspace = true
 scroll-codec.workspace = true
+scroll-db.workspace = true
 
 # misc
+futures.workspace = true
 thiserror.workspace = true
+tracing.workspace = true
 
 [dev-dependencies]
 async-trait.workspace = true
 eyre.workspace = true
+scroll-db = { workspace = true, features = ["test-utils"] }
 scroll-codec = { workspace = true, features = ["test-utils"] }
 tokio = { workspace = true, features = ["macros"] }
 
@@ -45,4 +49,5 @@ std = [
     "reth-scroll-chainspec/std",
     "scroll-alloy-consensus/std",
     "scroll-alloy-rpc-types-engine/std",
+    "futures/std",
 ]

crates/derivation-pipeline/src/error.rs

@@ -1,18 +1,25 @@
 use rollup_node_providers::L1ProviderError;
 use scroll_codec::CodecError;
+use scroll_db::DatabaseError;
 
 /// An error occurred during the derivation process.
 #[derive(Debug, thiserror::Error)]
 pub enum DerivationPipelineError {
-    /// An error in the codec.
-    #[error(transparent)]
-    Codec(#[from] CodecError),
     /// Missing L1 messages cursor.
     #[error("missing l1 message queue cursor")]
     MissingL1MessageQueueCursor,
     /// Missing L1 message.
     #[error("missing l1 message")]
     MissingL1Message,
+    /// Unknown batch.
+    #[error("unknown batch for index {0}")]
+    UnknownBatch(u64),
+    /// An error in the codec.
+    #[error(transparent)]
+    Codec(#[from] CodecError),
+    /// An error in the database.
+    #[error(transparent)]
+    Database(#[from] DatabaseError),
     /// An error at the L1 provider.
     #[error(transparent)]
     L1Provider(#[from] L1ProviderError),