From 29bd8d7e5b9dd3438ff3ec2b09d6f723d98b71b4 Mon Sep 17 00:00:00 2001 From: erhant Date: Mon, 20 Jul 2026 23:42:37 +0300 Subject: [PATCH] fix(sequencer): add better handling of `store_followed_blocks` handling --- lez/sequencer/core/src/block_publisher.rs | 16 ++++++++++++ lez/sequencer/core/src/lib.rs | 2 ++ lez/sequencer/core/src/mock.rs | 8 ++++++ lez/sequencer/service/src/lib.rs | 30 ++++++++++++++++++++--- 4 files changed, 52 insertions(+), 4 deletions(-) diff --git a/lez/sequencer/core/src/block_publisher.rs b/lez/sequencer/core/src/block_publisher.rs index 47da1e44..3826177f 100644 --- a/lez/sequencer/core/src/block_publisher.rs +++ b/lez/sequencer/core/src/block_publisher.rs @@ -128,6 +128,11 @@ pub trait BlockPublisherTrait: Clone { /// Whether this sequencer is currently authorized to write to the channel. fn is_our_turn(&self) -> bool; + + /// A [`watch::Receiver`] that closes when the publisher's background driver terminates + /// (a panicked sink, an ended event stream). No channel events are + /// processed past that point, so the node must halt. + fn driver_watch(&self) -> watch::Receiver<()>; } /// Real block publisher backed by zone-sdk's `ZoneSequencer`. @@ -136,6 +141,8 @@ pub struct ZoneSdkPublisher { channel_id: ChannelId, command_tx: CommandSender, turn_rx: watch::Receiver, + // Closes when the drive task ends for any reason, including a panic. + driver_alive_rx: watch::Receiver<()>, // Aborts the drive task when the last clone is dropped. _drive_task: Arc, } @@ -185,7 +192,11 @@ impl BlockPublisherTrait for ZoneSdkPublisher { let (command_tx, mut command_rx): (CommandSender, _) = mpsc::channel(PUBLISH_INBOX_CAPACITY); + let (driver_alive_tx, driver_alive_rx) = watch::channel(()); let drive_task = tokio::spawn(async move { + // Dropped when this task ends (including panics in the sinks), + // closing every `driver_watch`. + let _driver_alive = driver_alive_tx; loop { #[expect( clippy::integer_division_remainder_used, @@ -326,6 +337,7 @@ impl BlockPublisherTrait for ZoneSdkPublisher { channel_id: config.channel_id, command_tx, turn_rx, + driver_alive_rx, _drive_task: Arc::new(DriveTaskGuard(drive_task)), }) } @@ -385,6 +397,10 @@ impl BlockPublisherTrait for ZoneSdkPublisher { fn is_our_turn(&self) -> bool { self.turn_rx.borrow().our_turn_to_write } + + fn driver_watch(&self) -> watch::Receiver<()> { + self.driver_alive_rx.clone() + } } /// Deserialize an inscription payload into `(this_msg, Block)`. Bad payloads are diff --git a/lez/sequencer/core/src/lib.rs b/lez/sequencer/core/src/lib.rs index 790fad1f..64a3ff3a 100644 --- a/lez/sequencer/core/src/lib.rs +++ b/lez/sequencer/core/src/lib.rs @@ -822,6 +822,8 @@ fn apply_follow_update( // failure is fatal: the in-memory chain has already advanced, and // continuing would leave a permanent gap in the store. // + // The `panic!` ends the drive task, whose `driver_watch` halts the node. + // // TODO: the zone-sdk checkpoint is persisted by `on_checkpoint` // before this write; a crash in between resumes past these blocks // without them ever landing in the store. Full `BlocksProcessed` diff --git a/lez/sequencer/core/src/mock.rs b/lez/sequencer/core/src/mock.rs index 5cc46900..1197d40a 100644 --- a/lez/sequencer/core/src/mock.rs +++ b/lez/sequencer/core/src/mock.rs @@ -8,6 +8,7 @@ use common::block::Block; use logos_blockchain_core::mantle::ops::channel::{ChannelId, MsgId}; use logos_blockchain_key_management_system_service::keys::Ed25519Key; use logos_blockchain_zone_sdk::sequencer::WithdrawArg; +use tokio::sync::watch; use crate::{ block_publisher::{ @@ -33,6 +34,8 @@ pub struct ConfigureChannelCall { pub struct MockBlockPublisher { channel_id: ChannelId, configure_channel_calls: Arc>>, + // Kept alive so `driver_watch` never closes: the mock driver never dies. + driver_alive: Arc>, } impl MockBlockPublisher { @@ -60,6 +63,7 @@ impl BlockPublisherTrait for MockBlockPublisher { Ok(Self { channel_id: config.channel_id, configure_channel_calls: Arc::default(), + driver_alive: Arc::new(watch::channel(()).0), }) } @@ -102,4 +106,8 @@ impl BlockPublisherTrait for MockBlockPublisher { fn is_our_turn(&self) -> bool { true } + + fn driver_watch(&self) -> watch::Receiver<()> { + self.driver_alive.subscribe() + } } diff --git a/lez/sequencer/service/src/lib.rs b/lez/sequencer/service/src/lib.rs index 2275102d..aaea1d6e 100644 --- a/lez/sequencer/service/src/lib.rs +++ b/lez/sequencer/service/src/lib.rs @@ -11,10 +11,13 @@ use mempool::MemPoolHandle; use sequencer_core::SequencerCore; #[cfg(feature = "standalone")] use sequencer_core::SequencerCoreWithMockClients as SequencerCore; -use sequencer_core::TransactionOrigin; pub use sequencer_core::config::*; +use sequencer_core::{TransactionOrigin, block_publisher::BlockPublisherTrait as _}; use sequencer_service_rpc::RpcServer as _; -use tokio::{sync::Mutex, task::JoinHandle}; +use tokio::{ + sync::{Mutex, watch}, + task::JoinHandle, +}; pub mod service; @@ -28,6 +31,9 @@ pub struct SequencerHandle { /// Option because of `Drop` which forbids to simply move out of `self` in `stopped()`. server_handle: Option, main_loop_handle: JoinHandle>, + /// Closes when the publisher's drive task terminates (e.g. a panicked + /// persist sink); no channel events are processed past that point. + driver_watch: watch::Receiver<()>, } impl SequencerHandle { @@ -35,11 +41,13 @@ impl SequencerHandle { addr: SocketAddr, server_handle: ServerHandle, main_loop_handle: JoinHandle>, + driver_watch: watch::Receiver<()>, ) -> Self { Self { addr, server_handle: Some(server_handle), main_loop_handle, + driver_watch, } } @@ -53,6 +61,7 @@ impl SequencerHandle { addr: _, server_handle, main_loop_handle, + driver_watch, } = &mut self; let server_handle = server_handle.take().expect("Server handle is set"); @@ -65,6 +74,10 @@ impl SequencerHandle { .context("Main loop task panicked")? .context("Main loop exited unexpectedly") } + // `changed()` only resolves on channel closure: nothing sends. + _ = driver_watch.changed() => { + Err(anyhow!("Publisher drive task terminated")) + } } } @@ -78,10 +91,12 @@ impl SequencerHandle { addr: _, server_handle, main_loop_handle, + driver_watch, } = self; let stopped = server_handle.as_ref().is_none_or(ServerHandle::is_stopped) - || main_loop_handle.is_finished(); + || main_loop_handle.is_finished() + || driver_watch.has_changed().is_err(); !stopped } @@ -97,6 +112,7 @@ impl Drop for SequencerHandle { addr: _, server_handle, main_loop_handle, + driver_watch: _, } = self; main_loop_handle.abort(); @@ -119,6 +135,7 @@ pub async fn run(config: SequencerConfig, listen_addr: SocketAddr) -> Result Result