diff --git a/Cargo.lock b/Cargo.lock index 101c953d4..476b0f921 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -9582,13 +9582,19 @@ name = "sequencer_executor_actor" version = "0.1.0" dependencies = [ "anyhow", + "bytesize", "common", + "env_logger", "kameo", + "lee", "lee_core", "log", "mempool", + "num-bigint 0.4.6", "sequencer_core", "storage", + "tempfile", + "test_programs", "thiserror 2.0.18", "tokio", "tokio-util", diff --git a/Cargo.toml b/Cargo.toml index 6a4dc90ec..5c79e6bf5 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -349,6 +349,8 @@ clippy.let-underscore-untyped = "allow" # Reason: this lint is actually bad as it forces to use wildcard `..` instead of # field-by-field `_` which may lead to subtle bugs when new fields are added to the struct. clippy.unneeded-field-pattern = "allow" +# Reason: this lint makes no sense for us. +clippy.error_impl_error = "allow" # Nursery clippy.nursery = { level = "deny", priority = -1 } diff --git a/lez/sequencer/actors/executor/Cargo.toml b/lez/sequencer/actors/executor/Cargo.toml index e8ac0ba5a..377b6126c 100644 --- a/lez/sequencer/actors/executor/Cargo.toml +++ b/lez/sequencer/actors/executor/Cargo.toml @@ -20,3 +20,13 @@ tokio-util.workspace = true log.workspace = true anyhow.workspace = true thiserror.workspace = true + +[dev-dependencies] +lee.workspace = true +sequencer_core = { workspace = true, features = ["mock"] } +test_programs.workspace = true + +env_logger.workspace = true +tempfile.workspace = true +bytesize.workspace = true +num-bigint.workspace = true diff --git a/lez/sequencer/actors/executor/src/lib.rs b/lez/sequencer/actors/executor/src/lib.rs index 4ce3d5923..5b87d6ead 100644 --- a/lez/sequencer/actors/executor/src/lib.rs +++ b/lez/sequencer/actors/executor/src/lib.rs @@ -16,10 +16,8 @@ use lee_core::{ use log::{info, warn}; use mempool::MemPoolHandle; use sequencer_core::{ - SequencerCore, TransactionOrigin, - block_publisher::{BlockPublisherTrait as _, ZoneSdkPublisher}, - config::SequencerConfig, - task_group::TaskGroup, + SequencerCore, TransactionOrigin, block_publisher::BlockPublisherTrait, + config::SequencerConfig, task_group::TaskGroup, }; use tokio::select; use tokio_util::sync::CancellationToken; @@ -32,11 +30,14 @@ use crate::protocol::{ pub mod error; pub mod protocol; +#[cfg(test)] +mod tests; pub type Result = std::result::Result; -pub struct ExecutorActor { - sequencer: SequencerCore, +// TODO: Remove `BP` once this part is moved to a separate actor +pub struct ExecutorActor { + sequencer: SequencerCore, mempool_handle: MemPoolHandle<(TransactionOrigin, LeeTransaction)>, // --- TODO: Remove these fields below --- @@ -49,10 +50,9 @@ pub struct ExecutorActor { background_tasks: Vec, } -impl ExecutorActor { +impl ExecutorActor { pub async fn new(config: SequencerConfig) -> Self { - let (sequencer, mempool_handle): (SequencerCore, _) = - SequencerCore::start_from_config(config).await; + let (sequencer, mempool_handle) = SequencerCore::::start_from_config(config).await; let driver_cancellation = sequencer.block_publisher().driver_cancellation(); let background_tasks = sequencer.background_tasks(); @@ -66,7 +66,7 @@ impl ExecutorActor { } } -impl Actor for ExecutorActor { +impl Actor for ExecutorActor { type Args = Self; type Error = Error; @@ -95,7 +95,7 @@ impl Actor for ExecutorActor { Ok(signal) } () = self.driver_cancellation.cancelled() => { - return Err(Error::BlockPublisherFinishedUnexpectedly); + Err(Error::BlockPublisherFinishedUnexpectedly) } } } @@ -113,7 +113,7 @@ impl Actor for ExecutorActor { } } -impl Message for ExecutorActor { +impl Message for ExecutorActor { type Reply = Result<()>; async fn handle( @@ -123,11 +123,11 @@ impl Message for ExecutorActor { ) -> Self::Reply { self.mempool_handle .try_push((TransactionOrigin::User, transaction)) - .map_err(|_| Error::MempoolIsFull) + .map_err(|_err| Error::MempoolIsFull) } } -impl Message for ExecutorActor { +impl Message for ExecutorActor { type Reply = Result>; async fn handle( @@ -142,7 +142,7 @@ impl Message for ExecutorActor { } } -impl Message for ExecutorActor { +impl Message for ExecutorActor { type Reply = Result>; async fn handle( @@ -162,7 +162,7 @@ impl Message for ExecutorActor { } } -impl Message for ExecutorActor { +impl Message for ExecutorActor { type Reply = Result; async fn handle( @@ -174,7 +174,7 @@ impl Message for ExecutorActor { } } -impl Message for ExecutorActor { +impl Message for ExecutorActor { type Reply = Balance; async fn handle( @@ -187,7 +187,7 @@ impl Message for ExecutorActor { } } -impl Message for ExecutorActor { +impl Message for ExecutorActor { type Reply = Option<(LeeTransaction, BlockId)>; async fn handle( @@ -201,7 +201,7 @@ impl Message for ExecutorActor { } } -impl Message for ExecutorActor { +impl Message for ExecutorActor { type Reply = Vec; async fn handle( @@ -218,7 +218,7 @@ impl Message for ExecutorActor { } } -impl Message for ExecutorActor { +impl Message for ExecutorActor { type Reply = ( Vec>, lee_core::CommitmentSetDigest, @@ -239,7 +239,7 @@ impl Message for ExecutorActor { } } -impl Message for ExecutorActor { +impl Message for ExecutorActor { type Reply = GetAccountReply; async fn handle( @@ -255,7 +255,7 @@ impl Message for ExecutorActor { } } -impl Message for ExecutorActor { +impl Message for ExecutorActor { type Reply = GetChannelIdReply; async fn handle( @@ -269,7 +269,7 @@ impl Message for ExecutorActor { } } -impl Message for ExecutorActor { +impl Message for ExecutorActor { type Reply = Result<()>; async fn handle( diff --git a/lez/sequencer/actors/executor/src/tests.rs b/lez/sequencer/actors/executor/src/tests.rs new file mode 100644 index 000000000..07bf61292 --- /dev/null +++ b/lez/sequencer/actors/executor/src/tests.rs @@ -0,0 +1,91 @@ +use anyhow::Result; +use bytesize::ByteSize; +use common::transaction::LeeTransaction; +use kameo::{actor::Spawn as _, error::SendError}; +use lee::{ + AccountId, PrivateKey, PublicKey, PublicTransaction, + public_transaction::{Message, WitnessSet}, +}; +use num_bigint::BigUint; +use sequencer_core::{ + config::{BedrockConfig, SequencerConfig}, + mock::MockBlockPublisher, +}; +use tokio::test; + +use crate::{ExecutorActor, protocol}; + +fn sequencer_config() -> (SequencerConfig, tempfile::TempDir) { + let home = tempfile::tempdir().expect("Failed to create tmp home dir"); + + let config = SequencerConfig { + home: home.path().to_path_buf(), + max_num_tx_in_block: 10, + max_block_size: ByteSize::kib(1024), + mempool_max_size: 10, + block_create_timeout: std::time::Duration::from_secs(5), + retry_pending_blocks_timeout: std::time::Duration::from_secs(5), + signing_key: [37; 32], + bedrock_config: BedrockConfig { + channel_id: [0; 32].into(), + node_url: "http://not-used".parse().expect("Failed to parse URL"), + auth: None, + funding_key: BigUint::default().into(), + priority_fee: sequencer_core::config::default_priority_fee(), + }, + genesis: Vec::new(), + cross_zone: None, + metrics_address: None, + }; + + (config, home) +} + +fn test_transaction() -> LeeTransaction { + let key1 = PrivateKey::new_os_random(); + let key2 = PrivateKey::new_os_random(); + let acc1 = AccountId::from(&PublicKey::new_from_private_key(&key1)); + let acc2 = AccountId::from(&PublicKey::new_from_private_key(&key2)); + + let nonces = vec![0_u128.into(), 0_u128.into()]; + let instruction = 1337; + let message = Message::try_new( + test_programs::simple_balance_transfer().id(), + vec![acc1, acc2], + nonces, + instruction, + ) + .unwrap(); + + let witness_set = WitnessSet::for_message(&message, &[&key1, &key2]); + PublicTransaction::new(message, witness_set).into() +} + +#[test] +async fn handle_transaction_fails_on_full_mempool() -> Result<()> { + let _res = env_logger::try_init(); + + let (config, _home) = sequencer_config(); + let mempool_max_size = config.mempool_max_size; + let executor = ExecutorActor::spawn(ExecutorActor::::new(config).await); + + // Fill mempool + for _ in 0..mempool_max_size { + let tx = test_transaction(); + executor + .ask(protocol::Transaction { transaction: tx }) + .await?; + } + + // Now the mempool is full, the next transaction should fail + let tx = test_transaction(); + assert!(matches!( + executor + .ask(protocol::Transaction { transaction: tx }) + .await + .map_err(SendError::err), + Err(Some(crate::error::Error::MempoolIsFull)) + )); + + Ok(()) +} diff --git a/lez/sequencer/actors/rpc_server/Cargo.toml b/lez/sequencer/actors/rpc_server/Cargo.toml index 9c4b86ab0..2b37752bf 100644 --- a/lez/sequencer/actors/rpc_server/Cargo.toml +++ b/lez/sequencer/actors/rpc_server/Cargo.toml @@ -7,6 +7,10 @@ license = { workspace = true } [lints] workspace = true +[features] +default = [] +standalone = ["sequencer_core/mock"] + [dependencies] lee.workspace = true common.workspace = true diff --git a/lez/sequencer/actors/rpc_server/src/lib.rs b/lez/sequencer/actors/rpc_server/src/lib.rs index b9b6eaf7e..1e29210d7 100644 --- a/lez/sequencer/actors/rpc_server/src/lib.rs +++ b/lez/sequencer/actors/rpc_server/src/lib.rs @@ -7,6 +7,7 @@ use bytesize::ByteSize; use jsonrpsee::server::ServerHandle; use kameo::{Actor, actor::ActorRef, mailbox::Signal}; use log::info; +use sequencer_core::block_publisher::BlockPublisherTrait; use sequencer_service_rpc::RpcServer as _; use tokio::select; @@ -20,8 +21,8 @@ pub struct RpcServerActor { } impl RpcServerActor { - pub async fn new( - executor_ref: ActorRef, + pub async fn new( + executor_ref: ActorRef>, listen_addr: SocketAddr, max_block_size: ByteSize, ) -> Result { diff --git a/lez/sequencer/actors/rpc_server/src/service.rs b/lez/sequencer/actors/rpc_server/src/service.rs index 70de44581..eb511a4c8 100644 --- a/lez/sequencer/actors/rpc_server/src/service.rs +++ b/lez/sequencer/actors/rpc_server/src/service.rs @@ -8,19 +8,20 @@ use jsonrpsee::{ }; use kameo::actor::ActorRef; use log::{error, warn}; +use sequencer_core::block_publisher::BlockPublisherTrait; use sequencer_service_protocol::{ Account, AccountId, Block, BlockId, ChannelId, Commitment, CommitmentSetDigest, CrossZoneDeadLetter, CrossZoneDeadLetterReport, HashType, MembershipProof, Nonce, ProgramId, }; -pub struct Service { - executor_ref: ActorRef, +pub struct Service { + executor_ref: ActorRef>, max_block_size: ByteSize, } -impl Service { +impl Service { pub fn new( - executor_ref: ActorRef, + executor_ref: ActorRef>, max_block_size: ByteSize, ) -> Self { sequencer_rpc_server_actor_metrics::init(); @@ -33,7 +34,7 @@ impl Service { } #[async_trait] -impl sequencer_service_rpc::RpcServer for Service { +impl sequencer_service_rpc::RpcServer for Service { async fn send_transaction(&self, tx: LeeTransaction) -> Result { sequencer_rpc_server_actor_metrics::increment_submitted_transactions_total(); diff --git a/lez/sequencer/core/src/block_publisher.rs b/lez/sequencer/core/src/block_publisher.rs index ace601608..01fad67cf 100644 --- a/lez/sequencer/core/src/block_publisher.rs +++ b/lez/sequencer/core/src/block_publisher.rs @@ -122,11 +122,11 @@ pub trait BlockPublisherTrait: Sized { /// /// The checkpoint must be persisted with the block — restoring an older one /// drops the inscription from the pending set, and it is never resubmitted. - async fn publish_block( - &self, - block: &Block, + fn publish_block<'blk, 'pbl: 'blk>( + &'pbl self, + block: &'blk Block, withdrawals: Vec, - ) -> Result; + ) -> impl Future> + Send + 'blk; fn channel_id(&self) -> ChannelId; @@ -350,9 +350,9 @@ impl BlockPublisherTrait for ZoneSdkPublisher { }) } - async fn publish_block( - &self, - block: &Block, + async fn publish_block<'blk, 'pbl: 'blk>( + &'pbl self, + block: &'blk Block, withdrawals: Vec, ) -> Result { let data = borsh::to_vec(block).context("Failed to serialize block")?; diff --git a/lez/sequencer/core/src/mock.rs b/lez/sequencer/core/src/mock.rs index b35e3be39..a8dd490b9 100644 --- a/lez/sequencer/core/src/mock.rs +++ b/lez/sequencer/core/src/mock.rs @@ -70,9 +70,9 @@ impl BlockPublisherTrait for MockBlockPublisher { }) } - async fn publish_block( - &self, - block: &Block, + async fn publish_block<'blk, 'pbl: 'blk>( + &'pbl self, + block: &'blk Block, withdrawals: Vec, ) -> Result { // Deterministic per-block id so head dedup behaves in tests. diff --git a/lez/sequencer/service/src/lib.rs b/lez/sequencer/service/src/lib.rs index 8f4caaa27..1f7b78140 100644 --- a/lez/sequencer/service/src/lib.rs +++ b/lez/sequencer/service/src/lib.rs @@ -10,11 +10,17 @@ use sequencer_executor_actor::ExecutorActor; use sequencer_rpc_server_actor::RpcServerActor; use tokio::select; +#[cfg(not(feature = "standalone"))] +type BlockPublisher = sequencer_core::block_publisher::ZoneSdkPublisher; + +#[cfg(feature = "standalone")] +type BlockPublisher = sequencer_core::mock::MockBlockPublisher; + /// Handle to manage the sequencer and its tasks. /// /// Implements `Drop` to ensure all actors are killed when dropped. pub struct SequencerHandle { - executor_ref: ActorRef, + executor_ref: ActorRef>, rpc_server_ref: ActorRef, scheduler_ref: ActorRef, addr: SocketAddr, @@ -22,7 +28,7 @@ pub struct SequencerHandle { impl SequencerHandle { const fn new( - executor_ref: ActorRef, + executor_ref: ActorRef>, rpc_server_ref: ActorRef, scheduler_ref: ActorRef, addr: SocketAddr, diff --git a/test_programs/guest/src/bin/simple_balance_transfer.rs b/test_programs/guest/src/bin/simple_balance_transfer.rs new file mode 100644 index 000000000..addc4a191 --- /dev/null +++ b/test_programs/guest/src/bin/simple_balance_transfer.rs @@ -0,0 +1,57 @@ +use lee_core::program::{AccountPostState, Claim, ProgramInput, ProgramOutput, read_lee_inputs}; + +type Instruction = u128; + +fn main() { + let ( + ProgramInput { + self_program_id, + caller_program_id, + pre_states, + instruction: balance, + }, + instruction_words, + ) = read_lee_inputs::(); + + if let Ok([account_pre]) = <[_; 1]>::try_from(pre_states.clone()) { + let account_post = + AccountPostState::new_claimed_if_default(account_pre.account, Claim::Authorized); + + ProgramOutput::new( + self_program_id, + caller_program_id, + instruction_words, + pre_states, + vec![account_post], + ) + .write(); + return; + } + + let Ok([sender_pre, receiver_pre]) = <[_; 2]>::try_from(pre_states) else { + return; + }; + + let mut sender_post = sender_pre.account.clone(); + let mut receiver_post = receiver_pre.account.clone(); + sender_post.balance = sender_post + .balance + .checked_sub(balance) + .expect("Not enough balance to transfer"); + receiver_post.balance = receiver_post + .balance + .checked_add(balance) + .expect("Overflow when adding balance"); + + ProgramOutput::new( + self_program_id, + caller_program_id, + instruction_words, + vec![sender_pre, receiver_pre], + vec![ + AccountPostState::new_claimed_if_default(sender_post, Claim::Authorized), + AccountPostState::new_claimed_if_default(receiver_post, Claim::Authorized), + ], + ) + .write(); +} diff --git a/test_programs/src/lib.rs b/test_programs/src/lib.rs index c13a10282..f1df3e961 100644 --- a/test_programs/src/lib.rs +++ b/test_programs/src/lib.rs @@ -11,6 +11,21 @@ mod guests { include!(concat!(env!("OUT_DIR"), "/methods.rs")); } +#[must_use] +#[inline] +pub const fn simple_balance_transfer() -> Program { + use guests::{ + SIMPLE_BALANCE_TRANSFER_ELF, SIMPLE_BALANCE_TRANSFER_ID, SIMPLE_BALANCE_TRANSFER_PATH, + }; + + let _unused = SIMPLE_BALANCE_TRANSFER_PATH; + + Program::new_unchecked( + SIMPLE_BALANCE_TRANSFER_ID, + Cow::Borrowed(SIMPLE_BALANCE_TRANSFER_ELF), + ) +} + #[must_use] #[inline] pub const fn chain_caller() -> Program {