feat: chain orchestrator (#185)

* feat: chain orchestrator

* lint

* lint

* docs lint

* fix: clone the in-memory chain instead of taking it

* update chain consolidation and l1 message validation

* add missing network block error

* update fork sync / reconcilliation to use network as opossed to database

* fix merege

* refactor and add test cases

* cleanup

* rename chain orchestrator crate

* cleanup

* remove expect for sequencer in rnm

* sync test

* sync test

* sync test

* add error handling for missing paylaod id

* remove networking for L1 consolidation sync test

* improve test coverage and fix bugs

* lint

* fix cli NetworkArgs

* add block gap to reorg integration test

* make test more robust

* merge upstream

* refactor

* address comments

* address comments

* merge changes

* add migration script

Author: frisitano

.codespellrc

@@ -1,3 +1,3 @@
 [codespell]
-skip = .git,target,Cargo.toml,Cargo.lock
+skip = .git,target,Cargo.toml,Cargo.lock,docker-compose
 ignore-words-list = crate

.github/workflows/lint.yaml

@@ -128,11 +128,11 @@ jobs:
           - type: wasm
             target: wasm32-unknown-unknown
             exclude: |
-              scroll-engine,scroll-wire,rollup-node,scroll-network,rollup-node-manager,rollup-node-watcher,scroll-db,scroll-migration,rollup-node-indexer,scroll-codec,scroll-derivation-pipeline,rollup-node-providers,rollup-node-sequencer,rollup-node-signer,tests
+              scroll-engine,scroll-wire,rollup-node,scroll-network,rollup-node-manager,rollup-node-watcher,scroll-db,scroll-migration,rollup-node-chain-orchestrator,scroll-codec,scroll-derivation-pipeline,rollup-node-providers,rollup-node-sequencer,rollup-node-signer,tests
           - type: riscv
             target: riscv32imac-unknown-none-elf
             exclude: |
-              scroll-engine,scroll-wire,rollup-node,scroll-network,rollup-node-manager,rollup-node-watcher,scroll-db,scroll-migration,rollup-node-indexer,scroll-codec,scroll-derivation-pipeline,rollup-node-providers,rollup-node-sequencer,rollup-node-signer,tests
+              scroll-engine,scroll-wire,rollup-node,scroll-network,rollup-node-manager,rollup-node-watcher,scroll-db,scroll-migration,rollup-node-chain-orchestrator,scroll-codec,scroll-derivation-pipeline,rollup-node-providers,rollup-node-sequencer,rollup-node-signer,tests
     steps:
       - uses: actions/checkout@v5
       - uses: rui314/setup-mold@v1

Cargo.lock

@@ -12,7 +12,7 @@ members = [
     "crates/database/migration",
     "crates/derivation-pipeline",
     "crates/engine",
-    "crates/indexer",
+    "crates/chain-orchestrator",
     "crates/l1",
     "crates/manager",
     "crates/network",
@@ -155,6 +155,7 @@ reth-primitives-traits = { git = "https://github.com/scroll-tech/reth.git", defa
 reth-provider = { git = "https://github.com/scroll-tech/reth.git", default-features = false }
 reth-rpc-builder = { git = "https://github.com/scroll-tech/reth.git", default-features = false }
 reth-rpc-server-types = { git = "https://github.com/scroll-tech/reth.git", default-features = false }
+reth-storage-api = { git = "https://github.com/scroll-tech/reth.git", default-features = false }
 reth-tasks = { git = "https://github.com/scroll-tech/reth.git", default-features = false }
 reth-tokio-util = { git = "https://github.com/scroll-tech/reth.git", default-features = false }
 reth-tracing = { git = "https://github.com/scroll-tech/reth.git", default-features = false }
@@ -168,7 +169,7 @@ reth-scroll-primitives = { git = "https://github.com/scroll-tech/reth.git", defa
 
 # rollup node
 rollup-node = { path = "crates/node" }
-rollup-node-indexer = { path = "crates/indexer" }
+rollup-node-chain-orchestrator = { path = "crates/chain-orchestrator" }
 rollup-node-manager = { path = "crates/manager" }
 rollup-node-primitives = { path = "crates/primitives" }
 rollup-node-providers = { path = "crates/providers" }

Cargo.toml

@@ -24,7 +24,7 @@ This repository is a modular Rust workspace for the Scroll rollup node. It is de
 │   │   └── migration/
 │   ├── derivation-pipeline/
 │   ├── engine/
-│   ├── indexer/
+│   ├── chain-orchestrator/
 │   ├── l1/
 │   ├── network/
 │   ├── node/
@@ -46,7 +46,7 @@ This repository is a modular Rust workspace for the Scroll rollup node. It is de
 - **crates/database/migration/**: Database schema migrations using SeaORM.
 - **crates/derivation-pipeline/**: Stateless pipeline for transforming batches into block-building payloads.
 - **crates/engine/**: Core engine logic for block execution, fork choice, and payload management.
-- **crates/indexer/**: Indexes L1 and L2 data for efficient querying and notification.
+- **crates/chain-orchestrator/**: Responsible for orchestrating the L2 chain based on events from L1 and data gossiped over the P2P network.
 - **crates/l1/**: Primitives and ABI bindings for L1 contracts and messages.
 - **crates/network/**: P2P networking stack for node communication.
 - **crates/node/**: Node manager and orchestration logic.

README.md

@@ -1,5 +1,5 @@
 [package]
-name = "rollup-node-indexer"
+name = "rollup-node-chain-orchestrator"
 version.workspace = true
 edition.workspace = true
 rust-version.workspace = true
@@ -11,19 +11,30 @@ workspace = true
 
 [dependencies]
 # alloy
+alloy-consensus = { workspace = true }
+alloy-eips = { workspace = true }
+alloy-json-rpc.workspace = true
 alloy-primitives.workspace = true
+alloy-provider.workspace = true
+alloy-transport.workspace = true
 
 # rollup-node
 scroll-db.workspace = true
 rollup-node-primitives.workspace = true
 rollup-node-watcher.workspace = true
 
 # scroll
+reth-scroll-primitives.workspace = true
 scroll-alloy-consensus.workspace = true
 scroll-alloy-hardforks.workspace = true
+scroll-alloy-network.workspace = true
+scroll-network.workspace = true
 
 # reth
 reth-chainspec.workspace = true
+reth-network-p2p = { git = "https://github.com/scroll-tech/reth.git", default-features = false }
+reth-network-peers.workspace = true
+reth-primitives-traits.workspace = true
 
 # misc
 futures.workspace = true
@@ -36,7 +47,10 @@ tokio-stream.workspace = true
 tracing.workspace = true
 
 [dev-dependencies]
+alloy-consensus = { workspace = true, features = ["arbitrary"] }
 alloy-primitives = { workspace = true, features = ["arbitrary"] }
+alloy-rpc-client.workspace = true
+alloy-transport.workspace = true
 
 # rollup-node
 scroll-db = { workspace = true, features = ["test-utils"] }
@@ -46,8 +60,15 @@ rollup-node-primitives = { workspace = true, features = ["arbitrary"] }
 reth-scroll-chainspec.workspace = true
 reth-scroll-forks.workspace = true
 
+# reth
+reth-eth-wire-types.workspace = true
+reth-network-peers.workspace = true
+
 # misc
 arbitrary.workspace = true
 futures.workspace = true
+parking_lot.workspace = true
 rand.workspace = true
+reqwest.workspace = true
+serde_json = { version = "1.0" }
 tokio.workspace = true

crates/indexer/Cargo.toml renamed to crates/chain-orchestrator/Cargo.toml

@@ -0,0 +1,57 @@
+use super::{ChainOrchestratorError, ChainOrchestratorEvent};
+use std::{
+    fmt,
+    future::Future,
+    pin::Pin,
+    task::{Context, Poll},
+};
+
+/// A future that resolves to a `Result<ChainOrchestratorEvent, ChainOrchestratorError>`.
+pub(super) type PendingChainOrchestratorFuture = Pin<
+    Box<dyn Future<Output = Result<Option<ChainOrchestratorEvent>, ChainOrchestratorError>> + Send>,
+>;
+
+/// A type that represents a future that is being executed by the chain orchestrator.
+pub(super) enum ChainOrchestratorFuture {
+    HandleReorg(PendingChainOrchestratorFuture),
+    HandleFinalized(PendingChainOrchestratorFuture),
+    HandleBatchCommit(PendingChainOrchestratorFuture),
+    HandleBatchFinalization(PendingChainOrchestratorFuture),
+    HandleL1Message(PendingChainOrchestratorFuture),
+    HandleDerivedBlock(PendingChainOrchestratorFuture),
+    HandleL2Block(PendingChainOrchestratorFuture),
+}
+
+impl ChainOrchestratorFuture {
+    /// Polls the future to completion.
+    pub(super) fn poll(
+        &mut self,
+        cx: &mut Context<'_>,
+    ) -> Poll<Result<Option<ChainOrchestratorEvent>, ChainOrchestratorError>> {
+        match self {
+            Self::HandleReorg(fut) |
+            Self::HandleFinalized(fut) |
+            Self::HandleBatchCommit(fut) |
+            Self::HandleBatchFinalization(fut) |
+            Self::HandleL1Message(fut) |
+            Self::HandleDerivedBlock(fut) |
+            Self::HandleL2Block(fut) => fut.as_mut().poll(cx),
+        }
+    }
+}
+
+// We implement the Debug trait for ChainOrchestratorFuture to provide a human-readable
+// representation of the enum variants.
+impl fmt::Debug for ChainOrchestratorFuture {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            Self::HandleReorg(_) => write!(f, "HandleReorg"),
+            Self::HandleFinalized(_) => write!(f, "HandleFinalized"),
+            Self::HandleBatchCommit(_) => write!(f, "HandleBatchCommit"),
+            Self::HandleBatchFinalization(_) => write!(f, "HandleBatchFinalization"),
+            Self::HandleL1Message(_) => write!(f, "HandleL1Message"),
+            Self::HandleDerivedBlock(_) => write!(f, "HandleDerivedBlock"),
+            Self::HandleL2Block(_) => write!(f, "HandleL2Block"),
+        }
+    }
+}

crates/chain-orchestrator/src/action.rs

@@ -0,0 +1,47 @@
+use alloy_json_rpc::RpcError;
+use alloy_primitives::B256;
+use alloy_transport::TransportErrorKind;
+use scroll_db::{DatabaseError, L1MessageStart};
+
+/// A type that represents an error that occurred in the chain orchestrator.
+#[derive(Debug, thiserror::Error)]
+pub enum ChainOrchestratorError {
+    /// An error occurred while interacting with the database.
+    #[error("database error occurred: {0}")]
+    DatabaseError(#[from] DatabaseError),
+    /// An error occurred while trying to fetch the L2 block from the database.
+    #[error("L2 block not found - block number: {0}")]
+    L2BlockNotFoundInDatabase(u64),
+    /// An error occurred while trying to fetch the L2 block from the L2 client.
+    #[error("L2 block not found in L2 client - block number: {0}")]
+    L2BlockNotFoundInL2Client(u64),
+    /// A fork was received from the peer that is associated with a reorg of the safe chain.
+    #[error("L2 safe block reorg detected")]
+    L2SafeBlockReorgDetected,
+    /// A block contains invalid L1 messages.
+    #[error("Block contains invalid L1 message. Expected: {expected:?}, Actual: {actual:?}")]
+    L1MessageMismatch {
+        /// The expected L1 messages hash.
+        expected: B256,
+        /// The actual L1 messages hash.
+        actual: B256,
+    },
+    /// An L1 message was not found in the database.
+    #[error("L1 message not found at {0}")]
+    L1MessageNotFound(L1MessageStart),
+    /// An inconsistency was detected when trying to consolidate the chain.
+    #[error("Chain inconsistency detected")]
+    ChainInconsistency,
+    /// The peer did not provide the requested block header.
+    #[error("A peer did not provide the requested block header")]
+    MissingBlockHeader {
+        /// The hash of the block header that was requested.
+        hash: B256,
+    },
+    /// An error occurred while making a network request.
+    #[error("Network request error: {0}")]
+    NetworkRequestError(#[from] reth_network_p2p::error::RequestError),
+    /// An error occurred while making a JSON-RPC request to the Execution Node (EN).
+    #[error("An error occurred while making a JSON-RPC request to the EN: {0}")]
+    RpcError(#[from] RpcError<TransportErrorKind>),
+}

crates/chain-orchestrator/src/error.rs

@@ -0,0 +1,67 @@
+use alloy_consensus::Header;
+use alloy_primitives::{Signature, B256};
+use reth_network_peers::PeerId;
+use reth_scroll_primitives::ScrollBlock;
+use rollup_node_primitives::{BatchInfo, BlockInfo, ChainImport, L2BlockInfoWithL1Messages};
+
+/// An event emitted by the `ChainOrchestrator`.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum ChainOrchestratorEvent {
+    /// A new block has been received from the network but we have insufficient data to process it
+    /// due to being in optimistic mode.
+    InsufficientDataForReceivedBlock(B256),
+    /// The block that we have received is already known.
+    BlockAlreadyKnown(B256, PeerId),
+    /// A fork of the chain that is older than the current chain has been received.
+    OldForkReceived {
+        /// The headers of the old fork.
+        headers: Vec<Header>,
+        /// The peer that provided the old fork.
+        peer_id: PeerId,
+        /// The signature of the old fork.
+        signature: Signature,
+    },
+    /// The chain should be optimistically synced to the provided block.
+    OptimisticSync(ScrollBlock),
+    /// The chain has been extended, returning the new blocks.
+    ChainExtended(ChainImport),
+    /// The chain has reorged, returning the new chain and the peer that provided them.
+    ChainReorged(ChainImport),
+    /// A `BatchCommit` event has been indexed returning the batch info and L1 block number at
+    /// which the event was emitted. If this event is associated with a batch revert then the
+    /// `safe_head` will also be populated with the `BlockInfo` that represents the new L2 head.
+    BatchCommitIndexed {
+        /// The batch info.
+        batch_info: BatchInfo,
+        /// The L1 block number in which the batch was committed.
+        l1_block_number: u64,
+        /// The safe L2 block info.
+        safe_head: Option<BlockInfo>,
+    },
+    /// A batch has been finalized returning the batch hash and new an optional finalized
+    /// L2 block.
+    BatchFinalized(B256, Option<BlockInfo>),
+    /// An L1 block has been finalized returning the L1 block number and an optional
+    /// finalized L2 block.
+    L1BlockFinalized(u64, Option<BlockInfo>),
+    /// A `L1Message` event has been committed returning the message queue index.
+    L1MessageCommitted(u64),
+    /// A reorg has occurred on L1, returning the L1 block number of the new L1 head,
+    /// the L1 message queue index of the new L1 head, and optionally the L2 head and safe block
+    /// info if the reorg resulted in a new L2 head or safe block.
+    L1Reorg {
+        /// The L1 block number of the new L1 head.
+        l1_block_number: u64,
+        /// The L1 message queue index of the new L1 head.
+        queue_index: Option<u64>,
+        /// The L2 head block info.
+        l2_head_block_info: Option<BlockInfo>,
+        /// The L2 safe block info.
+        l2_safe_block_info: Option<BlockInfo>,
+    },
+    /// An L2 block has been committed returning the [`L2BlockInfoWithL1Messages`] and an
+    /// optional [`BatchInfo`] if the block is associated with a committed batch.
+    L2ChainCommitted(L2BlockInfoWithL1Messages, Option<BatchInfo>, bool),
+    /// An L2 consolidated block has been committed returning the [`L2BlockInfoWithL1Messages`].
+    L2ConsolidatedBlockCommitted(L2BlockInfoWithL1Messages),
+}