fix(zone-sdk): make block event processing cancellation-safe (#3334)

zone-sdk/src/sequencer/actor.rs

@@ -6,7 +6,7 @@
 use std::collections::HashSet;
 
 use lb_common_http_client::{ProcessedBlockEvent, Slot};
-use lb_core::mantle::channel::ChannelState;
+use lb_core::mantle::{channel::ChannelState, ops::channel::ChannelId};
 use tracing::{debug, error, warn};
 
 use super::{
@@ -26,29 +26,34 @@ impl<Node> ZoneSequencer<Node>
 where
     Node: adapter::Node + Clone + Send + Sync + 'static,
 {
-    /// Handle a single item from the blocks stream. `None` means the stream
-    /// disconnected; any other value is processed as a block event and
-    /// produces an [`Event::BlocksProcessed`] carrying the checkpoint, the
-    /// optional `ChannelUpdate`, and the block's finalized txs.
-    pub(super) async fn handle_stream_item(
-        &mut self,
-        maybe_event: Option<ProcessedBlockEvent>,
-    ) -> Option<Event> {
-        let Some(block_event) = maybe_event else {
-            warn!(target: TARGET, "Blocks stream disconnected, will reconnect on next call");
-            self.blocks_stream = None;
-            self.handle_stream_drop();
-            return None;
-        };
+    /// Process the block retained by the drive loop.
+    ///
+    /// The pending block is cleared only after processing completes. Dropping
+    /// the surrounding [`ZoneSequencer::next_event`] future at an `.await`
+    /// therefore leaves the block available for the next call.
+    pub(super) async fn process_pending_block_event(&mut self) -> Option<Event> {
+        let block_event = self
+            .pending_block_event
+            .clone()
+            .expect("called only with a pending block event");
 
         if let Ok(result) = self.process_block_event(&block_event).await {
+            self.pending_block_event = None;
             self.finish_block_processing(result)
+                .map(|event| self.emit_now(event))
         } else {
+            self.pending_block_event = None;
             self.handle_stream_drop();
             None
         }
     }
 
+    pub(super) fn handle_stream_disconnect(&mut self) {
+        warn!(target: TARGET, "Blocks stream disconnected, will reconnect on next call");
+        self.blocks_stream = None;
+        self.handle_stream_drop();
+    }
+
     /// Ingest one live block event into local state. On any per-block error
     /// (block processing, channel-state refresh) the stream is dropped so
     /// the reconnect path retries the same event, and `Err(())` is returned
@@ -57,6 +62,19 @@ where
         &mut self,
         block_event: &ProcessedBlockEvent,
     ) -> Result<BlockEventResult, ()> {
+        // Fetch first, then install the new channel state only after the block
+        // has been processed. This avoids an `.await` after block state starts
+        // changing, which is the unsafe cancellation boundary.
+        let channel_state = fetch_channel_state(&self.node, self.channel_id)
+            .await
+            .map_err(|err| {
+                error!(
+                    target: TARGET,
+                    "Failed to refresh channel state before block processing; dropping stream so reconnect retries: {err}"
+                );
+                self.blocks_stream = None;
+            })?;
+
         let result = handle_block_event(
             block_event,
             &mut self.state,
@@ -78,13 +96,7 @@ where
             slot_clock.observe_slot(block_event.tip_slot);
         }
 
-        self.refresh_channel_state().await.map_err(|err| {
-            error!(
-                target: TARGET,
-                "Failed to refresh channel state after block; dropping stream so reconnect retries: {err}"
-            );
-            self.blocks_stream = None;
-        })?;
+        self.install_channel_state(channel_state);
 
         Ok(result)
     }
@@ -190,16 +202,17 @@ where
         checkpoint
     }
 
-    pub(super) async fn refresh_channel_state(&mut self) -> Result<(), Error> {
-        let channel = self
-            .node
-            .channel_state(self.channel_id)
-            .await
-            .map_err(|err| Error::Network(err.to_string()))?;
+    fn install_channel_state(&mut self, channel: Option<ChannelState>) {
         self.own_key_index = channel
             .as_ref()
             .and_then(|channel| self.own_key_index_for(channel));
         self.channel_state = channel;
+    }
+
+    pub(super) async fn refresh_channel_state(&mut self) -> Result<(), Error> {
+        let channel = fetch_channel_state(&self.node, self.channel_id).await?;
+        self.install_channel_state(channel);
+
         Ok(())
     }
 
@@ -593,6 +606,18 @@ where
     }
 }
 
+async fn fetch_channel_state<Node>(
+    node: &Node,
+    channel_id: ChannelId,
+) -> Result<Option<ChannelState>, Error>
+where
+    Node: adapter::Node + Sync,
+{
+    node.channel_state(channel_id)
+        .await
+        .map_err(|err| Error::Network(err.to_string()))
+}
+
 #[cfg(test)]
 mod tests {
     use lb_core::{
@@ -604,7 +629,7 @@ mod tests {
             ops::{
                 OpProof,
                 channel::{
-                    ChannelId, MsgId,
+                    MsgId,
                     config::{ChannelConfigOp, Keys},
                     deposit::DepositOp,
                     inscribe::{Inscription, InscriptionOp},
@@ -618,7 +643,7 @@ mod tests {
     use lb_key_management_system_service::keys::{Ed25519Key, ZkKey};
     use num_bigint::BigUint;
     use rand::{RngCore as _, thread_rng};
-    use tokio::sync::watch;
+    use tokio::sync::{mpsc, watch};
 
     use super::{
         super::{
@@ -628,8 +653,8 @@ mod tests {
         *,
     };
     use crate::test_support::{
-        MockNode, StreamEnd, StreamScript, api_block, funding_config, live_event, scripts,
-        single_key_channel_state, unverified_tx_with_ops,
+        MockNode, StreamEnd, StreamScript, api_block, funding_config, header_id, live_event,
+        scripts, single_key_channel_state, unverified_tx_with_ops,
     };
 
     #[must_use]
@@ -716,6 +741,158 @@ mod tests {
         }
     }
 
+    #[tokio::test]
+    async fn cancelled_next_event_resumes_the_pulled_block() {
+        let channel_id = ChannelId::from([0; 32]);
+        let sequencer_key = Ed25519Key::from_bytes(&[0; 32]);
+        let first_block = api_block(1, 0, 1, Vec::new());
+        let second_block = api_block(2, 1, 2, Vec::new());
+        let (gate_tx, gate_rx) = watch::channel(true);
+        let (calls_tx, mut calls_rx) = mpsc::unbounded_channel();
+        let node = MockNode {
+            scripts: scripts(vec![StreamScript {
+                events: vec![live_event(&first_block), live_event(&second_block)],
+                then: StreamEnd::Hang,
+            }]),
+            channel_state_gate: Some(gate_rx),
+            channel_state_calls: Some(calls_tx),
+            ..MockNode::default()
+        };
+        let mut sequencer =
+            ZoneSequencer::init(channel_id, sequencer_key, node, funding_config(), None);
+
+        loop {
+            if matches!(sequencer.next_event().await, Event::Ready) {
+                break;
+            }
+        }
+
+        assert!(matches!(
+            sequencer.next_event().await,
+            Event::BlocksProcessed { .. }
+        ));
+
+        while calls_rx.try_recv().is_ok() {}
+        gate_tx.send(false).unwrap();
+
+        {
+            let next_event = sequencer.next_event();
+            tokio::pin!(next_event);
+
+            tokio::select! {
+                call = calls_rx.recv() => {
+                    call.expect("channel-state call should be observed");
+                }
+                event = &mut next_event => {
+                    panic!("block processing completed while its node request was gated: {event:?}");
+                }
+            }
+        }
+
+        assert_eq!(
+            sequencer
+                .pending_block_event
+                .as_ref()
+                .map(|event| event.block.header.id),
+            Some(header_id(2))
+        );
+
+        gate_tx.send(true).unwrap();
+        let resumed =
+            tokio::time::timeout(std::time::Duration::from_secs(1), sequencer.next_event())
+                .await
+                .expect("the retained block should resume after cancellation");
+
+        assert!(matches!(resumed, Event::BlocksProcessed { .. }));
+        assert_eq!(sequencer.current_tip, Some(header_id(2)));
+        assert!(sequencer.pending_block_event.is_none());
+
+        assert!(
+            tokio::time::timeout(std::time::Duration::from_millis(20), sequencer.next_event(),)
+                .await
+                .is_err(),
+            "the resumed block must not be emitted twice"
+        );
+    }
+
+    #[tokio::test]
+    async fn cancelled_finalized_backfill_restarts_without_partial_state() {
+        let channel_id = ChannelId::from([0; 32]);
+        let sequencer_key = Ed25519Key::from_bytes(&[0; 32]);
+        let first_block = api_block(1, 0, 1, Vec::new());
+        let second_block = api_block(2, 1, 2, Vec::new());
+        let second_event = ProcessedBlockEvent {
+            block: second_block.clone(),
+            tip: second_block.header.id,
+            tip_slot: second_block.header.slot,
+            lib: first_block.header.id,
+            lib_slot: first_block.header.slot,
+        };
+        let (gate_tx, gate_rx) = watch::channel(true);
+        let (calls_tx, mut calls_rx) = mpsc::unbounded_channel();
+        let node = MockNode {
+            scripts: scripts(vec![StreamScript {
+                events: vec![live_event(&first_block), second_event],
+                then: StreamEnd::Hang,
+            }]),
+            immutable: vec![first_block.clone()],
+            immutable_blocks_gate: Some(gate_rx),
+            immutable_blocks_calls: Some(calls_tx),
+            ..MockNode::default()
+        };
+        let mut sequencer =
+            ZoneSequencer::init(channel_id, sequencer_key, node, funding_config(), None);
+
+        loop {
+            if matches!(sequencer.next_event().await, Event::Ready) {
+                break;
+            }
+        }
+
+        assert!(matches!(
+            sequencer.next_event().await,
+            Event::BlocksProcessed { .. }
+        ));
+
+        while calls_rx.try_recv().is_ok() {}
+        gate_tx.send(false).unwrap();
+
+        {
+            let next_event = sequencer.next_event();
+            tokio::pin!(next_event);
+
+            tokio::select! {
+                call = calls_rx.recv() => {
+                    call.expect("immutable-blocks call should be observed");
+                }
+                event = &mut next_event => {
+                    panic!("finalized backfill completed while its node request was gated: {event:?}");
+                }
+            }
+        }
+
+        assert_eq!(sequencer.lib_slot, Slot::genesis());
+        assert_eq!(sequencer.current_tip, Some(first_block.header.id));
+        assert_eq!(
+            sequencer
+                .pending_block_event
+                .as_ref()
+                .map(|event| event.block.header.id),
+            Some(second_block.header.id)
+        );
+
+        gate_tx.send(true).unwrap();
+        let resumed =
+            tokio::time::timeout(std::time::Duration::from_secs(1), sequencer.next_event())
+                .await
+                .expect("the finalized backfill should resume after cancellation");
+
+        assert!(matches!(resumed, Event::BlocksProcessed { .. }));
+        assert_eq!(sequencer.lib_slot, first_block.header.slot);
+        assert_eq!(sequencer.current_tip, Some(second_block.header.id));
+        assert!(sequencer.pending_block_event.is_none());
+    }
+
     /// A `SequencerClient::publish` issued while the node is down (reconnect
     /// in progress) must resolve promptly with [`Error::Unavailable`] —
     /// funding needs the node — instead of blocking until connectivity is

zone-sdk/src/sequencer/block_fetch.rs

@@ -1,6 +1,6 @@
-use std::collections::HashMap;
+use std::collections::{HashMap, HashSet};
 
-use lb_common_http_client::{ProcessedBlockEvent, Slot};
+use lb_common_http_client::{ApiBlock, ProcessedBlockEvent, Slot};
 use lb_core::{
     crypto::Hash,
     events::DepositRecreatedNotes,
@@ -46,6 +46,19 @@ pub(super) struct BlockEventResult {
     pub(super) mined_inscriptions: Vec<InscriptionInfo>,
 }
 
+struct PreparedBlockEvent<'a> {
+    block: &'a ApiBlock,
+    tip: HeaderId,
+    lib: HeaderId,
+    lib_slot: Slot,
+    lib_advanced: bool,
+    finalized: Vec<PreparedFinalizedBlock>,
+    canonical_backfill: Vec<ApiBlock>,
+    our_txs: Vec<TxHash>,
+    channel_txs: Vec<BlockChannelTx>,
+    mined_inscriptions: Vec<InscriptionInfo>,
+}
+
 /// Process a block event. Returns finalized tx hashes and optional channel
 /// update.
 ///
@@ -64,23 +77,103 @@ pub(super) async fn handle_block_event<Node>(
 where
     Node: adapter::Node + Sync,
 {
-    let block_id = event.block.header.id;
+    let prepared = prepare_block_event(event, state.as_ref(), *lib_slot, channel_id, node).await?;
+
+    Ok(apply_prepared_block_event(
+        prepared,
+        state,
+        current_tip,
+        lib_slot,
+        channel_id,
+    ))
+}
+
+async fn prepare_block_event<'a, Node>(
+    event: &'a ProcessedBlockEvent,
+    state: Option<&TxState>,
+    lib_slot: Slot,
+    channel_id: ChannelId,
+    node: &Node,
+) -> Result<PreparedBlockEvent<'a>, Error>
+where
+    Node: adapter::Node + Sync,
+{
+    let state_lib = state.map_or(event.lib, TxState::lib);
+    let lib_advanced = event.lib != state_lib;
+    let finalized = if lib_advanced {
+        let from: u64 = lib_slot.into();
+        let to: u64 = event.lib_slot.into();
+        if from < to {
+            prepare_finalized_blocks(from + 1, to, channel_id, node).await?
+        } else {
+            Vec::new()
+        }
+    } else {
+        Vec::new()
+    };
+
+    let finalized_block_ids: HashSet<HeaderId> =
+        finalized.iter().map(|block| block.block_id).collect();
     let parent_id = event.block.header.parent_block;
-    let tip = event.tip;
-    let lib = event.lib;
+    let parent_known = block_is_known(state, &finalized_block_ids, state_lib, parent_id);
+    let canonical_backfill = if parent_known {
+        Vec::new()
+    } else {
+        walk_back_to_known(state, &finalized_block_ids, state_lib, parent_id, node).await
+    };
+
+    let our_txs: Vec<TxHash> = event
+        .block
+        .transactions
+        .iter()
+        .filter(|tx| touches_channel_tip(tx, channel_id))
+        .map(|tx| tx.mantle_tx().hash())
+        .collect();
+    let channel_txs = classify_channel_txs(&event.block.transactions, channel_id);
+    let mined_inscriptions = channel_txs
+        .iter()
+        .flat_map(BlockChannelTx::infos)
+        .cloned()
+        .collect();
+
+    Ok(PreparedBlockEvent {
+        block: &event.block,
+        tip: event.tip,
+        lib: event.lib,
+        lib_slot: event.lib_slot,
+        lib_advanced,
+        finalized,
+        canonical_backfill,
+        our_txs,
+        channel_txs,
+        mined_inscriptions,
+    })
+}
+
+fn apply_prepared_block_event(
+    prepared: PreparedBlockEvent<'_>,
+    state: &mut Option<TxState>,
+    current_tip: &mut Option<HeaderId>,
+    lib_slot: &mut Slot,
+    channel_id: ChannelId,
+) -> BlockEventResult {
+    let PreparedBlockEvent {
+        block,
+        tip,
+        lib,
+        lib_slot: next_lib_slot,
+        lib_advanced,
+        finalized,
+        canonical_backfill,
+        our_txs,
+        channel_txs,
+        mined_inscriptions,
+    } = prepared;
 
-    // Initialize state on first event
     if state.is_none() {
         *state = Some(TxState::new(lib, MsgId::root()));
     }
-
-    let Some(s) = state.as_mut() else {
-        return Ok(BlockEventResult {
-            finalized_items: Vec::new(),
-            channel_update: None,
-            mined_inscriptions: Vec::new(),
-        });
-    };
+    let s = state.as_mut().expect("state initialized above");
 
     let old_tip = *current_tip;
 
@@ -90,67 +183,41 @@ where
     // observed ones) from genuinely new network entries.
     let tracked_before = s.tracked_tx_hashes();
 
-    // Backfill if needed (self-healing on every event)
-    // 1. Backfill finalized blocks up to LIB (only when state's LIB is behind).
-    //    Done BEFORE we advance `*lib_slot` and BEFORE we mutate state for the live
-    //    event — so on a fetch failure the caller can retry the same event next
-    //    time around. Deliberately does NOT observe inscriptions into the pending
-    //    set: these blocks become finalized in this very event, and pending mirrors
-    //    the channel ABOVE LIB — observing here would insert entries only for the
-    //    `remove_pending(lib_finalized)` sweep below to delete.
-    let mut lib_finalized = Vec::new();
-    let mut finalized_items: Vec<FinalizedTx> = Vec::new();
-    if lib != s.lib() {
-        let new_lib_slot = event.lib_slot;
-        let from: u64 = (*lib_slot).into();
-        let to: u64 = new_lib_slot.into();
-        if from < to {
-            let batch = fetch_and_process_blocks(s, from + 1, to, channel_id, node).await?;
-            lib_finalized = batch.our_tx_hashes;
-            finalized_items = batch.items;
-        }
-        *lib_slot = new_lib_slot;
+    // Install finalized history first. It is not mirrored into pending: the
+    // matching local entries are removed below using the returned hashes.
+    let finalized_batch = apply_finalized_blocks(s, finalized);
+    if lib_advanced {
+        *lib_slot = next_lib_slot;
     }
 
-    // Capture the old-tip lineage before anything below adds blocks: the
+    // Capture the old-tip lineage before canonical backfill adds blocks: the
     // lineage walk bridges through held blocks, and whatever is already in
-    // the store lands on the "before" side of the update diff. Kept after
-    // the LIB backfill, whose content surfaces via `finalized` instead.
+    // the store lands on the "before" side of the update diff.
     let old_lineage = old_tip.map(|old| s.channel_lineage(old));
 
-    // 2. Backfill canonical chain if parent is missing
-    if !s.has_block(&parent_id) && parent_id != s.lib() {
-        backfill_canonical(s, parent_id, channel_id, node).await;
+    let current_lib = s.lib();
+    for block in &canonical_backfill {
+        apply_backfilled_block(s, block, channel_id, current_lib);
     }
 
-    // Extract tx hashes and inscription info for our channel
-    let our_txs: Vec<TxHash> = event
-        .block
-        .transactions
-        .iter()
-        .filter(|tx| touches_channel_tip(tx, channel_id))
-        .map(|tx| tx.mantle_tx().hash())
-        .collect();
-
-    let channel_txs = classify_channel_txs(&event.block.transactions, channel_id);
-    let mined_inscriptions: Vec<InscriptionInfo> = channel_txs
-        .iter()
-        .flat_map(BlockChannelTx::infos)
-        .cloned()
-        .collect();
-
     // Mirror this block's inscriptions into the pending set BEFORE
     // `process_block`, so on-branch entries land in the block's safe set and
     // are excluded from re-posting while canonical.
-    observe_channel_inscriptions(s, &channel_txs, &event.block.transactions);
+    observe_channel_inscriptions(s, &channel_txs, &block.transactions);
 
     // Process the actual event block
-    s.process_block(block_id, parent_id, lib, our_txs, channel_txs);
+    s.process_block(
+        block.header.id,
+        block.header.parent_block,
+        lib,
+        our_txs,
+        channel_txs,
+    );
 
     // Remove our pending txs that were finalized in the backfilled LIB blocks.
     // `finalized_items` already carries the typed payloads (built before
     // pending was mutated) so we just need to clean up state here.
-    for tx_hash in &lib_finalized {
+    for tx_hash in &finalized_batch.our_tx_hashes {
         s.remove_pending(tx_hash);
     }
 
@@ -194,11 +261,11 @@ where
         update
     });
 
-    Ok(BlockEventResult {
-        finalized_items,
+    BlockEventResult {
+        finalized_items: finalized_batch.items,
         channel_update,
         mined_inscriptions,
-    })
+    }
 }
 
 /// Mirror a block's channel inscriptions into the pending set
@@ -296,31 +363,23 @@ pub(super) struct FetchedBatch {
     pub(super) items: Vec<FinalizedTx>,
 }
 
-/// Fetch blocks in a slot range, process them into state, and return our
-/// finalized tx hashes plus the user-facing items grouped per Mantle tx.
-///
-/// State is mutated only after the per-block fetch (blocks + events) has
-/// fully succeeded. On any failure the function returns [`Err`] without
-/// having advanced `state` for the failing block (earlier blocks in the
-/// range are kept — they were independent successful units of work). The
-/// caller is expected to abandon the current attempt and retry the range
-/// later; the partial advance ensures progress on transient errors that
-/// resolve mid-range.
-pub(super) async fn fetch_and_process_blocks<Node>(
-    state: &mut TxState,
+struct PreparedFinalizedBlock {
+    block_id: HeaderId,
+    parent_id: HeaderId,
+    our_txs: Vec<TxHash>,
+    channel_txs: Vec<BlockChannelTx>,
+    items: Vec<FinalizedTx>,
+}
+
+async fn prepare_finalized_blocks<Node>(
     from_slot: u64,
     to_slot: u64,
     channel_id: ChannelId,
     node: &Node,
-) -> Result<FetchedBatch, Error>
+) -> Result<Vec<PreparedFinalizedBlock>, Error>
 where
     Node: adapter::Node + Sync,
 {
-    let mut result = FetchedBatch {
-        our_tx_hashes: Vec::new(),
-        items: Vec::new(),
-    };
-
     let blocks = node
         .immutable_blocks(Slot::from(from_slot), Slot::from(to_slot))
         .await
@@ -331,6 +390,7 @@ where
             ))
         })?;
 
+    let mut prepared = Vec::with_capacity(blocks.len());
     for block in blocks {
         let our_txs: Vec<TxHash> = block
             .transactions
@@ -353,20 +413,59 @@ where
             &deposit_events,
         );
 
-        result.our_tx_hashes.extend(our_txs.iter().copied());
-        result.items.extend(block_items);
-
-        let current_lib = state.lib();
-        state.process_block(
-            block.header.id,
-            block.header.parent_block,
-            current_lib,
+        prepared.push(PreparedFinalizedBlock {
+            block_id: block.header.id,
+            parent_id: block.header.parent_block,
             our_txs,
             channel_txs,
+            items: block_items,
+        });
+    }
+
+    Ok(prepared)
+}
+
+fn apply_finalized_blocks(
+    state: &mut TxState,
+    blocks: Vec<PreparedFinalizedBlock>,
+) -> FetchedBatch {
+    let mut result = FetchedBatch {
+        our_tx_hashes: Vec::new(),
+        items: Vec::new(),
+    };
+
+    for block in blocks {
+        result.our_tx_hashes.extend(block.our_txs.iter().copied());
+        result.items.extend(block.items);
+
+        state.process_block(
+            block.block_id,
+            block.parent_id,
+            state.lib(),
+            block.our_txs,
+            block.channel_txs,
         );
     }
 
-    Ok(result)
+    result
+}
+
+/// Fetch a finalized slot range, then apply it without another suspension
+/// point. Dropping the caller's future during any node request leaves
+/// `state` untouched, so retry starts from the same boundary.
+pub(super) async fn fetch_and_process_blocks<Node>(
+    state: &mut TxState,
+    from_slot: u64,
+    to_slot: u64,
+    channel_id: ChannelId,
+    node: &Node,
+) -> Result<FetchedBatch, Error>
+where
+    Node: adapter::Node + Sync,
+{
+    let prepared = prepare_finalized_blocks(from_slot, to_slot, channel_id, node).await?;
+
+    Ok(apply_finalized_blocks(state, prepared))
 }
 
 /// Fetch the deposit-amount lookup for a single block, gated on whether the
@@ -545,71 +644,69 @@ fn extract_finalized_items(
     items
 }
 
-/// Backfill canonical chain backwards from a missing parent to LIB.
-///
-/// Uses `state.lib()` during replay to avoid premature finalization.
-/// The caller is responsible for triggering finalization after backfill
-/// completes.
-async fn backfill_canonical<Node>(
-    state: &mut TxState,
-    missing_parent: HeaderId,
-    channel_id: ChannelId,
-    node: &Node,
-) where
-    Node: adapter::Node + Sync,
-{
-    debug!(target: TARGET, "Backfilling canonical chain from {:?}", missing_parent);
-    let blocks = walk_back_to_known(state, missing_parent, node).await;
-    let lib = state.lib();
-    for block in &blocks {
-        apply_backfilled_block(state, block, channel_id, lib);
-    }
-    debug!(target: TARGET, "Canonical backfill complete");
+/// Walk backwards from `from` until reaching a block already present in the
+/// current state or the finalized batch prepared for this event. Returns
+/// blocks in forward order (oldest first) without mutating state.
+fn block_is_known(
+    state: Option<&TxState>,
+    additionally_known: &HashSet<HeaderId>,
+    lib: HeaderId,
+    block: HeaderId,
+) -> bool {
+    block == lib
+        || additionally_known.contains(&block)
+        || state.is_some_and(|state| state.has_block(&block))
 }
 
-/// Walk backwards from `from` until a block the state already knows about (or
-/// LIB) is reached. Returns blocks in forward order (oldest first).
 async fn walk_back_to_known<Node>(
-    state: &TxState,
+    state: Option<&TxState>,
+    additionally_known: &HashSet<HeaderId>,
+    lib: HeaderId,
     from: HeaderId,
     node: &Node,
-) -> Vec<lb_common_http_client::ApiBlock>
+) -> Vec<ApiBlock>
 where
     Node: adapter::Node + Sync,
 {
+    debug!(target: TARGET, "Backfilling canonical chain from {from:?}");
+
     let mut blocks = Vec::new();
     let mut current = from;
-    let lib = state.lib();
 
-    while !state.has_block(&current) && current != lib {
-        match node.block(current).await {
-            Ok(Some(block)) => {
-                let parent = block.header.parent_block;
-                blocks.push(block);
-                current = parent;
-            }
-            Ok(None) => {
-                warn!(target: TARGET, "Block {:?} not found during canonical backfill", current);
-                break;
-            }
-            Err(e) => {
-                warn!(
-                    target: TARGET,
-                    "Failed to fetch block {:?} during canonical backfill: {e}",
-                    current
-                );
-                break;
-            }
-        }
+    while !block_is_known(state, additionally_known, lib, current) {
+        let Some(block) = fetch_backfill_block(node, current).await else {
+            break;
+        };
+
+        current = block.header.parent_block;
+        blocks.push(block);
     }
 
     blocks.reverse();
+    debug!(target: TARGET, blocks = blocks.len(), "Canonical backfill prepared");
     blocks
 }
 
+async fn fetch_backfill_block<Node>(node: &Node, block_id: HeaderId) -> Option<ApiBlock>
+where
+    Node: adapter::Node + Sync,
+{
+    match node.block(block_id).await {
+        Ok(Some(block)) => Some(block),
+        Ok(None) => {
+            warn!(target: TARGET, ?block_id, "Block not found during canonical backfill");
+            None
+        }
+        Err(error) => {
+            warn!(target: TARGET, ?block_id, %error, "Failed to fetch block during canonical backfill");
+            None
+        }
+    }
+}
+
 fn apply_backfilled_block(
     state: &mut TxState,
-    block: &lb_common_http_client::ApiBlock,
+    block: &ApiBlock,
     channel_id: ChannelId,
     lib: HeaderId,
 ) {
@@ -1258,8 +1355,7 @@ mod tests {
 
         // shed_off_branch_pending should drop the stale one but not the live one.
         let shed = state.shed_off_branch_pending(block);
-        let shed_hashes: std::collections::HashSet<TxHash> =
-            shed.iter().map(PendingTx::tx_hash).collect();
+        let shed_hashes: HashSet<TxHash> = shed.iter().map(PendingTx::tx_hash).collect();
         assert!(
             shed_hashes.contains(&pending_stale_hash),
             "pending chained from I1 must be shed (no longer reachable from tip)"
@@ -1362,8 +1458,7 @@ mod tests {
 
         // Our pending's parent matches the new tip, so it should stay.
         let shed = state.shed_off_branch_pending(block_b);
-        let shed_hashes: std::collections::HashSet<TxHash> =
-            shed.iter().map(PendingTx::tx_hash).collect();
+        let shed_hashes: HashSet<TxHash> = shed.iter().map(PendingTx::tx_hash).collect();
         assert!(
             !shed_hashes.contains(&pending_hash),
             "pending chained from the (still-current) config tip must remain on-branch"

zone-sdk/src/sequencer/zone_sequencer.rs

@@ -73,6 +73,12 @@ pub struct ZoneSequencer<Node> {
     // Block stream
     pub(super) blocks_stream: Option<BoxStream<ProcessedBlockEvent>>,
 
+    // A block pulled from the stream but not yet converted into a public
+    // event. Keeping it on the sequencer makes `next_event` cancellation-safe:
+    // if its future is dropped while block processing awaits node data, the
+    // next call resumes the same block instead of pulling past it.
+    pub(super) pending_block_event: Option<Arc<ProcessedBlockEvent>>,
+
     // True while the blocks stream is alive AND we have processed at least
     // one event since (re)connecting — i.e. cached `channel_state` and
     // `current_tip` reflect the latest block we observed. Cleared on stream
@@ -273,6 +279,7 @@ where
             channel_state: None,
             own_key_index: None,
             blocks_stream: None,
+            pending_block_event: None,
             connected: false,
             resubmit_interval,
             in_flight: FuturesUnordered::new(),
@@ -421,6 +428,17 @@ where
     /// (in-progress backfill batches, periodic resubmit ticks, in-flight post
     /// completions, reconnect retries), so the caller's loop body always
     /// receives a real [`Event`] — no `Option` unwrapping required.
+    ///
+    /// # Block-event cancellation safety
+    ///
+    /// Cancelling this future does not lose or partially apply a block event.
+    /// A pulled block is retained until its event is returned, and all fallible
+    /// node reads complete before the corresponding state mutation.
+    ///
+    /// A [`SequencerClient`](super::SequencerClient) command selected from the
+    /// request queue may instead fail with [`Error::Unavailable`] if this
+    /// future is cancelled while handling it. The client can retry that
+    /// command.
     pub async fn next_event(&mut self) -> Event {
         loop {
             if let Some(ev) = self.step().await {
@@ -440,6 +458,13 @@ where
             return Some(self.emit_now(event));
         }
 
+        // Finish a block already pulled from the stream before doing any
+        // other drive work. `process_pending_block_event` only clears it once
+        // processing has produced the corresponding public event.
+        if self.pending_block_event.is_some() {
+            return self.process_pending_block_event().await;
+        }
+
         // Process incremental backfill — one batch per call.
         // Returns Some(Some(event)) or Some(None) while active, None when done.
         if let Some(maybe_event) = self.process_incremental_backfill().await {
@@ -455,9 +480,13 @@ where
 
         tokio::select! {
             maybe_event = stream.next() => {
-                self.handle_stream_item(maybe_event)
-                    .await
-                    .map(|event| self.emit_now(event))
+                let Some(block_event) = maybe_event else {
+                    self.handle_stream_disconnect();
+                    return None;
+                };
+
+                self.pending_block_event = Some(Arc::new(block_event));
+                None
             }
             _ = self.resubmit_interval.tick(), if self.current_tip.is_some() => {
                 self.resubmit_pending();

zone-sdk/src/test_support.rs

@@ -66,6 +66,10 @@ pub fn scripts(scripts: Vec<StreamScript>) -> Arc<Mutex<VecDeque<StreamScript>>>
 pub struct MockNode {
     /// Served by `channel_state()`.
     pub channel_state: Option<ChannelState>,
+    /// Optional gate for pausing `channel_state()` calls in cancellation tests.
+    pub channel_state_gate: Option<watch::Receiver<bool>>,
+    /// Optional notification sent whenever `channel_state()` is called.
+    pub channel_state_calls: Option<mpsc::UnboundedSender<()>>,
     /// LIB and tip ids reported by `consensus_info()`.
     pub lib: HeaderId,
     pub tip: HeaderId,
@@ -78,6 +82,11 @@ pub struct MockNode {
     pub blocks: Vec<ApiBlock>,
     /// Served by `immutable_blocks()`, filtered by the queried slot range.
     pub immutable: Vec<ApiBlock>,
+    /// Optional gate for pausing `immutable_blocks()` calls in cancellation
+    /// tests.
+    pub immutable_blocks_gate: Option<watch::Receiver<bool>>,
+    /// Optional notification sent whenever `immutable_blocks()` is called.
+    pub immutable_blocks_calls: Option<mpsc::UnboundedSender<()>>,
     /// Served by `zone_messages_in_blocks()`, filtered by the queried slot
     /// range.
     pub zone_messages: Vec<(ZoneMessage, Slot)>,
@@ -94,6 +103,8 @@ impl Default for MockNode {
     fn default() -> Self {
         Self {
             channel_state: Some(single_key_channel_state()),
+            channel_state_gate: None,
+            channel_state_calls: None,
             lib: header_id(0),
             tip: header_id(0),
             lib_slot: Slot::genesis(),
@@ -103,6 +114,8 @@ impl Default for MockNode {
             }]),
             blocks: Vec::new(),
             immutable: Vec::new(),
+            immutable_blocks_gate: None,
+            immutable_blocks_calls: None,
             zone_messages: Vec::new(),
             up: None,
             posted: None,
@@ -166,6 +179,21 @@ impl adapter::Node for MockNode {
         &self,
         _channel_id: ChannelId,
     ) -> Result<Option<ChannelState>, lb_common_http_client::Error> {
+        if let Some(calls) = &self.channel_state_calls {
+            let _ = calls.send(());
+        }
+
+        if let Some(gate) = &self.channel_state_gate {
+            let mut gate = gate.clone();
+            while !*gate.borrow_and_update() {
+                gate.changed().await.map_err(|_| {
+                    lb_common_http_client::Error::Client(
+                        "channel-state test gate closed".to_owned(),
+                    )
+                })?;
+            }
+        }
+
         Ok(self.channel_state.clone())
     }
 
@@ -220,6 +248,21 @@ impl adapter::Node for MockNode {
         slot_from: Slot,
         slot_to: Slot,
     ) -> Result<Vec<ApiBlock>, lb_common_http_client::Error> {
+        if let Some(calls) = &self.immutable_blocks_calls {
+            let _ = calls.send(());
+        }
+
+        if let Some(gate) = &self.immutable_blocks_gate {
+            let mut gate = gate.clone();
+            while !*gate.borrow_and_update() {
+                gate.changed().await.map_err(|_| {
+                    lb_common_http_client::Error::Client(
+                        "immutable-blocks test gate closed".to_owned(),
+                    )
+                })?;
+            }
+        }
+
         Ok(self
             .immutable
             .iter()