feat(sequencer): use actors in SequencerHandle

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

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"

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

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<ZoneSdkPublisher>,
     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<TaskGroup>,
+    /// 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<Self>) -> Result<Self, Self::Error> {
+        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<Self>,
+        mailbox_rx: &mut MailboxReceiver<Self>,
+    ) -> Result<Option<Signal<Self>>, 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<Self>,
+        _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<Transaction> for ExecutorActor {
     type Reply = ();
 
     async fn handle(
         &mut self,
         Transaction { transaction }: Transaction,
-        _ctx: &mut kameo::prelude::Context<Self, Self::Reply>,
+        _ctx: &mut Context<Self, Self::Reply>,
     ) -> Self::Reply {
         self.mempool_handle
             .push((TransactionOrigin::User, transaction))
@@ -60,7 +144,7 @@ impl Message<GetBlock> for ExecutorActor {
     async fn handle(
         &mut self,
         GetBlock { block_id }: GetBlock,
-        _ctx: &mut kameo::prelude::Context<Self, Self::Reply>,
+        _ctx: &mut Context<Self, Self::Reply>,
     ) -> Self::Reply {
         self.sequencer
             .block_store()
@@ -75,7 +159,7 @@ impl Message<GetBlockRange> for ExecutorActor {
     async fn handle(
         &mut self,
         GetBlockRange { range }: GetBlockRange,
-        _ctx: &mut kameo::prelude::Context<Self, Self::Reply>,
+        _ctx: &mut Context<Self, Self::Reply>,
     ) -> Self::Reply {
         range
             .map_while(|block_id| {
@@ -95,7 +179,7 @@ impl Message<GetLastBlockId> for ExecutorActor {
     async fn handle(
         &mut self,
         GetLastBlockId: GetLastBlockId,
-        _ctx: &mut kameo::prelude::Context<Self, Self::Reply>,
+        _ctx: &mut Context<Self, Self::Reply>,
     ) -> Self::Reply {
         Ok(self.sequencer.chain_height())
     }
@@ -107,7 +191,7 @@ impl Message<GetAccountBalance> for ExecutorActor {
     async fn handle(
         &mut self,
         GetAccountBalance { account_id }: GetAccountBalance,
-        _ctx: &mut kameo::prelude::Context<Self, Self::Reply>,
+        _ctx: &mut Context<Self, Self::Reply>,
     ) -> Self::Reply {
         self.sequencer
             .with_state(|state| state.get_account_by_id(account_id).balance)
@@ -120,7 +204,7 @@ impl Message<GetTransaction> for ExecutorActor {
     async fn handle(
         &mut self,
         GetTransaction { tx_hash }: GetTransaction,
-        _ctx: &mut kameo::prelude::Context<Self, Self::Reply>,
+        _ctx: &mut Context<Self, Self::Reply>,
     ) -> Self::Reply {
         self.sequencer
             .block_store()
@@ -134,7 +218,7 @@ impl Message<GetAccountNonces> for ExecutorActor {
     async fn handle(
         &mut self,
         GetAccountNonces { account_ids }: GetAccountNonces,
-        _ctx: &mut kameo::prelude::Context<Self, Self::Reply>,
+        _ctx: &mut Context<Self, Self::Reply>,
     ) -> Self::Reply {
         self.sequencer.with_state(|state| {
             account_ids
@@ -154,7 +238,7 @@ impl Message<GetProofsAndRoot> for ExecutorActor {
     async fn handle(
         &mut self,
         GetProofsAndRoot { commitments }: GetProofsAndRoot,
-        _ctx: &mut kameo::prelude::Context<Self, Self::Reply>,
+        _ctx: &mut Context<Self, Self::Reply>,
     ) -> Self::Reply {
         self.sequencer.with_state(|state| {
             let proofs = commitments
@@ -172,7 +256,7 @@ impl Message<GetAccount> for ExecutorActor {
     async fn handle(
         &mut self,
         GetAccount { account_id }: GetAccount,
-        _ctx: &mut kameo::prelude::Context<Self, Self::Reply>,
+        _ctx: &mut Context<Self, Self::Reply>,
     ) -> Self::Reply {
         GetAccountReply {
             account: self
@@ -188,10 +272,72 @@ impl Message<GetChannelId> for ExecutorActor {
     async fn handle(
         &mut self,
         GetChannelId: GetChannelId,
-        _ctx: &mut kameo::prelude::Context<Self, Self::Reply>,
+        _ctx: &mut Context<Self, Self::Reply>,
     ) -> Self::Reply {
         GetChannelIdReply {
             channel_id: *self.sequencer.block_publisher().channel_id().as_ref(),
         }
     }
 }
+
+impl Message<ProduceBlock> for ExecutorActor {
+    type Reply = Result<()>;
+
+    async fn handle(
+        &mut self,
+        ProduceBlock: ProduceBlock,
+        _ctx: &mut Context<Self, Self::Reply>,
+    ) -> 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()
+        );
+    }
+}

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;

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

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<sequencer_executor_actor::ExecutorActor>,
         listen_addr: SocketAddr,
-        max_block_size: u64,
+        max_block_size: ByteSize,
     ) -> Result<Self> {
         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<Self>,
-        _mailbox_rx: &mut kameo::prelude::MailboxReceiver<Self>,
+        mailbox_rx: &mut kameo::prelude::MailboxReceiver<Self>,
     ) -> Result<Option<Signal<Self>>, 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<GetAddress> for RpcServerActor {
-    type Reply = GetAddressReply;
-
-    async fn handle(
-        &mut self,
-        GetAddress: GetAddress,
-        _ctx: &mut kameo::prelude::Context<Self, Self::Reply>,
-    ) -> Self::Reply {
-        GetAddressReply { addr: self.addr }
-    }
-}

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

@@ -1,10 +0,0 @@
-use std::net::SocketAddr;
-
-use kameo::Reply;
-
-pub struct GetAddress;
-
-#[derive(Reply)]
-pub struct GetAddressReply {
-    pub addr: SocketAddr,
-}

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<sequencer_executor_actor::ExecutorActor>,
-    max_block_size: u64,
+    max_block_size: ByteSize,
 }
 
 impl Service {
     pub fn new(
         executor_ref: ActorRef<sequencer_executor_actor::ExecutorActor>,
-        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(

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 = []

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<ExecutorActor>,
+    rpc_server_ref: ActorRef<RpcServerActor>,
+    scheduler_ref: ActorRef<Scheduler>,
     addr: SocketAddr,
-    server_handle: ServerHandle,
-    main_loop_handle: JoinHandle<Result<Never>>,
-    /// 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<TaskGroup>,
-    /// 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<ExecutorActor>,
+        rpc_server_ref: ActorRef<RpcServerActor>,
+        scheduler_ref: ActorRef<Scheduler>,
         addr: SocketAddr,
-        server_handle: ServerHandle,
-        main_loop_handle: JoinHandle<Result<Never>>,
-        driver_cancellation: CancellationToken,
-        background_tasks: Vec<TaskGroup>,
-        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<Never> {
         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<SequencerHandle> {
     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<Mutex<SequencerCore>>, block_timeout: Duration) -> Result<Never> {
-    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");
-    }
-}