From 436908e08f8fd21e2fa8fb1c211e1f2bfb81ce9d Mon Sep 17 00:00:00 2001 From: Daniil Polyakov Date: Thu, 6 Aug 2026 17:02:21 +0300 Subject: [PATCH] feat(sequencer): add Executor actor --- Cargo.lock | 50 ++++- Cargo.toml | 3 + lez/sequencer/actors/executor/Cargo.toml | 17 ++ lez/sequencer/actors/executor/src/lib.rs | 197 ++++++++++++++++++ lez/sequencer/actors/executor/src/protocol.rs | 54 +++++ 5 files changed, 319 insertions(+), 2 deletions(-) create mode 100644 lez/sequencer/actors/executor/Cargo.toml create mode 100644 lez/sequencer/actors/executor/src/lib.rs create mode 100644 lez/sequencer/actors/executor/src/protocol.rs diff --git a/Cargo.lock b/Cargo.lock index d67437d7c..1c00aa4ee 100644 --- a/Cargo.lock +++ b/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", ] diff --git a/Cargo.toml b/Cargo.toml index dc5df07d9..c25b58810 100644 --- a/Cargo.toml +++ b/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" diff --git a/lez/sequencer/actors/executor/Cargo.toml b/lez/sequencer/actors/executor/Cargo.toml new file mode 100644 index 000000000..d12b91106 --- /dev/null +++ b/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 diff --git a/lez/sequencer/actors/executor/src/lib.rs b/lez/sequencer/actors/executor/src/lib.rs new file mode 100644 index 000000000..4b905aac9 --- /dev/null +++ b/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, + 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 for ExecutorActor { + type Reply = (); + + async fn handle( + &mut self, + Transaction { transaction }: Transaction, + _ctx: &mut kameo::prelude::Context, + ) -> Self::Reply { + self.mempool_handle + .push((TransactionOrigin::User, transaction)) + .await + .expect("Mempool is closed, this is a bug"); + } +} + +impl Message for ExecutorActor { + type Reply = Result>; + + async fn handle( + &mut self, + GetBlock { block_id }: GetBlock, + _ctx: &mut kameo::prelude::Context, + ) -> Self::Reply { + self.sequencer + .block_store() + .get_block_at_id(block_id) + .map_err(Into::into) + } +} + +impl Message for ExecutorActor { + type Reply = Result>; + + async fn handle( + &mut self, + GetBlockRange { range }: GetBlockRange, + _ctx: &mut kameo::prelude::Context, + ) -> Self::Reply { + range + .map_while(|block_id| { + self.sequencer + .block_store() + .get_block_at_id(block_id) + .map_err(Into::into) + .transpose() + }) + .collect::, _>>() + } +} + +impl Message for ExecutorActor { + type Reply = Result; + + async fn handle( + &mut self, + GetLastBlockId: GetLastBlockId, + _ctx: &mut kameo::prelude::Context, + ) -> Self::Reply { + Ok(self.sequencer.chain_height()) + } +} + +impl Message for ExecutorActor { + type Reply = Balance; + + async fn handle( + &mut self, + GetAccountBalance { account_id }: GetAccountBalance, + _ctx: &mut kameo::prelude::Context, + ) -> Self::Reply { + self.sequencer + .with_state(|state| state.get_account_by_id(account_id).balance) + } +} + +impl Message for ExecutorActor { + type Reply = Option<(LeeTransaction, BlockId)>; + + async fn handle( + &mut self, + GetTransaction { tx_hash }: GetTransaction, + _ctx: &mut kameo::prelude::Context, + ) -> Self::Reply { + self.sequencer + .block_store() + .get_transaction_by_hash(tx_hash) + } +} + +impl Message for ExecutorActor { + type Reply = Vec; + + async fn handle( + &mut self, + GetAccountNonces { account_ids }: GetAccountNonces, + _ctx: &mut kameo::prelude::Context, + ) -> 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 for ExecutorActor { + type Reply = ( + Vec>, + lee_core::CommitmentSetDigest, + ); + + async fn handle( + &mut self, + GetProofsAndRoot { commitments }: GetProofsAndRoot, + _ctx: &mut kameo::prelude::Context, + ) -> 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 for ExecutorActor { + type Reply = GetAccountReply; + + async fn handle( + &mut self, + GetAccount { account_id }: GetAccount, + _ctx: &mut kameo::prelude::Context, + ) -> Self::Reply { + GetAccountReply { + account: self + .sequencer + .with_state(|state| state.get_account_by_id(account_id)), + } + } +} + +impl Message for ExecutorActor { + type Reply = GetChannelIdReply; + + async fn handle( + &mut self, + GetChannelId: GetChannelId, + _ctx: &mut kameo::prelude::Context, + ) -> Self::Reply { + GetChannelIdReply { + channel_id: *self.sequencer.block_publisher().channel_id().as_ref(), + } + } +} diff --git a/lez/sequencer/actors/executor/src/protocol.rs b/lez/sequencer/actors/executor/src/protocol.rs new file mode 100644 index 000000000..282adfc99 --- /dev/null +++ b/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, +} + +pub struct GetLastBlockId; + +pub struct GetAccountBalance { + pub account_id: AccountId, +} + +pub struct GetTransaction { + pub tx_hash: HashType, +} + +pub struct GetAccountNonces { + pub account_ids: Vec, +} + +pub struct GetProofsAndRoot { + pub commitments: Vec, +} + +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], +}