From 7541d7658bc913fa354afca175bed4ca644b098b Mon Sep 17 00:00:00 2001 From: Daniil Polyakov Date: Thu, 6 Aug 2026 19:44:54 +0300 Subject: [PATCH] feat(sequencer): use actors in SequencerHandle --- Cargo.lock | 26 +- Cargo.toml | 1 + lez/sequencer/actors/executor/Cargo.toml | 3 + lez/sequencer/actors/executor/src/lib.rs | 174 ++++++++++- lez/sequencer/actors/executor/src/protocol.rs | 3 + lez/sequencer/actors/rpc_server/Cargo.toml | 1 + lez/sequencer/actors/rpc_server/src/lib.rs | 53 ++-- .../actors/rpc_server/src/protocol.rs | 10 - .../actors/rpc_server/src/service.rs | 10 +- lez/sequencer/service/Cargo.toml | 9 +- lez/sequencer/service/src/lib.rs | 271 +++++------------- 11 files changed, 298 insertions(+), 263 deletions(-) delete mode 100644 lez/sequencer/actors/rpc_server/src/protocol.rs diff --git a/Cargo.lock b/Cargo.lock index 03663aa04..5541f8f97 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4837,6 +4837,19 @@ dependencies = [ "tracing", ] +[[package]] +name = "kameo_actors" +version = "0.8.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "069ef0ae25f4da6f817ce7d81f7990b3d12c3ae398b59e6e16e37b5fcc92a443" +dependencies = [ + "futures", + "glob", + "kameo", + "thiserror 2.0.18", + "tokio", +] + [[package]] name = "kameo_macros" version = "0.21.1" @@ -9572,8 +9585,11 @@ dependencies = [ "common", "kameo", "lee_core", + "log", "mempool", "sequencer_core", + "tokio", + "tokio-util", ] [[package]] @@ -9594,6 +9610,7 @@ dependencies = [ "sequencer_rpc_server_actor_metrics", "sequencer_service_protocol", "sequencer_service_rpc", + "tokio", ] [[package]] @@ -9608,18 +9625,17 @@ name = "sequencer_service" version = "0.1.0" dependencies = [ "anyhow", - "bytesize", "clap", - "common", "env_logger", "futures", "hex", - "jsonrpsee", + "kameo", + "kameo_actors", "log", - "mempool", "metrics-exporter-prometheus", "sequencer_core", - "sequencer_service_rpc", + "sequencer_executor_actor", + "sequencer_rpc_server_actor", "tokio", "tokio-util", ] diff --git a/Cargo.toml b/Cargo.toml index 40ceca680..6a4dc90ec 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -135,6 +135,7 @@ 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" +kameo_actors = "0.8.1" 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 index d12b91106..32783be06 100644 --- a/lez/sequencer/actors/executor/Cargo.toml +++ b/lez/sequencer/actors/executor/Cargo.toml @@ -14,4 +14,7 @@ lee_core.workspace = true mempool.workspace = true kameo.workspace = true +tokio.workspace = true +tokio-util.workspace = true +log.workspace = true anyhow.workspace = true diff --git a/lez/sequencer/actors/executor/src/lib.rs b/lez/sequencer/actors/executor/src/lib.rs index 4b905aac9..3775ec2ff 100644 --- a/lez/sequencer/actors/executor/src/lib.rs +++ b/lez/sequencer/actors/executor/src/lib.rs @@ -1,30 +1,55 @@ //! Executor Actor performs the main logic of the Sequencer. -use anyhow::Result; +use std::time::Duration; + +use anyhow::{Ok, Result, anyhow, bail}; use common::{block::Block, transaction::LeeTransaction}; -use kameo::{Actor, message::Message}; +use kameo::{ + Actor, + actor::{ActorRef, WeakActorRef}, + error::ActorStopReason, + mailbox::{MailboxReceiver, Signal}, + message::{Context, Message}, +}; use lee_core::{ BlockId, account::{Balance, Nonce}, }; +use log::{error, info, warn}; use mempool::MemPoolHandle; use sequencer_core::{ SequencerCore, TransactionOrigin, block_publisher::{BlockPublisherTrait as _, ZoneSdkPublisher}, config::SequencerConfig, + task_group::{StoreRelease, TaskGroup}, }; +use tokio::select; +use tokio_util::sync::CancellationToken; use crate::protocol::{ GetAccount, GetAccountBalance, GetAccountNonces, GetAccountReply, GetBlock, GetBlockRange, - GetChannelId, GetChannelIdReply, GetLastBlockId, GetProofsAndRoot, GetTransaction, Transaction, + GetChannelId, GetChannelIdReply, GetLastBlockId, GetProofsAndRoot, GetTransaction, + ProduceBlock, Transaction, }; pub mod protocol; -#[derive(Actor)] pub struct ExecutorActor { sequencer: SequencerCore, mempool_handle: MemPoolHandle<(TransactionOrigin, LeeTransaction)>, + + // --- TODO: Remove these fields below --- + /// Cancelled when the publisher's drive task terminates (e.g. a panicked + /// persist sink); no channel events are processed past that point. + driver_cancellation: CancellationToken, + /// The core's background tasks, taken before the core was shared. This + /// handle owns no reference to the core itself, so without these there is + /// nothing to wait on: aborting the main loop only starts the teardown. + background_tasks: Vec, + /// The store, weakly. Every strong reference lives inside something this + /// handle stops, so watching the count go to zero is how shutdown knows the + /// database file is actually closed rather than assuming it from drop order. + store: StoreRelease, } impl ExecutorActor { @@ -32,20 +57,79 @@ impl ExecutorActor { let (sequencer, mempool_handle): (SequencerCore, _) = SequencerCore::start_from_config(config).await; + let driver_cancellation = sequencer.block_publisher().driver_cancellation(); + let background_tasks = sequencer.background_tasks(); + let store = sequencer.store_release(); + Self { sequencer, mempool_handle, + driver_cancellation, + background_tasks, + store, } } } +impl Actor for ExecutorActor { + type Args = Self; + type Error = anyhow::Error; + + async fn on_start(args: Self::Args, _actor_ref: ActorRef) -> Result { + Ok(args) + } + + #[expect( + clippy::integer_division_remainder_used, + reason = "Generated by select! macro, can't be easily rewritten to avoid this lint" + )] + async fn next( + &mut self, + _actor_ref: WeakActorRef, + mailbox_rx: &mut MailboxReceiver, + ) -> Result>, Self::Error> { + // TODO: Remove this please + for task in &self.background_tasks { + if task.any_finished() { + bail!("One of the sequencer's background tasks has finished unexpectedly"); + } + } + + select! { + signal = mailbox_rx.recv() => { + Ok(signal) + } + () = self.driver_cancellation.cancelled() => { + Err(anyhow!("The sequencer's block publisher has stopped unexpectedly")) + } + } + } + + async fn on_stop( + &mut self, + _actor_ref: WeakActorRef, + _reason: ActorStopReason, + ) -> std::prelude::v1::Result<(), Self::Error> { + for tasks in &self.background_tasks { + tasks.shutdown().await; + } + + // Nothing this handle owns holds the store, so waiting here rather than + // after the drop is the same thing, and it keeps the guarantee inside + // the call the caller awaits. + wait_for_store_release(&self.store).await; + + Ok(()) + } +} + impl Message for ExecutorActor { type Reply = (); async fn handle( &mut self, Transaction { transaction }: Transaction, - _ctx: &mut kameo::prelude::Context, + _ctx: &mut Context, ) -> Self::Reply { self.mempool_handle .push((TransactionOrigin::User, transaction)) @@ -60,7 +144,7 @@ impl Message for ExecutorActor { async fn handle( &mut self, GetBlock { block_id }: GetBlock, - _ctx: &mut kameo::prelude::Context, + _ctx: &mut Context, ) -> Self::Reply { self.sequencer .block_store() @@ -75,7 +159,7 @@ impl Message for ExecutorActor { async fn handle( &mut self, GetBlockRange { range }: GetBlockRange, - _ctx: &mut kameo::prelude::Context, + _ctx: &mut Context, ) -> Self::Reply { range .map_while(|block_id| { @@ -95,7 +179,7 @@ impl Message for ExecutorActor { async fn handle( &mut self, GetLastBlockId: GetLastBlockId, - _ctx: &mut kameo::prelude::Context, + _ctx: &mut Context, ) -> Self::Reply { Ok(self.sequencer.chain_height()) } @@ -107,7 +191,7 @@ impl Message for ExecutorActor { async fn handle( &mut self, GetAccountBalance { account_id }: GetAccountBalance, - _ctx: &mut kameo::prelude::Context, + _ctx: &mut Context, ) -> Self::Reply { self.sequencer .with_state(|state| state.get_account_by_id(account_id).balance) @@ -120,7 +204,7 @@ impl Message for ExecutorActor { async fn handle( &mut self, GetTransaction { tx_hash }: GetTransaction, - _ctx: &mut kameo::prelude::Context, + _ctx: &mut Context, ) -> Self::Reply { self.sequencer .block_store() @@ -134,7 +218,7 @@ impl Message for ExecutorActor { async fn handle( &mut self, GetAccountNonces { account_ids }: GetAccountNonces, - _ctx: &mut kameo::prelude::Context, + _ctx: &mut Context, ) -> Self::Reply { self.sequencer.with_state(|state| { account_ids @@ -154,7 +238,7 @@ impl Message for ExecutorActor { async fn handle( &mut self, GetProofsAndRoot { commitments }: GetProofsAndRoot, - _ctx: &mut kameo::prelude::Context, + _ctx: &mut Context, ) -> Self::Reply { self.sequencer.with_state(|state| { let proofs = commitments @@ -172,7 +256,7 @@ impl Message for ExecutorActor { async fn handle( &mut self, GetAccount { account_id }: GetAccount, - _ctx: &mut kameo::prelude::Context, + _ctx: &mut Context, ) -> Self::Reply { GetAccountReply { account: self @@ -188,10 +272,72 @@ impl Message for ExecutorActor { async fn handle( &mut self, GetChannelId: GetChannelId, - _ctx: &mut kameo::prelude::Context, + _ctx: &mut Context, ) -> Self::Reply { GetChannelIdReply { channel_id: *self.sequencer.block_publisher().channel_id().as_ref(), } } } + +impl Message for ExecutorActor { + type Reply = Result<()>; + + async fn handle( + &mut self, + ProduceBlock: ProduceBlock, + _ctx: &mut Context, + ) -> Self::Reply { + // Only produce on our turn. + if !self.sequencer.is_our_turn() { + info!("Not our turn to produce a block, skipping"); + return Ok(()); + } + + // Never inscribe a second block at a height we already published: the + // channel would carry two chains from there and nothing resolves that. + // The head rewinds under us when the sdk orphans our own unfinalized + // blocks, and recovers once they finalize, so this is a wait. + if let Some(high_water) = self.sequencer.rewound_below_published() { + warn!( + "Skipping turn: head rewound to {} but block {high_water} is already inscribed; \ + waiting for the channel to restore it", + self.sequencer.next_block_height().saturating_sub(1), + ); + return Ok(()); + } + + info!("Our turn: collecting transactions from mempool, creating block"); + let id = self.sequencer.produce_new_block().await?; + info!("Block with id {id} created"); + Ok(()) + } +} + +/// Waits until nothing holds the store any more. +/// +/// Everything that holds one lives inside a task or a server this handle has +/// already stopped, but the last drop happens on whichever thread ran them, not +/// on this one. Without this the caller can reopen the database a moment too +/// early and hit a `RocksDB` lock error, which is the kind of failure that shows +/// up as an occasional flake rather than a bug. +async fn wait_for_store_release(store: &StoreRelease) { + /// Long enough for a drop that is already in flight, short enough that a + /// leak is reported rather than hung on. + const RELEASE_TIMEOUT: Duration = Duration::from_secs(10); + const POLL: Duration = Duration::from_millis(10); + + let released = tokio::time::timeout(RELEASE_TIMEOUT, async { + while store.holders() > 0 { + tokio::time::sleep(POLL).await; + } + }) + .await; + + if released.is_err() { + error!( + "Sequencer store still held by {} reference(s) after shutdown; something outlived the tasks it should have died with", + store.holders() + ); + } +} diff --git a/lez/sequencer/actors/executor/src/protocol.rs b/lez/sequencer/actors/executor/src/protocol.rs index 282adfc99..9fb5be123 100644 --- a/lez/sequencer/actors/executor/src/protocol.rs +++ b/lez/sequencer/actors/executor/src/protocol.rs @@ -52,3 +52,6 @@ pub struct GetChannelId; pub struct GetChannelIdReply { pub channel_id: [u8; 32], } + +#[derive(Copy, Clone)] +pub struct ProduceBlock; diff --git a/lez/sequencer/actors/rpc_server/Cargo.toml b/lez/sequencer/actors/rpc_server/Cargo.toml index e0e82ecb8..9c4b86ab0 100644 --- a/lez/sequencer/actors/rpc_server/Cargo.toml +++ b/lez/sequencer/actors/rpc_server/Cargo.toml @@ -18,6 +18,7 @@ sequencer_rpc_server_actor_metrics = { workspace = true, features = ["record"] } sequencer_executor_actor.workspace = true kameo.workspace = true +tokio.workspace = true log.workspace = true jsonrpsee.workspace = true borsh.workspace = true diff --git a/lez/sequencer/actors/rpc_server/src/lib.rs b/lez/sequencer/actors/rpc_server/src/lib.rs index b9cce7792..6837aca06 100644 --- a/lez/sequencer/actors/rpc_server/src/lib.rs +++ b/lez/sequencer/actors/rpc_server/src/lib.rs @@ -2,16 +2,14 @@ use std::net::SocketAddr; -use anyhow::{Context as _, Result}; +use anyhow::{Context as _, Result, anyhow}; use bytesize::ByteSize; -use jsonrpsee::{server::ServerHandle, tracing::warn}; -use kameo::{Actor, actor::ActorRef, mailbox::Signal, message::Message}; +use jsonrpsee::server::ServerHandle; +use kameo::{Actor, actor::ActorRef, mailbox::Signal}; use log::info; use sequencer_service_rpc::RpcServer as _; +use tokio::select; -use crate::protocol::{GetAddress, GetAddressReply}; - -pub mod protocol; mod service; const REQUEST_BODY_MAX_SIZE: ByteSize = ByteSize::mib(10); @@ -25,7 +23,7 @@ impl RpcServerActor { pub async fn new( executor_ref: ActorRef, listen_addr: SocketAddr, - max_block_size: u64, + max_block_size: ByteSize, ) -> Result { let server = jsonrpsee::server::ServerBuilder::with_config( jsonrpsee::server::ServerConfigBuilder::new() @@ -53,6 +51,11 @@ impl RpcServerActor { addr, }) } + + #[must_use] + pub const fn addr(&self) -> SocketAddr { + self.addr + } } impl Actor for RpcServerActor { @@ -63,17 +66,28 @@ impl Actor for RpcServerActor { Ok(args) } + #[expect( + clippy::integer_division_remainder_used, + reason = "Generated by select! macro, can't be easily rewritten to avoid this lint" + )] async fn next( &mut self, _actor_ref: kameo::prelude::WeakActorRef, - _mailbox_rx: &mut kameo::prelude::MailboxReceiver, + mailbox_rx: &mut kameo::prelude::MailboxReceiver, ) -> Result>, Self::Error> { - if let Some(server_handle) = self.server_handle.take() { - server_handle.stopped().await; - warn!("RPC server stopped"); - } + let handle = self + .server_handle + .clone() + .expect("Server handle should be present while actor is running"); - Ok(Some(Signal::Stop)) + select! { + signal = mailbox_rx.recv() => { + Ok(signal) + } + () = handle.stopped() => { + Err(anyhow!("RPC server has stopped unexpectedly")) + } + } } async fn on_stop( @@ -83,21 +97,8 @@ impl Actor for RpcServerActor { ) -> Result<(), Self::Error> { if let Some(server_handle) = self.server_handle.take() { server_handle.stop()?; - info!("RPC server stopped"); } Ok(()) } } - -impl Message for RpcServerActor { - type Reply = GetAddressReply; - - async fn handle( - &mut self, - GetAddress: GetAddress, - _ctx: &mut kameo::prelude::Context, - ) -> Self::Reply { - GetAddressReply { addr: self.addr } - } -} diff --git a/lez/sequencer/actors/rpc_server/src/protocol.rs b/lez/sequencer/actors/rpc_server/src/protocol.rs deleted file mode 100644 index fd5978b09..000000000 --- a/lez/sequencer/actors/rpc_server/src/protocol.rs +++ /dev/null @@ -1,10 +0,0 @@ -use std::net::SocketAddr; - -use kameo::Reply; - -pub struct GetAddress; - -#[derive(Reply)] -pub struct GetAddressReply { - pub addr: SocketAddr, -} diff --git a/lez/sequencer/actors/rpc_server/src/service.rs b/lez/sequencer/actors/rpc_server/src/service.rs index 03fa29fa1..9c639d7e6 100644 --- a/lez/sequencer/actors/rpc_server/src/service.rs +++ b/lez/sequencer/actors/rpc_server/src/service.rs @@ -1,5 +1,6 @@ use std::collections::BTreeMap; +use bytesize::ByteSize; use common::transaction::LeeTransaction; use jsonrpsee::{ core::async_trait, @@ -14,13 +15,13 @@ use sequencer_service_protocol::{ pub struct Service { executor_ref: ActorRef, - max_block_size: u64, + max_block_size: ByteSize, } impl Service { pub fn new( executor_ref: ActorRef, - max_block_size: u64, + max_block_size: ByteSize, ) -> Self { sequencer_rpc_server_actor_metrics::init(); @@ -47,7 +48,10 @@ impl sequencer_service_rpc::RpcServer for Service { let tx_size = u64::try_from(encoded_tx.len()).expect("Transaction size should fit in u64"); - let max_tx_size = self.max_block_size.saturating_sub(BLOCK_HEADER_OVERHEAD); + let max_tx_size = self + .max_block_size + .as_u64() + .saturating_sub(BLOCK_HEADER_OVERHEAD); if tx_size > max_tx_size { return Err(ErrorObjectOwned::owned( diff --git a/lez/sequencer/service/Cargo.toml b/lez/sequencer/service/Cargo.toml index 06be7f847..cb3575c59 100644 --- a/lez/sequencer/service/Cargo.toml +++ b/lez/sequencer/service/Cargo.toml @@ -9,22 +9,21 @@ license = { workspace = true } workspace = true [dependencies] -common.workspace = true -mempool.workspace = true sequencer_core = { workspace = true, features = ["testnet"] } -sequencer_service_rpc = { workspace = true, features = ["server"] } +sequencer_executor_actor.workspace = true +sequencer_rpc_server_actor.workspace = true clap = { workspace = true, features = ["derive", "env"] } anyhow.workspace = true env_logger.workspace = true +kameo.workspace = true +kameo_actors.workspace = true hex.workspace = true log.workspace = true metrics-exporter-prometheus.workspace = true tokio.workspace = true tokio-util.workspace = true -jsonrpsee.workspace = true futures.workspace = true -bytesize.workspace = true [features] default = [] diff --git a/lez/sequencer/service/src/lib.rs b/lez/sequencer/service/src/lib.rs index c796c713a..e1c711be5 100644 --- a/lez/sequencer/service/src/lib.rs +++ b/lez/sequencer/service/src/lib.rs @@ -1,94 +1,69 @@ -use std::{net::SocketAddr, sync::Arc, time::Duration}; +use std::net::SocketAddr; -use anyhow::{Context as _, Result, anyhow}; +use anyhow::{Result, anyhow}; use futures::never::Never; -use jsonrpsee::server::ServerHandle; -use log::{error, info, warn}; -#[cfg(not(feature = "standalone"))] -use sequencer_core::SequencerCore; -#[cfg(feature = "standalone")] -use sequencer_core::SequencerCoreWithMockClients as SequencerCore; +use kameo::actor::{ActorRef, Spawn as _}; +use kameo_actors::scheduler::{Scheduler, SetInterval}; +use log::{error, info}; pub use sequencer_core::config::*; -use sequencer_core::{ - block_publisher::BlockPublisherTrait as _, - task_group::{StoreRelease, TaskGroup}, -}; -use tokio::{sync::Mutex, task::JoinHandle}; -use tokio_util::sync::CancellationToken; +use sequencer_executor_actor::ExecutorActor; +use sequencer_rpc_server_actor::RpcServerActor; +use tokio::select; /// Handle to manage the sequencer and its tasks. /// /// Implements `Drop` to ensure all tasks are aborted and the RPC server is stopped when dropped. pub struct SequencerHandle { + executor_ref: ActorRef, + rpc_server_ref: ActorRef, + scheduler_ref: ActorRef, addr: SocketAddr, - server_handle: ServerHandle, - main_loop_handle: JoinHandle>, - /// Cancelled when the publisher's drive task terminates (e.g. a panicked - /// persist sink); no channel events are processed past that point. - driver_cancellation: CancellationToken, - /// The core's background tasks, taken before the core was shared. This - /// handle owns no reference to the core itself, so without these there is - /// nothing to wait on: aborting the main loop only starts the teardown. - background_tasks: Vec, - /// The store, weakly. Every strong reference lives inside something this - /// handle stops, so watching the count go to zero is how shutdown knows the - /// database file is actually closed rather than assuming it from drop order. - store: StoreRelease, } impl SequencerHandle { const fn new( + executor_ref: ActorRef, + rpc_server_ref: ActorRef, + scheduler_ref: ActorRef, addr: SocketAddr, - server_handle: ServerHandle, - main_loop_handle: JoinHandle>, - driver_cancellation: CancellationToken, - background_tasks: Vec, - store: StoreRelease, ) -> Self { Self { + executor_ref, + rpc_server_ref, + scheduler_ref, addr, - server_handle, - main_loop_handle, - driver_cancellation, - background_tasks, - store, } } /// Stops the sequencer and waits for every part of it to be gone. - /// - /// `Drop` alone cannot do this: it aborts the main loop without awaiting it, - /// and the core lives behind `Arc`s held by that task and the RPC server, so - /// after a plain drop the store is still open for an unbounded stretch. That - /// is why restarting a sequencer on the same home directory used to need a - /// sleep, and why an in-process restart could fail outright with a `RocksDB` - /// lock error. - /// - /// Order matters: the main loop stops first so nothing new is produced while - /// the publisher is torn down, then the background tasks that hold the store, - /// then the server. Consuming `self` drops the last references, so the store - /// is closed by the time this returns. pub async fn shutdown(mut self) { - self.main_loop_handle.abort(); - if let Err(err) = (&mut self.main_loop_handle).await - && err.is_panic() - { - error!("Sequencer main loop panicked before shutdown: {err}"); - } + let Self { + executor_ref, + rpc_server_ref, + scheduler_ref, + addr: _, + } = &mut self; - for tasks in &self.background_tasks { - tasks.shutdown().await; + info!("Stopping Scheduler Actor..."); + if let Err(err) = scheduler_ref.stop_gracefully().await { + error!("Failed to stop Scheduler Actor gracefully: {err}"); } + scheduler_ref.wait_for_shutdown().await; + info!("Scheduler Actor stopped"); - if let Err(err) = self.server_handle.stop() { - error!("An error occurred while stopping Sequencer RPC server: {err}"); + info!("Stopping RPC Server Actor..."); + if let Err(err) = rpc_server_ref.stop_gracefully().await { + error!("Failed to stop RPC Server Actor gracefully: {err}"); } - self.server_handle.clone().stopped().await; + rpc_server_ref.wait_for_shutdown().await; + info!("RPC Server Actor stopped"); - // Nothing this handle owns holds the store, so waiting here rather than - // after the drop is the same thing, and it keeps the guarantee inside - // the call the caller awaits. - wait_for_store_release(&self.store).await; + info!("Stopping Executor Actor..."); + if let Err(err) = executor_ref.stop_gracefully().await { + error!("Failed to stop Executor Actor gracefully: {err}"); + } + executor_ref.wait_for_shutdown().await; + info!("Executor Actor stopped"); } /// Wait for any of the sequencer tasks to fail and return the error. @@ -98,28 +73,21 @@ impl SequencerHandle { )] pub async fn failed(&mut self) -> Result { let Self { + executor_ref, + rpc_server_ref, + scheduler_ref, addr: _, - server_handle, - main_loop_handle, - driver_cancellation, - background_tasks: _, - store: _, } = self; - // Cloned rather than taken: `stopped()` consumes a handle, and taking - // this one would leave `shutdown` with no way to stop the server. - let server_handle = server_handle.clone(); - tokio::select! { - () = server_handle.stopped() => { - Err(anyhow!("RPC Server stopped")) + select! { + () = executor_ref.wait_for_shutdown() => { + Err(anyhow!("Executor actor has been stopped")) } - res = main_loop_handle => { - res - .context("Main loop task panicked")? - .context("Main loop exited unexpectedly") + () = rpc_server_ref.wait_for_shutdown() => { + Err(anyhow!("RPC server actor has been stopped")) } - () = driver_cancellation.cancelled() => { - Err(anyhow!("Publisher drive task terminated")) + () = scheduler_ref.wait_for_shutdown() => { + Err(anyhow!("Scheduler actor has been stopped")) } } } @@ -131,21 +99,13 @@ impl SequencerHandle { #[must_use] pub fn is_healthy(&self) -> bool { let Self { + executor_ref, + rpc_server_ref, + scheduler_ref, addr: _, - server_handle, - main_loop_handle, - driver_cancellation, - background_tasks, - store: _, } = self; - let stopped = server_handle.is_stopped() - || main_loop_handle.is_finished() - || driver_cancellation.is_cancelled() - // A watcher only ends by panicking, and a peer whose deliveries have - // silently stopped is exactly what this predicate exists to catch. - || background_tasks.iter().any(TaskGroup::any_finished); - !stopped + executor_ref.is_alive() && rpc_server_ref.is_alive() && scheduler_ref.is_alive() } #[must_use] @@ -154,121 +114,32 @@ impl SequencerHandle { } } -impl Drop for SequencerHandle { - fn drop(&mut self) { - let Self { - addr: _, - server_handle, - main_loop_handle, - driver_cancellation: _, - background_tasks: _, - store: _, - } = self; - - main_loop_handle.abort(); - - if let Err(err) = server_handle.stop() { - error!("An error occurred while stopping Sequencer RPC server: {err}"); - } - } -} - -/// Waits until nothing holds the store any more. -/// -/// Everything that holds one lives inside a task or a server this handle has -/// already stopped, but the last drop happens on whichever thread ran them, not -/// on this one. Without this the caller can reopen the database a moment too -/// early and hit a `RocksDB` lock error, which is the kind of failure that shows -/// up as an occasional flake rather than a bug. -async fn wait_for_store_release(store: &StoreRelease) { - /// Long enough for a drop that is already in flight, short enough that a - /// leak is reported rather than hung on. - const RELEASE_TIMEOUT: Duration = Duration::from_secs(10); - const POLL: Duration = Duration::from_millis(10); - - let released = tokio::time::timeout(RELEASE_TIMEOUT, async { - while store.holders() > 0 { - tokio::time::sleep(POLL).await; - } - }) - .await; - - if released.is_err() { - error!( - "Sequencer store still held by {} reference(s) after shutdown; something outlived the tasks it should have died with", - store.holders() - ); - } -} - pub async fn run(config: SequencerConfig, listen_addr: SocketAddr) -> Result { let block_timeout = config.block_create_timeout; let max_block_size = config.max_block_size; - let (sequencer_core, mempool_handle): (SequencerCore, _) = - SequencerCore::start_from_config(config).await; + let executor_ref = ExecutorActor::spawn(ExecutorActor::new(config).await); + info!("Executor Actor spawned"); - info!("Sequencer core set up"); + let rpc_server = RpcServerActor::new(executor_ref.clone(), listen_addr, max_block_size).await?; + let addr = rpc_server.addr(); + let rpc_server_ref = RpcServerActor::spawn(rpc_server); + info!("RPC Server Actor spawned"); - let driver_cancellation = sequencer_core.block_publisher().driver_cancellation(); - // Taken while the core is still owned here: once it is behind the `Arc` - // below, the only owners are the RPC server and the main loop task, and - // neither hands it back. - let background_tasks = sequencer_core.background_tasks(); - let store = sequencer_core.store_release(); - let seq_core_wrapped = Arc::new(Mutex::new(sequencer_core)); - let mempool_handle_for_server = mempool_handle.clone(); - - let (server_handle, addr) = run_server( - Arc::clone(&seq_core_wrapped), - mempool_handle_for_server, - listen_addr, - max_block_size.as_u64(), - ) - .await?; - info!("RPC server started"); - - info!("Starting main sequencer loop"); - let main_loop_handle = tokio::spawn(main_loop(seq_core_wrapped, block_timeout)); - - let _ = mempool_handle; + let scheduler_ref = Scheduler::spawn(Scheduler::new()); + scheduler_ref + .tell(SetInterval::new( + executor_ref.downgrade(), + block_timeout, + sequencer_executor_actor::protocol::ProduceBlock, + )) + .await?; + info!("Block production scheduler started"); Ok(SequencerHandle::new( + executor_ref, + rpc_server_ref, + scheduler_ref, addr, - server_handle, - main_loop_handle, - driver_cancellation, - background_tasks, - store, )) } - -async fn main_loop(seq_core: Arc>, block_timeout: Duration) -> Result { - loop { - tokio::time::sleep(block_timeout).await; - - let mut state = seq_core.lock().await; - - // Only produce on our turn. - if !state.is_our_turn() { - continue; - } - - // Never inscribe a second block at a height we already published: the - // channel would carry two chains from there and nothing resolves that. - // The head rewinds under us when the sdk orphans our own unfinalized - // blocks, and recovers once they finalize, so this is a wait. - if let Some(high_water) = state.rewound_below_published() { - warn!( - "Skipping turn: head rewound to {} but block {high_water} is already inscribed; \ - waiting for the channel to restore it", - state.next_block_height().saturating_sub(1), - ); - continue; - } - - info!("Our turn: collecting transactions from mempool, creating block"); - let id = state.produce_new_block().await?; - info!("Block with id {id} created"); - } -}