refactor(sequencer): add test for mempool and bring back standalone feature

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",

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 }

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

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<T> = std::result::Result<T, Error>;
 
-pub struct ExecutorActor {
-    sequencer: SequencerCore<ZoneSdkPublisher>,
+// TODO: Remove `BP` once this part is moved to a separate actor
+pub struct ExecutorActor<BP: BlockPublisherTrait> {
+    sequencer: SequencerCore<BP>,
     mempool_handle: MemPoolHandle<(TransactionOrigin, LeeTransaction)>,
 
     // --- TODO: Remove these fields below ---
@@ -49,10 +50,9 @@ pub struct ExecutorActor {
     background_tasks: Vec<TaskGroup>,
 }
 
-impl ExecutorActor {
+impl<BP: BlockPublisherTrait> ExecutorActor<BP> {
     pub async fn new(config: SequencerConfig) -> Self {
-        let (sequencer, mempool_handle): (SequencerCore, _) =
-            SequencerCore::start_from_config(config).await;
+        let (sequencer, mempool_handle) = SequencerCore::<BP>::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<BP: BlockPublisherTrait + Send + 'static> Actor for ExecutorActor<BP> {
     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<Transaction> for ExecutorActor {
+impl<BP: BlockPublisherTrait + Send + 'static> Message<Transaction> for ExecutorActor<BP> {
     type Reply = Result<()>;
 
     async fn handle(
@@ -123,11 +123,11 @@ impl Message<Transaction> for ExecutorActor {
     ) -> Self::Reply {
         self.mempool_handle
             .try_push((TransactionOrigin::User, transaction))
-            .map_err(|_| Error::MempoolIsFull)
+            .map_err(|_err| Error::MempoolIsFull)
     }
 }
 
-impl Message<GetBlock> for ExecutorActor {
+impl<BP: BlockPublisherTrait + Send + 'static> Message<GetBlock> for ExecutorActor<BP> {
     type Reply = Result<Option<Block>>;
 
     async fn handle(
@@ -142,7 +142,7 @@ impl Message<GetBlock> for ExecutorActor {
     }
 }
 
-impl Message<GetBlockRange> for ExecutorActor {
+impl<BP: BlockPublisherTrait + Send + 'static> Message<GetBlockRange> for ExecutorActor<BP> {
     type Reply = Result<Vec<Block>>;
 
     async fn handle(
@@ -162,7 +162,7 @@ impl Message<GetBlockRange> for ExecutorActor {
     }
 }
 
-impl Message<GetLastBlockId> for ExecutorActor {
+impl<BP: BlockPublisherTrait + Send + 'static> Message<GetLastBlockId> for ExecutorActor<BP> {
     type Reply = Result<BlockId>;
 
     async fn handle(
@@ -174,7 +174,7 @@ impl Message<GetLastBlockId> for ExecutorActor {
     }
 }
 
-impl Message<GetAccountBalance> for ExecutorActor {
+impl<BP: BlockPublisherTrait + Send + 'static> Message<GetAccountBalance> for ExecutorActor<BP> {
     type Reply = Balance;
 
     async fn handle(
@@ -187,7 +187,7 @@ impl Message<GetAccountBalance> for ExecutorActor {
     }
 }
 
-impl Message<GetTransaction> for ExecutorActor {
+impl<BP: BlockPublisherTrait + Send + 'static> Message<GetTransaction> for ExecutorActor<BP> {
     type Reply = Option<(LeeTransaction, BlockId)>;
 
     async fn handle(
@@ -201,7 +201,7 @@ impl Message<GetTransaction> for ExecutorActor {
     }
 }
 
-impl Message<GetAccountNonces> for ExecutorActor {
+impl<BP: BlockPublisherTrait + Send + 'static> Message<GetAccountNonces> for ExecutorActor<BP> {
     type Reply = Vec<Nonce>;
 
     async fn handle(
@@ -218,7 +218,7 @@ impl Message<GetAccountNonces> for ExecutorActor {
     }
 }
 
-impl Message<GetProofsAndRoot> for ExecutorActor {
+impl<BP: BlockPublisherTrait + Send + 'static> Message<GetProofsAndRoot> for ExecutorActor<BP> {
     type Reply = (
         Vec<Option<lee_core::MembershipProof>>,
         lee_core::CommitmentSetDigest,
@@ -239,7 +239,7 @@ impl Message<GetProofsAndRoot> for ExecutorActor {
     }
 }
 
-impl Message<GetAccount> for ExecutorActor {
+impl<BP: BlockPublisherTrait + Send + 'static> Message<GetAccount> for ExecutorActor<BP> {
     type Reply = GetAccountReply;
 
     async fn handle(
@@ -255,7 +255,7 @@ impl Message<GetAccount> for ExecutorActor {
     }
 }
 
-impl Message<GetChannelId> for ExecutorActor {
+impl<BP: BlockPublisherTrait + Send + 'static> Message<GetChannelId> for ExecutorActor<BP> {
     type Reply = GetChannelIdReply;
 
     async fn handle(
@@ -269,7 +269,7 @@ impl Message<GetChannelId> for ExecutorActor {
     }
 }
 
-impl Message<ProduceBlock> for ExecutorActor {
+impl<BP: BlockPublisherTrait + Send + 'static> Message<ProduceBlock> for ExecutorActor<BP> {
     type Reply = Result<()>;
 
     async fn handle(

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::<MockBlockPublisher>::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(())
+}

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

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<sequencer_executor_actor::ExecutorActor>,
+    pub async fn new<BP: BlockPublisherTrait + Send + 'static>(
+        executor_ref: ActorRef<sequencer_executor_actor::ExecutorActor<BP>>,
         listen_addr: SocketAddr,
         max_block_size: ByteSize,
     ) -> Result<Self> {

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<sequencer_executor_actor::ExecutorActor>,
+pub struct Service<BP: BlockPublisherTrait + Send + 'static> {
+    executor_ref: ActorRef<sequencer_executor_actor::ExecutorActor<BP>>,
     max_block_size: ByteSize,
 }
 
-impl Service {
+impl<BP: BlockPublisherTrait + Send + 'static> Service<BP> {
     pub fn new(
-        executor_ref: ActorRef<sequencer_executor_actor::ExecutorActor>,
+        executor_ref: ActorRef<sequencer_executor_actor::ExecutorActor<BP>>,
         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<BP: BlockPublisherTrait + Send + 'static> sequencer_service_rpc::RpcServer for Service<BP> {
     async fn send_transaction(&self, tx: LeeTransaction) -> Result<HashType, ErrorObjectOwned> {
         sequencer_rpc_server_actor_metrics::increment_submitted_transactions_total();
 

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<WithdrawArg>,
-    ) -> Result<PublishOutcome>;
+    ) -> impl Future<Output = Result<PublishOutcome>> + 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<WithdrawArg>,
     ) -> Result<PublishOutcome> {
         let data = borsh::to_vec(block).context("Failed to serialize block")?;

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<WithdrawArg>,
     ) -> Result<PublishOutcome> {
         // Deterministic per-block id so head dedup behaves in tests.

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<ExecutorActor>,
+    executor_ref: ActorRef<ExecutorActor<BlockPublisher>>,
     rpc_server_ref: ActorRef<RpcServerActor>,
     scheduler_ref: ActorRef<Scheduler>,
     addr: SocketAddr,
@@ -22,7 +28,7 @@ pub struct SequencerHandle {
 
 impl SequencerHandle {
     const fn new(
-        executor_ref: ActorRef<ExecutorActor>,
+        executor_ref: ActorRef<ExecutorActor<BlockPublisher>>,
         rpc_server_ref: ActorRef<RpcServerActor>,
         scheduler_ref: ActorRef<Scheduler>,
         addr: SocketAddr,

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::<Instruction>();
+
+    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();
+}

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 {