fix(sequencer): add better handling of store_followed_blocks handling

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<TurnNotification>,
+    // 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<DriveTaskGuard>,
 }
@@ -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

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`

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<Mutex<Vec<ConfigureChannelCall>>>,
+    // Kept alive so `driver_watch` never closes: the mock driver never dies.
+    driver_alive: Arc<watch::Sender<()>>,
 }
 
 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()
+    }
 }

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<ServerHandle>,
     main_loop_handle: JoinHandle<Result<Never>>,
+    /// 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<Result<Never>>,
+        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<Seq
 
     info!("Sequencer core set up");
 
+    let driver_watch = sequencer_core.block_publisher().driver_watch();
     let seq_core_wrapped = Arc::new(Mutex::new(sequencer_core));
     let mempool_handle_for_server = mempool_handle.clone();
 
@@ -136,7 +153,12 @@ pub async fn run(config: SequencerConfig, listen_addr: SocketAddr) -> Result<Seq
 
     let _ = mempool_handle;
 
-    Ok(SequencerHandle::new(addr, server_handle, main_loop_handle))
+    Ok(SequencerHandle::new(
+        addr,
+        server_handle,
+        main_loop_handle,
+        driver_watch,
+    ))
 }
 
 async fn run_server(