diff --git a/zone-sdk/src/sequencer/actor.rs b/zone-sdk/src/sequencer/actor.rs index 6567896b2..dd2a33435 100644 --- a/zone-sdk/src/sequencer/actor.rs +++ b/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 ZoneSequencer 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, - ) -> Option { - 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 { + 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 { + // 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) { 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, + channel_id: ChannelId, +) -> Result, 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 diff --git a/zone-sdk/src/sequencer/block_fetch.rs b/zone-sdk/src/sequencer/block_fetch.rs index 1c2b0ca68..72f973833 100644 --- a/zone-sdk/src/sequencer/block_fetch.rs +++ b/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, } +struct PreparedBlockEvent<'a> { + block: &'a ApiBlock, + tip: HeaderId, + lib: HeaderId, + lib_slot: Slot, + lib_advanced: bool, + finalized: Vec, + canonical_backfill: Vec, + our_txs: Vec, + channel_txs: Vec, + mined_inscriptions: Vec, +} + /// Process a block event. Returns finalized tx hashes and optional channel /// update. /// @@ -64,23 +77,103 @@ pub(super) async fn handle_block_event( 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, 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 = + 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 = 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, + current_tip: &mut Option, + 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 = 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 = 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 = 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, } -/// 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( - state: &mut TxState, +struct PreparedFinalizedBlock { + block_id: HeaderId, + parent_id: HeaderId, + our_txs: Vec, + channel_txs: Vec, + items: Vec, +} + +async fn prepare_finalized_blocks( from_slot: u64, to_slot: u64, channel_id: ChannelId, node: &Node, -) -> Result +) -> Result, 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 = 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, +) -> 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( + state: &mut TxState, + from_slot: u64, + to_slot: u64, + channel_id: ChannelId, + node: &Node, +) -> Result +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( - 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, + 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( - state: &TxState, + state: Option<&TxState>, + additionally_known: &HashSet, + lib: HeaderId, from: HeaderId, node: &Node, -) -> Vec +) -> Vec 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(¤t) && 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, block_id: HeaderId) -> Option +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 = - shed.iter().map(PendingTx::tx_hash).collect(); + let shed_hashes: HashSet = 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 = - shed.iter().map(PendingTx::tx_hash).collect(); + let shed_hashes: HashSet = 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" diff --git a/zone-sdk/src/sequencer/zone_sequencer.rs b/zone-sdk/src/sequencer/zone_sequencer.rs index 16f7f5fa1..a06921d68 100644 --- a/zone-sdk/src/sequencer/zone_sequencer.rs +++ b/zone-sdk/src/sequencer/zone_sequencer.rs @@ -73,6 +73,12 @@ pub struct ZoneSequencer { // Block stream pub(super) blocks_stream: Option>, + // 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>, + // 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(); diff --git a/zone-sdk/src/test_support.rs b/zone-sdk/src/test_support.rs index b20d01a8e..34f5ccb45 100644 --- a/zone-sdk/src/test_support.rs +++ b/zone-sdk/src/test_support.rs @@ -66,6 +66,10 @@ pub fn scripts(scripts: Vec) -> Arc>> pub struct MockNode { /// Served by `channel_state()`. pub channel_state: Option, + /// Optional gate for pausing `channel_state()` calls in cancellation tests. + pub channel_state_gate: Option>, + /// Optional notification sent whenever `channel_state()` is called. + pub channel_state_calls: Option>, /// LIB and tip ids reported by `consensus_info()`. pub lib: HeaderId, pub tip: HeaderId, @@ -78,6 +82,11 @@ pub struct MockNode { pub blocks: Vec, /// Served by `immutable_blocks()`, filtered by the queried slot range. pub immutable: Vec, + /// Optional gate for pausing `immutable_blocks()` calls in cancellation + /// tests. + pub immutable_blocks_gate: Option>, + /// Optional notification sent whenever `immutable_blocks()` is called. + pub immutable_blocks_calls: Option>, /// 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, 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, 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()