feat(sequencer): add Executor actor

Cargo.lock

@@ -2326,7 +2326,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "ccc2776f0c61eca1ca32528f85548abd1a4be8fb53d1b21c013e4f18da1e7090"
 dependencies = [
  "data-encoding",
- "syn 1.0.109",
+ "syn 2.0.117",
 ]
 
 [[package]]
@@ -2572,6 +2572,12 @@ version = "1.2.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "75b325c5dbd37f80359721ad39aca5a29fb04c89279657cffdda8736d0c0b9d2"
 
+[[package]]
+name = "downcast-rs"
+version = "2.0.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "117240f60069e65410b3ae1bb213295bd828f707b5bec6596a1afc8793ce0cbc"
+
 [[package]]
 name = "downloader"
 version = "0.2.8"
@@ -4816,6 +4822,33 @@ dependencies = [
  "wnaf",
 ]
 
+[[package]]
+name = "kameo"
+version = "0.22.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7cbab7323ed30490812f43ef6416a9e21bb150e9633e451ed124ca276b1ca82a"
+dependencies = [
+ "downcast-rs 2.0.2",
+ "dyn-clone",
+ "futures",
+ "kameo_macros",
+ "serde",
+ "tokio",
+ "tracing",
+]
+
+[[package]]
+name = "kameo_macros"
+version = "0.21.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f7566055976eb86ee8e8fbafa0fbdad985c5d7c3f4eed04fc11bb495f71e3856"
+dependencies = [
+ "heck",
+ "proc-macro2",
+ "quote",
+ "syn 2.0.117",
+]
+
 [[package]]
 name = "keccak"
 version = "0.1.6"
@@ -9037,7 +9070,7 @@ version = "0.1.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "b4382d3af3a4ebdae7f64ba6edd9114fff92c89808004c4943b393377a25d001"
 dependencies = [
- "downcast-rs",
+ "downcast-rs 1.2.1",
  "paste",
 ]
 
@@ -9531,6 +9564,18 @@ dependencies = [
  "strum 0.28.0",
 ]
 
+[[package]]
+name = "sequencer_executor_actor"
+version = "0.1.0"
+dependencies = [
+ "anyhow",
+ "common",
+ "kameo",
+ "lee_core",
+ "mempool",
+ "sequencer_core",
+]
+
 [[package]]
 name = "sequencer_service"
 version = "0.1.0"
@@ -10709,6 +10754,7 @@ dependencies = [
  "signal-hook-registry",
  "socket2 0.6.4",
  "tokio-macros",
+ "tracing",
  "windows-sys 0.61.2",
 ]
 

Cargo.toml

@@ -23,6 +23,7 @@ members = [
   "lez/sequencer/service/protocol",
   "lez/sequencer/service/rpc",
   "lez/sequencer/service/metrics",
+  "lez/sequencer/actors/executor",
   "lez/indexer/core",
   "lez/indexer/service",
   "lez/indexer/service/protocol",
@@ -85,6 +86,7 @@ sequencer_core_metrics = { path = "lez/sequencer/core/metrics" }
 sequencer_service_protocol = { path = "lez/sequencer/service/protocol" }
 sequencer_service_rpc = { path = "lez/sequencer/service/rpc" }
 sequencer_service_metrics = { path = "lez/sequencer/service/metrics" }
+sequencer_executor_actor = { path = "lez/sequencer/actors/executor" }
 sequencer_service = { path = "lez/sequencer/service" }
 indexer_core = { path = "lez/indexer/core" }
 indexer_service = { path = "lez/indexer/service" }
@@ -130,6 +132,7 @@ tokio = { version = "1.50", features = [
 tokio-util = "0.7.18"
 risc0-zkvm = { version = "3.0.5", default-features = false, features = ['std'] }
 risc0-build = "3.0.5"
+kameo = "0.22.2"
 anyhow = "1.0.98"
 derive_more = "2.1.1"
 num_cpus = "1.13.1"

lez/sequencer/actors/executor/Cargo.toml

@@ -0,0 +1,17 @@
+[package]
+name = "sequencer_executor_actor"
+version = "0.1.0"
+edition = "2024"
+license = { workspace = true }
+
+[lints]
+workspace = true
+
+[dependencies]
+sequencer_core.workspace = true
+common.workspace = true
+lee_core.workspace = true
+mempool.workspace = true
+
+kameo.workspace = true
+anyhow.workspace = true

lez/sequencer/actors/executor/src/lib.rs

@@ -0,0 +1,197 @@
+//! Executor Actor performs the main logic of the Sequencer.
+
+use anyhow::Result;
+use common::{block::Block, transaction::LeeTransaction};
+use kameo::{Actor, message::Message};
+use lee_core::{
+    BlockId,
+    account::{Balance, Nonce},
+};
+use mempool::MemPoolHandle;
+use sequencer_core::{
+    SequencerCore, TransactionOrigin,
+    block_publisher::{BlockPublisherTrait as _, ZoneSdkPublisher},
+    config::SequencerConfig,
+};
+
+use crate::protocol::{
+    GetAccount, GetAccountBalance, GetAccountNonces, GetAccountReply, GetBlock, GetBlockRange,
+    GetChannelId, GetChannelIdReply, GetLastBlockId, GetProofsAndRoot, GetTransaction, Transaction,
+};
+
+pub mod protocol;
+
+#[derive(Actor)]
+pub struct ExecutorActor {
+    sequencer: SequencerCore<ZoneSdkPublisher>,
+    mempool_handle: MemPoolHandle<(TransactionOrigin, LeeTransaction)>,
+}
+
+impl ExecutorActor {
+    pub async fn new(config: SequencerConfig) -> Self {
+        let (sequencer, mempool_handle): (SequencerCore, _) =
+            SequencerCore::start_from_config(config).await;
+
+        Self {
+            sequencer,
+            mempool_handle,
+        }
+    }
+}
+
+impl Message<Transaction> for ExecutorActor {
+    type Reply = ();
+
+    async fn handle(
+        &mut self,
+        Transaction { transaction }: Transaction,
+        _ctx: &mut kameo::prelude::Context<Self, Self::Reply>,
+    ) -> Self::Reply {
+        self.mempool_handle
+            .push((TransactionOrigin::User, transaction))
+            .await
+            .expect("Mempool is closed, this is a bug");
+    }
+}
+
+impl Message<GetBlock> for ExecutorActor {
+    type Reply = Result<Option<Block>>;
+
+    async fn handle(
+        &mut self,
+        GetBlock { block_id }: GetBlock,
+        _ctx: &mut kameo::prelude::Context<Self, Self::Reply>,
+    ) -> Self::Reply {
+        self.sequencer
+            .block_store()
+            .get_block_at_id(block_id)
+            .map_err(Into::into)
+    }
+}
+
+impl Message<GetBlockRange> for ExecutorActor {
+    type Reply = Result<Vec<Block>>;
+
+    async fn handle(
+        &mut self,
+        GetBlockRange { range }: GetBlockRange,
+        _ctx: &mut kameo::prelude::Context<Self, Self::Reply>,
+    ) -> Self::Reply {
+        range
+            .map_while(|block_id| {
+                self.sequencer
+                    .block_store()
+                    .get_block_at_id(block_id)
+                    .map_err(Into::into)
+                    .transpose()
+            })
+            .collect::<Result<Vec<_>, _>>()
+    }
+}
+
+impl Message<GetLastBlockId> for ExecutorActor {
+    type Reply = Result<BlockId>;
+
+    async fn handle(
+        &mut self,
+        GetLastBlockId: GetLastBlockId,
+        _ctx: &mut kameo::prelude::Context<Self, Self::Reply>,
+    ) -> Self::Reply {
+        Ok(self.sequencer.chain_height())
+    }
+}
+
+impl Message<GetAccountBalance> for ExecutorActor {
+    type Reply = Balance;
+
+    async fn handle(
+        &mut self,
+        GetAccountBalance { account_id }: GetAccountBalance,
+        _ctx: &mut kameo::prelude::Context<Self, Self::Reply>,
+    ) -> Self::Reply {
+        self.sequencer
+            .with_state(|state| state.get_account_by_id(account_id).balance)
+    }
+}
+
+impl Message<GetTransaction> for ExecutorActor {
+    type Reply = Option<(LeeTransaction, BlockId)>;
+
+    async fn handle(
+        &mut self,
+        GetTransaction { tx_hash }: GetTransaction,
+        _ctx: &mut kameo::prelude::Context<Self, Self::Reply>,
+    ) -> Self::Reply {
+        self.sequencer
+            .block_store()
+            .get_transaction_by_hash(tx_hash)
+    }
+}
+
+impl Message<GetAccountNonces> for ExecutorActor {
+    type Reply = Vec<Nonce>;
+
+    async fn handle(
+        &mut self,
+        GetAccountNonces { account_ids }: GetAccountNonces,
+        _ctx: &mut kameo::prelude::Context<Self, Self::Reply>,
+    ) -> Self::Reply {
+        self.sequencer.with_state(|state| {
+            account_ids
+                .into_iter()
+                .map(|account_id| state.get_account_by_id(account_id).nonce)
+                .collect()
+        })
+    }
+}
+
+impl Message<GetProofsAndRoot> for ExecutorActor {
+    type Reply = (
+        Vec<Option<lee_core::MembershipProof>>,
+        lee_core::CommitmentSetDigest,
+    );
+
+    async fn handle(
+        &mut self,
+        GetProofsAndRoot { commitments }: GetProofsAndRoot,
+        _ctx: &mut kameo::prelude::Context<Self, Self::Reply>,
+    ) -> Self::Reply {
+        self.sequencer.with_state(|state| {
+            let proofs = commitments
+                .iter()
+                .map(|commitment| state.get_proof_for_commitment(commitment))
+                .collect();
+            (proofs, state.commitment_root())
+        })
+    }
+}
+
+impl Message<GetAccount> for ExecutorActor {
+    type Reply = GetAccountReply;
+
+    async fn handle(
+        &mut self,
+        GetAccount { account_id }: GetAccount,
+        _ctx: &mut kameo::prelude::Context<Self, Self::Reply>,
+    ) -> Self::Reply {
+        GetAccountReply {
+            account: self
+                .sequencer
+                .with_state(|state| state.get_account_by_id(account_id)),
+        }
+    }
+}
+
+impl Message<GetChannelId> for ExecutorActor {
+    type Reply = GetChannelIdReply;
+
+    async fn handle(
+        &mut self,
+        GetChannelId: GetChannelId,
+        _ctx: &mut kameo::prelude::Context<Self, Self::Reply>,
+    ) -> Self::Reply {
+        GetChannelIdReply {
+            channel_id: *self.sequencer.block_publisher().channel_id().as_ref(),
+        }
+    }
+}

lez/sequencer/actors/executor/src/protocol.rs

@@ -0,0 +1,54 @@
+use std::ops::RangeInclusive;
+
+use common::{HashType, transaction::LeeTransaction};
+use kameo::Reply;
+use lee_core::{
+    BlockId, Commitment,
+    account::{Account, AccountId},
+};
+
+pub struct Transaction {
+    pub transaction: LeeTransaction,
+}
+
+pub struct GetBlock {
+    pub block_id: BlockId,
+}
+
+pub struct GetBlockRange {
+    pub range: RangeInclusive<BlockId>,
+}
+
+pub struct GetLastBlockId;
+
+pub struct GetAccountBalance {
+    pub account_id: AccountId,
+}
+
+pub struct GetTransaction {
+    pub tx_hash: HashType,
+}
+
+pub struct GetAccountNonces {
+    pub account_ids: Vec<AccountId>,
+}
+
+pub struct GetProofsAndRoot {
+    pub commitments: Vec<Commitment>,
+}
+
+pub struct GetAccount {
+    pub account_id: AccountId,
+}
+
+#[derive(Reply)]
+pub struct GetAccountReply {
+    pub account: Account,
+}
+
+pub struct GetChannelId;
+
+#[derive(Reply)]
+pub struct GetChannelIdReply {
+    pub channel_id: [u8; 32],
+}