diff --git a/Cargo.lock b/Cargo.lock index 7d5f633a..dcda1132 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -9307,6 +9307,7 @@ dependencies = [ "logos-blockchain-zone-sdk", "mempool", "num-bigint 0.4.6", + "ping_core", "programs", "rand 0.8.6", "risc0-zkvm", diff --git a/integration_tests/tests/cross_zone_watcher_restart.rs b/integration_tests/tests/cross_zone_watcher_restart.rs new file mode 100644 index 00000000..155a29fc --- /dev/null +++ b/integration_tests/tests/cross_zone_watcher_restart.rs @@ -0,0 +1,227 @@ +#![expect( + clippy::tests_outside_test_module, + reason = "top-level test functions are conventional for integration tests" +)] + +//! A sequencer restart must resume its cross-zone watcher from the persisted +//! per-peer delivery floor instead of re-reading the peer channel from genesis. +//! +//! Re-reading is safe (the dispatch key is content-addressed and the inbox +//! no-ops a replay) so on-chain state cannot tell the two apart. What does tell +//! them apart is the transactions: a watcher that lost its cursor re-injects +//! every already-delivered dispatch, which shows up as inbox transactions in +//! blocks produced after the restart. This is the only test that covers the +//! wiring from `spawn_watchers` through the store, so a silent regression here +//! would leave every other test green while the feature does nothing. + +use std::time::Duration; + +use anyhow::{Context as _, Result}; +use common::transaction::LeeTransaction; +use cross_zone_outbox_core::outbox_pda; +use integration_tests::{ + config::{self, SequencerPartialConfig}, + setup::{SequencerSetup, sequencer_client, setup_bedrock_node}, +}; +use lee::{AccountId, PublicTransaction, public_transaction::Message}; +use lee_core::program::ProgramId; +use ping_core::{ReceiverInstruction, SenderInstruction, ping_record_pda}; +use sequencer_core::config::{CrossZoneConfig, CrossZonePeer}; +use sequencer_service_rpc::{RpcClient as _, SequencerClient}; +use tokio::test; + +const DELIVERY_TIMEOUT: Duration = Duration::from_secs(480); +/// Blocks zone B must produce after the restart before we judge the watcher. +/// A watcher that lost its cursor re-reads the peer channel on its first pass, +/// so a handful of blocks is ample room for the replay to appear. +const BLOCKS_AFTER_RESTART: u64 = 5; +const RESTART_TIMEOUT: Duration = Duration::from_secs(240); +const PING_PAYLOAD: &[u8] = b"hello-cross-zone"; + +#[test] +async fn restarted_watcher_resumes_instead_of_replaying_the_peer_channel() -> Result<()> { + // Declared first so it outlives both zones (drops run in reverse order). + let (_bedrock, bedrock_addr) = setup_bedrock_node() + .await + .context("Failed to set up shared Bedrock node")?; + + let partial = SequencerPartialConfig::default(); + let channel_a = config::bedrock_channel_id(); + let channel_b = config::bedrock_channel_id_b(); + let zone_a: [u8; 32] = *channel_a.as_ref(); + let zone_b: [u8; 32] = *channel_b.as_ref(); + let receiver_id = programs::ping_receiver().id(); + + let cross_zone = CrossZoneConfig { + peers: vec![CrossZonePeer { + channel_id: zone_a, + allowed_targets: vec![receiver_id], + expected_block_signing_pubkey: None, + }], + }; + + let (seq_a, _seq_a_home) = SequencerSetup::new(partial, bedrock_addr) + .with_channel_id(channel_a) + .with_genesis(vec![]) + .setup() + .await + .context("Failed to set up zone A sequencer")?; + + // Zone B keeps an explicit home so it can be restarted on the same store. + let home_b = tempfile::tempdir().context("Failed to create zone B home")?; + let mut seq_b = SequencerSetup::new(partial, bedrock_addr) + .with_channel_id(channel_b) + .with_genesis(vec![]) + .with_cross_zone(cross_zone.clone()) + .setup_at(home_b.path()) + .await + .context("Failed to set up zone B sequencer")?; + + // Deliver one ping, so the peer channel holds a dispatch worth replaying. + sequencer_client(seq_a.addr())? + .send_transaction(build_ping_tx(zone_b, receiver_id)) + .await + .context("Failed to submit ping on zone A")?; + let record_id = ping_record_pda(receiver_id); + let delivered = wait_for_delivery(sequencer_client(seq_b.addr())?, record_id).await?; + assert_eq!( + delivered, PING_PAYLOAD, + "Zone B must record the payload before the restart" + ); + + let tip_before = sequencer_client(seq_b.addr())?.get_last_block_id().await?; + + // Restart zone B on the same home. Zone A stays quiet from here, so any + // inbox transaction after the restart is a replay, not a new delivery. + // + // `shutdown` rather than `drop`: dropping aborts the main loop without + // awaiting it and leaves the watchers and the publisher's drive task holding + // the store, so the reopen below would race the `RocksDB` lock. + seq_b.shutdown().await; + seq_b = SequencerSetup::new(partial, bedrock_addr) + .with_channel_id(channel_b) + .with_genesis(vec![]) + .with_cross_zone(cross_zone) + .setup_at(home_b.path()) + .await + .context("Failed to restart zone B sequencer")?; + let client_b = sequencer_client(seq_b.addr())?; + + let tip_after = wait_for_block_id( + &client_b, + tip_before.saturating_add(BLOCKS_AFTER_RESTART), + RESTART_TIMEOUT, + ) + .await?; + + let replayed = count_inbox_transactions(&client_b, tip_before.saturating_add(1), tip_after) + .await + .context("Failed to scan zone B blocks after the restart")?; + assert_eq!( + replayed, + 0, + "a restarted watcher must resume from its persisted delivery floor; found {replayed} inbox transaction(s) in blocks {}..={tip_after}, which means it re-read the peer channel from genesis", + tip_before.saturating_add(1) + ); + + // The delivery itself must survive the restart untouched. + let account = client_b.get_account(record_id).await?; + assert_eq!( + account.data.into_inner(), + PING_PAYLOAD, + "the delivered payload must survive the restart" + ); + Ok(()) +} + +/// Counts inbox transactions across `from..=to`, the signature of a re-injected +/// dispatch. +async fn count_inbox_transactions(client: &SequencerClient, from: u64, to: u64) -> Result { + let inbox_id = programs::cross_zone_inbox().id(); + let mut count = 0_usize; + for block_id in from..=to { + let Some(block) = client.get_block(block_id).await? else { + continue; + }; + for tx in &block.body.transactions { + if let LeeTransaction::Public(public_tx) = tx + && public_tx.message().program_id == inbox_id + { + count = count.saturating_add(1); + } + } + } + Ok(count) +} + +/// Waits until the sequencer's tip reaches `target`, returning the tip. +async fn wait_for_block_id( + client: &SequencerClient, + target: u64, + timeout: Duration, +) -> Result { + let wait = async { + loop { + let tip = client.get_last_block_id().await?; + if tip >= target { + return Ok::(tip); + } + tokio::time::sleep(Duration::from_secs(2)).await; + } + }; + tokio::time::timeout(timeout, wait) + .await + .context("Zone B did not produce enough blocks after the restart")? +} + +/// Builds a top-level `ping_sender` transaction that chains into the outbox to emit +/// a message carrying a `ping_receiver::Record` instruction for the target zone. +fn build_ping_tx(target_zone: [u8; 32], receiver_id: ProgramId) -> LeeTransaction { + let outbox_id = programs::cross_zone_outbox().id(); + let ordinal = 0; + + let words = risc0_zkvm::serde::to_vec(&ReceiverInstruction::Record { + payload: PING_PAYLOAD.to_vec(), + }) + .expect("serialize ping instruction"); + let payload: Vec = words.iter().flat_map(|word| word.to_le_bytes()).collect(); + + let send = SenderInstruction::Send { + outbox_program_id: outbox_id, + target_zone, + target_program_id: receiver_id, + target_accounts: vec![ping_record_pda(receiver_id).into_value()], + payload, + ordinal, + }; + + let outbox_account = outbox_pda(outbox_id, &target_zone, ordinal); + let message = Message::try_new( + programs::ping_sender().id(), + vec![outbox_account], + vec![], + send, + ) + .expect("build ping message"); + LeeTransaction::Public(PublicTransaction::new( + message, + lee::public_transaction::WitnessSet::from_raw_parts(vec![]), + )) +} + +/// Polls zone B's sequencer until the ping record PDA holds a payload. +async fn wait_for_delivery(client: SequencerClient, record_id: AccountId) -> Result> { + let wait = async { + loop { + let account = client.get_account(record_id).await?; + let data = account.data.into_inner(); + if !data.is_empty() { + return Ok::, anyhow::Error>(data); + } + tokio::time::sleep(Duration::from_secs(3)).await; + } + }; + tokio::time::timeout(DELIVERY_TIMEOUT, wait) + .await + .context("Zone B did not record the cross-zone payload in time")? +} diff --git a/lez/sequencer/core/Cargo.toml b/lez/sequencer/core/Cargo.toml index c8590c2a..f23a11a0 100644 --- a/lez/sequencer/core/Cargo.toml +++ b/lez/sequencer/core/Cargo.toml @@ -57,3 +57,4 @@ test_programs.workspace = true lee = { workspace = true, features = ["test-utils"] } key_protocol.workspace = true token_core.workspace = true +ping_core.workspace = true diff --git a/lez/sequencer/core/src/block_publisher.rs b/lez/sequencer/core/src/block_publisher.rs index 86d9dc0c..e2b8610e 100644 --- a/lez/sequencer/core/src/block_publisher.rs +++ b/lez/sequencer/core/src/block_publisher.rs @@ -1,4 +1,4 @@ -use std::{sync::Arc, time::Duration}; +use std::time::Duration; use anyhow::{Context as _, Result, anyhow, ensure}; use common::block::Block; @@ -34,13 +34,10 @@ use logos_blockchain_zone_sdk::{ ZoneSequencer, }, }; -use tokio::{ - sync::{mpsc, oneshot, watch}, - task::JoinHandle, -}; +use tokio::sync::{mpsc, oneshot, watch}; use tokio_util::sync::CancellationToken; -use crate::config::BedrockConfig; +use crate::{config::BedrockConfig, task_group::TaskGroup}; /// Channel capacity for the publish inbox. One publish per produced block, drained /// in microseconds by the drive task — 32 is huge headroom and just provides @@ -120,6 +117,14 @@ pub trait BlockPublisherTrait: Sized { /// are processed past that point, so the node must halt. fn driver_cancellation(&self) -> CancellationToken; + /// The publisher's background tasks, for a caller that needs to know when + /// they have actually stopped. Its sinks capture a store handle, so the + /// `RocksDB` lock outlives the sequencer until the drive task is gone. + /// Empty by default, for publishers that run no tasks. + fn background_tasks(&self) -> TaskGroup { + TaskGroup::default() + } + /// Current channel frontier slot on the connected chain, or `None` if the /// channel does not exist there. Drives the startup frontier check. async fn channel_tip_slot(&self) -> Result>; @@ -143,19 +148,12 @@ pub struct ZoneSdkPublisher { turn_rx: watch::Receiver, // Cancelled when the drive task ends for any reason, including a panic. driver_cancellation: CancellationToken, - // Aborts the drive task when the last clone is dropped. - _drive_task: Arc, + // Stops the drive task when the last clone is dropped, and lets a shutdown + // path wait until it has actually stopped. + drive_task: TaskGroup, indexer: ZoneIndexer, } -struct DriveTaskGuard(JoinHandle<()>); - -impl Drop for DriveTaskGuard { - fn drop(&mut self) { - self.0.abort(); - } -} - impl BlockPublisherTrait for ZoneSdkPublisher { async fn new( config: &BedrockConfig, @@ -318,7 +316,7 @@ impl BlockPublisherTrait for ZoneSdkPublisher { command_tx, turn_rx, driver_cancellation, - _drive_task: Arc::new(DriveTaskGuard(drive_task)), + drive_task: TaskGroup::new(vec![drive_task]), }) } @@ -359,6 +357,10 @@ impl BlockPublisherTrait for ZoneSdkPublisher { self.driver_cancellation.clone() } + fn background_tasks(&self) -> TaskGroup { + self.drive_task.clone() + } + async fn channel_tip_slot(&self) -> Result> { Ok(self .node diff --git a/lez/sequencer/core/src/block_store.rs b/lez/sequencer/core/src/block_store.rs index 57c94990..df61b38b 100644 --- a/lez/sequencer/core/src/block_store.rs +++ b/lez/sequencer/core/src/block_store.rs @@ -9,10 +9,12 @@ use common::{ use lee::V03State; use lee_core::BlockId; use log::info; -use logos_blockchain_zone_sdk::sequencer::SequencerCheckpoint; +use logos_blockchain_zone_sdk::{Slot, sequencer::SequencerCheckpoint}; use storage::sequencer::{ RocksDBIO, - sequencer_cells::{PendingDepositEventRecord, WithdrawalReconciliationKey, ZoneAnchorRecord}, + sequencer_cells::{ + PeerZoneKey, PendingDepositEventRecord, WithdrawalReconciliationKey, ZoneAnchorRecord, + }, }; pub use storage::{DbResult, sequencer::DbDump}; @@ -233,6 +235,36 @@ pub(crate) fn block_to_transactions_map(block: &Block) -> HashMap .collect() } +/// A cross-zone watcher's delivery floor on `peer_zone`'s channel. +/// +/// The highest slot every message of which was delivered, or `None` before it +/// has delivered anything from that peer. Stored as a little-endian `u64`. +/// +/// Free functions rather than only [`SequencerStore`] methods because each +/// watcher runs as its own spawned task and holds an `Arc`; +/// `SequencerStore` is not `Clone`. +pub fn get_cross_zone_peer_floor(dbio: &RocksDBIO, peer_zone: PeerZoneKey) -> Result> { + let Some(bytes) = dbio.get_cross_zone_peer_floor_bytes(peer_zone)? else { + return Ok(None); + }; + let bytes: [u8; 8] = bytes.as_slice().try_into().with_context(|| { + format!( + "Stored cross-zone peer floor is {} bytes, expected 8", + bytes.len() + ) + })?; + Ok(Some(Slot::new(u64::from_le_bytes(bytes)))) +} + +pub fn set_cross_zone_peer_floor( + dbio: &RocksDBIO, + peer_zone: PeerZoneKey, + floor: Slot, +) -> Result<()> { + dbio.put_cross_zone_peer_floor_bytes(peer_zone, &floor.to_le_bytes())?; + Ok(()) +} + #[cfg(test)] mod tests { use common::{block::HashableBlockData, test_utils::sequencer_sign_key_for_testing}; diff --git a/lez/sequencer/core/src/cross_zone_watcher.rs b/lez/sequencer/core/src/cross_zone_watcher.rs index ffa238c4..e3ca8ba3 100644 --- a/lez/sequencer/core/src/cross_zone_watcher.rs +++ b/lez/sequencer/core/src/cross_zone_watcher.rs @@ -1,34 +1,207 @@ -use std::time::Duration; +use std::{sync::Arc, time::Duration}; use common::{block::Block, transaction::LeeTransaction}; use cross_zone::{build_dispatch_from_emission, extract_emission}; -use futures::StreamExt as _; +use cross_zone_inbox_core::message_key; +use futures::{Stream, StreamExt as _}; use lee::PublicKey; use lee_core::program::ProgramId; use log::{debug, error, info, warn}; use logos_blockchain_core::mantle::ops::channel::ChannelId; use logos_blockchain_zone_sdk::{ - CommonHttpClient, ZoneMessage, adapter::NodeHttpClient, indexer::ZoneIndexer, + CommonHttpClient, Slot, ZoneMessage, adapter::NodeHttpClient, indexer::ZoneIndexer, }; -use mempool::MemPoolHandle; +use storage::sequencer::{RocksDBIO, sequencer_cells::PendingCrossZoneDispatchRecord}; use crate::{ - TransactionOrigin, + block_store::{get_cross_zone_peer_floor, set_cross_zone_peer_floor}, config::{BedrockConfig, CrossZoneConfig}, + task_group::TaskGroup, }; +/// Consecutive passes a watcher re-reads the same undecodable slot before giving +/// up and reading past it. +/// +/// One pass per poll interval, which is the block time, so this is minutes of +/// retrying rather than seconds. A transient failure (a truncated read, a peer +/// mid-upgrade) heals well inside that; a block this node genuinely cannot +/// decode does not heal at all, and waiting longer only delays every later +/// message behind it. +const DECODE_RETRY_LIMIT: u32 = 20; + +/// The per-peer settings one watcher pass needs. +struct PeerContext { + peer_zone: [u8; 32], + self_zone: [u8; 32], + allowed_targets: Vec, + expected_pubkey: Option, +} + +/// What a pass may do about a slot the watcher cannot decode, and whether it may +/// still move the durable delivery floor. +/// +/// The two are one decision, not two. Past a skipped slot everything is +/// delivered on top of a gap, and persisting past that gap would make the skip +/// survive restarts, so the floor has to stop moving and stay stopped. Holding +/// them in one value is what makes "skipping while still persisting", which +/// would quietly restore that bug, unrepresentable. +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +enum SkipPolicy { + /// Nothing has been given up on: deliver everything and move the floor. + #[default] + DeliverAll, + /// Read past this slot, and stop moving the floor. + Skipping(Slot), + /// A slot was skipped earlier in this run. Nothing is being skipped now, but + /// everything read from here sits above the gap, so the floor stays put. + FloorFrozen, +} + +/// Why one pass over a peer's stream ended. +/// +/// A pass that gave up inside a slot says which kind of failure did it. Only a +/// block this node cannot decode is a reason to eventually read past a slot; +/// a delivery that could not be recorded or handed off is our own problem, and +/// counting it towards the decode budget would read past a slot that is fine. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +enum PassOutcome { + /// The stream drained. + Drained, + /// Ended inside this slot: its block would not deserialize. + Undecodable(Slot), + /// Ended inside this slot: a delivery could not be recorded or enqueued. + Undelivered(Slot), +} + +/// The pass-to-pass state of one watcher: what it is stuck on, and what it is +/// allowed to do about it. +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +struct WatcherState { + /// The slot the watcher is stuck on and how many passes it has spent there. + /// Keyed by slot so a failure at a new slot does not inherit an older + /// slot's count. + stalled: Option<(Slot, u32)>, + skip: SkipPolicy, +} + +impl SkipPolicy { + /// The slot this pass reads past rather than stalling on. + const fn skip_slot(self) -> Option { + match self { + Self::Skipping(slot) => Some(slot), + Self::DeliverAll | Self::FloorFrozen => None, + } + } + + /// Whether this pass may still move the durable delivery floor. + const fn persists_floor(self) -> bool { + matches!(self, Self::DeliverAll) + } + + /// The policy once a pass has read past whatever it was stuck on. + /// + /// Nothing is being skipped any more, but a run that has skipped once keeps + /// its floor frozen: everything from here sits above the gap, and moving the + /// floor over it would make the skip survive a restart. Deliberately not + /// named for clearing: it downgrades, it does not reset. + const fn after_clean_pass(self) -> Self { + match self { + Self::DeliverAll => Self::DeliverAll, + Self::Skipping(_) | Self::FloorFrozen => Self::FloorFrozen, + } + } + + /// Whether a pass that ended at `cursor` actually got past the slot this + /// policy is skipping. + /// + /// A stream can end without reaching it: the zone-sdk ends a stream on a + /// fetch failure exactly as on catching up. Downgrading on such a pass would + /// disarm the skip before it was ever used, and the slot would have to be + /// given up on again from scratch, so a peer endpoint that is flaky around + /// one bad slot would never be read past. + fn used_its_skip(self, cursor: Option) -> bool { + match self { + Self::Skipping(slot) => cursor.is_some_and(|read_to| read_to >= slot), + Self::DeliverAll | Self::FloorFrozen => true, + } + } +} + +impl WatcherState { + /// Folds one pass's outcome in, returning a slot the watcher has just given + /// up on so the caller can report it. + /// + /// `cursor` is the read position after the pass. It is what tells a stream + /// that truncated early apart from one that genuinely drained: the zone-sdk + /// ends a stream on a fetch failure exactly as it does on catching up, so + /// without this a flaky peer endpoint would reset the retry count for ever + /// and the watcher would never escape a slot it cannot decode. + fn after_pass(&mut self, outcome: PassOutcome, cursor: Option) -> Option { + match outcome { + PassOutcome::Undecodable(slot) => { + let attempts = match self.stalled { + Some((stuck_on, attempts)) if stuck_on == slot => attempts.saturating_add(1), + _ => 1, + }; + if attempts < DECODE_RETRY_LIMIT { + self.stalled = Some((slot, attempts)); + return None; + } + // Set before the pass that reads past the bad slot, so the + // stored floor stays below it. + self.stalled = None; + self.skip = SkipPolicy::Skipping(slot); + Some(slot) + } + // Ours to fix, not the peer's: retry the slot without spending the + // decode budget on it, or a store outage would read past good blocks. + PassOutcome::Undelivered(_) => None, + PassOutcome::Drained => { + if self.passed_the_stall(cursor) { + self.stalled = None; + } + // Checked against the skip's own slot, not against `stalled`, + // which arming a skip clears. Otherwise the first truncated + // stream after arming would downgrade the skip before it had + // read past anything. + if self.skip.used_its_skip(cursor) { + self.skip = self.skip.after_clean_pass(); + } + None + } + } + } + + /// Whether the read position is now past whatever the watcher was stuck on. + /// Vacuously true when it was not stuck. + fn passed_the_stall(self, cursor: Option) -> bool { + self.stalled + .is_none_or(|(stuck_on, _)| cursor.is_some_and(|read_to| read_to >= stuck_on)) + } +} + /// Spawns one watcher task per configured peer. /// /// Each task reads the peer's finalized blocks from Bedrock, recognizes outbound -/// messages addressed to this zone, and injects the matching inbox dispatch as a -/// sequencer-origin transaction into the local mempool. +/// messages addressed to this zone, and records the matching inbox dispatch in +/// the store. Delivering it is block production's job, which drains those +/// records every turn. +/// +/// The returned group must be kept alive for as long as the watchers should +/// run; dropping it stops them, and awaiting +/// [`TaskGroup::shutdown`](crate::task_group::TaskGroup::shutdown) is what +/// proves they have stopped. Each watcher holds an `Arc`, so a +/// watcher still running keeps the `RocksDB` lock held and a restarting +/// sequencer cannot reopen its home directory. +#[must_use] pub fn spawn_watchers( bedrock_config: &BedrockConfig, cross_zone: &CrossZoneConfig, poll_interval: Duration, - mempool_handle: &MemPoolHandle<(TransactionOrigin, LeeTransaction)>, -) { + dbio: &Arc, +) -> TaskGroup { let self_zone: [u8; 32] = *bedrock_config.channel_id.as_ref(); + let mut tasks = Vec::new(); for peer in cross_zone.peers.clone() { let node = NodeHttpClient::new( @@ -38,16 +211,20 @@ pub fn spawn_watchers( let expected_pubkey = peer.expected_block_signing_pubkey.map(|bytes| { PublicKey::try_new(bytes).expect("configured peer block-signing pubkey is a valid key") }); - tokio::spawn(watch_peer( + tasks.push(tokio::spawn(watch_peer( ZoneIndexer::new(ChannelId::from(peer.channel_id), node), - peer.channel_id, - peer.allowed_targets, - expected_pubkey, - self_zone, + PeerContext { + peer_zone: peer.channel_id, + self_zone, + allowed_targets: peer.allowed_targets, + expected_pubkey, + }, poll_interval, - mempool_handle.clone(), - )); + Arc::clone(dbio), + ))); } + + TaskGroup::new(tasks) } #[expect( @@ -56,19 +233,48 @@ pub fn spawn_watchers( )] async fn watch_peer( zone_indexer: ZoneIndexer, - peer_zone: [u8; 32], - allowed_targets: Vec, - expected_pubkey: Option, - self_zone: [u8; 32], + peer: PeerContext, poll_interval: Duration, - mempool_handle: MemPoolHandle<(TransactionOrigin, LeeTransaction)>, + dbio: Arc, ) { + let peer_zone = peer.peer_zone; info!( "Cross-zone watcher started for peer {}", hex::encode(peer_zone) ); - let mut cursor = None; + // Resume from the delivery floor: the highest slot every message of which + // was decoded and recorded. Re-reading a peer channel is safe (the dispatch + // key is content-addressed and the inbox no-ops a replay) but re-records + // every already-delivered message, so without this a restart replayed the + // peer's whole history into the store. + let mut cursor = match get_cross_zone_peer_floor(&dbio, peer_zone) { + Ok(floor) => floor, + Err(err) => { + // Falling back to `None` would re-read the peer's whole history and + // re-inject every message it ever delivered. Stopping is the smaller + // failure, and a stopped watcher shows up as unhealthy. + error!( + "Watcher failed to load the stored delivery floor for peer {}: {err:#}. Stopping this watcher rather than re-reading the channel from the beginning.", + hex::encode(peer_zone) + ); + return; + } + }; + if let Some(slot) = cursor { + info!( + "Resuming watcher for peer {} from slot {slot:?}", + hex::encode(peer_zone) + ); + } + + // The slot the watcher is stuck on and how many passes it has spent there, + // and the slot it has given up on. Keyed by slot so a failure at a new slot + // does not inherit an older slot's count. Both stay in memory: a skip must + // not outlive the process, or a peer whose blocks this build cannot decode + // would be skipped past for good and its messages never delivered, even + // after the decoder is fixed. + let mut state = WatcherState::default(); loop { let stream = match zone_indexer.next_messages(cursor).await { Ok(stream) => stream, @@ -81,46 +287,14 @@ async fn watch_peer( continue; } }; - let mut stream = std::pin::pin!(stream); + let outcome = consume_peer_stream(stream, &peer, &dbio, &mut cursor, state.skip).await; - while let Some((msg, slot)) = stream.next().await { - let zone_block = match msg { - ZoneMessage::Block(block) => block, - ZoneMessage::Deposit(_) | ZoneMessage::Withdraw(_) => continue, - }; - match borsh::from_slice::(&zone_block.data) { - Ok(block) => { - debug!( - "Watcher observed finalized peer {} block {}", - hex::encode(peer_zone), - block.header.block_id - ); - // Reject blocks not signed by the pinned peer key (equivocation): - // the channel signer is authenticated by the zone-sdk, but that - // does not prove the peer's honest sequencer produced the block. - if expected_pubkey - .as_ref() - .is_some_and(|pk| !block.is_signed_by(pk)) - { - warn!( - "Watcher dropping peer {} block {}: block-signing key does not match the pinned key", - hex::encode(peer_zone), - block.header.block_id - ); - } else { - deliver_block( - &block, - peer_zone, - self_zone, - &allowed_targets, - &mempool_handle, - ) - .await; - } - } - Err(err) => error!("Watcher failed to deserialize peer block: {err}"), - } - cursor = Some(slot); + if let Some(slot) = state.after_pass(outcome, cursor) { + error!( + "Watcher for peer {} could not decode slot {slot:?} after {DECODE_RETRY_LIMIT} attempts; reading past it. Messages in that block are undelivered until this node can decode it, and the delivery floor stops advancing, so every restart re-reads from {:?} onwards.", + hex::encode(peer_zone), + get_cross_zone_peer_floor(&dbio, peer_zone).ok().flatten() + ); } // Stream ended (caught up to the peer's last finalized block); poll again. @@ -128,14 +302,142 @@ async fn watch_peer( } } -/// Scans one peer block for outbound messages and injects a dispatch per match. -async fn deliver_block( - block: &Block, +/// Delivers the peer blocks carried by `stream`, moving `cursor` as it goes and +/// persisting the delivery floor behind it. Says why the pass ended, since only +/// a block this node cannot decode counts towards [`DECODE_RETRY_LIMIT`]. +/// +/// A block that fails to deserialize ends the pass without advancing, so the +/// next poll re-reads it and a transient failure heals. [`SkipPolicy`] names a +/// slot the caller gave up on after [`DECODE_RETRY_LIMIT`] attempts, which is +/// read past so a permanently undecodable inscription cannot wedge the watcher, +/// and says whether the floor may still move: past a skipped slot it may not, +/// because the floor is what a restart resumes from and the skipped messages +/// have to stay reachable. +async fn consume_peer_stream( + stream: S, + peer: &PeerContext, + dbio: &RocksDBIO, + cursor: &mut Option, + skip: SkipPolicy, +) -> PassOutcome +where + S: Stream, +{ + let mut stream = std::pin::pin!(stream); + // The slot being consumed: every message of it seen so far is handled, but + // there may be more to come, so the cursor may not advance onto it yet. + let mut in_progress: Option = None; + + while let Some((msg, slot)) = stream.next().await { + if in_progress != Some(slot) { + // A message from a later slot means the previous one completed. + if let Some(done) = in_progress { + advance_cursor(dbio, peer.peer_zone, cursor, done, skip.persists_floor()); + } + in_progress = Some(slot); + } + + let zone_block = match msg { + ZoneMessage::Block(block) => block, + ZoneMessage::Deposit(_) | ZoneMessage::Withdraw(_) => continue, + }; + match borsh::from_slice::(&zone_block.data) { + Ok(block) => { + debug!( + "Watcher observed finalized peer {} block {}", + hex::encode(peer.peer_zone), + block.header.block_id + ); + // Reject blocks not signed by the pinned peer key (equivocation): + // the channel signer is authenticated by the zone-sdk, but that + // does not prove the peer's honest sequencer produced the block. + if peer + .expected_pubkey + .as_ref() + .is_some_and(|pk| !block.is_signed_by(pk)) + { + warn!( + "Watcher dropping peer {} block {}: block-signing key does not match the pinned key", + hex::encode(peer.peer_zone), + block.header.block_id + ); + continue; + } + + if !record_block_deliveries(&block, peer, dbio) { + // Recording a delivery is what makes it survive the mempool. + // Letting the pass finish here would move the floor past this + // slot on a store that just refused the write, and nothing + // re-reads a slot below the floor. + error!( + "Watcher could not record every delivery in peer {} block {}. Holding the floor and retrying the slot.", + hex::encode(peer.peer_zone), + block.header.block_id + ); + return PassOutcome::Undelivered(slot); + } + } + Err(err) if skip.skip_slot() == Some(slot) => { + debug!( + "Watcher skipping undecodable peer {} block at slot {slot:?}: {err}", + hex::encode(peer.peer_zone) + ); + } + Err(err) => { + error!( + "Watcher failed to deserialize peer {} block at slot {slot:?}: {err}. Holding the cursor and retrying.", + hex::encode(peer.peer_zone) + ); + return PassOutcome::Undecodable(slot); + } + } + } + + // The stream drained cleanly, so the slot in progress completed too. + if let Some(done) = in_progress { + advance_cursor(dbio, peer.peer_zone, cursor, done, skip.persists_floor()); + } + PassOutcome::Drained +} + +/// Moves the in-memory read cursor past `slot`, and the durable delivery floor +/// with it while `persist_floor` holds. +/// +/// A persist failure is only logged: the worst case is re-reading from the last +/// stored slot after a restart, which delivery handles idempotently. +fn advance_cursor( + dbio: &RocksDBIO, peer_zone: [u8; 32], - self_zone: [u8; 32], - allowed_targets: &[ProgramId], - mempool_handle: &MemPoolHandle<(TransactionOrigin, LeeTransaction)>, + cursor: &mut Option, + slot: Slot, + persist_floor: bool, ) { + *cursor = Some(slot); + if !persist_floor { + return; + } + if let Err(err) = set_cross_zone_peer_floor(dbio, peer_zone, slot) { + warn!( + "Failed to persist watcher delivery floor for peer {}: {err:#}", + hex::encode(peer_zone) + ); + } +} + +/// Scans one peer block for outbound messages and records a dispatch per match. +/// +/// Returns `false` if a delivery could not be recorded, which the caller turns +/// into a stall: the record is the only thing standing between a durable read +/// position and a lost message. +fn record_block_deliveries(block: &Block, peer: &PeerContext, dbio: &RocksDBIO) -> bool { + let peer_zone = peer.peer_zone; + let self_zone = peer.self_zone; + let allowed_targets = peer.allowed_targets.as_slice(); + // Collected and written once. The pending list is a single value, so a write + // per delivery would rewrite the whole list once per message, which is + // quadratic in a peer block that carries many of them, on a task holding the + // lock block production needs. + let mut deliveries = Vec::new(); for (index, tx) in block.body.transactions.iter().enumerate() { let LeeTransaction::Public(public_tx) = tx else { continue; @@ -156,30 +458,588 @@ async fn deliver_block( continue; } + let src_tx_index = u32::try_from(index).unwrap_or(u32::MAX); let dispatch = build_dispatch_from_emission( peer_zone, block.header.block_id, - u32::try_from(index).unwrap_or(u32::MAX), + src_tx_index, message.program_id, emission.target_program_id, &emission.target_accounts, emission.payload, ); + let dispatch = LeeTransaction::Public(dispatch); - match mempool_handle - .push(( - TransactionOrigin::Sequencer, - LeeTransaction::Public(dispatch), - )) - .await - { - Ok(()) => info!( - "Watcher injected cross-zone dispatch from peer {} block {} tx {}", + // Recording is the delivery. The floor is durable, so once it advances + // this peer block is never re-read; the record is what block production + // drains on its next turn, and what a restart still has. It is dropped + // when the delivery itself becomes irreversible. + let key = message_key(&peer_zone, block.header.block_id, src_tx_index); + let encoded = match borsh::to_vec(&dispatch) { + Ok(encoded) => encoded, + Err(err) => { + error!( + "Failed to encode cross-zone dispatch {}: {err}", + hex::encode(key) + ); + return false; + } + }; + deliveries.push(PendingCrossZoneDispatchRecord::recorded(key, encoded)); + } + + let offered = deliveries.len(); + match dbio.add_pending_cross_zone_dispatches(deliveries) { + // Fewer accepted than offered means the rest were recorded by an earlier + // pass over the same slot, which the retry loop does up to + // [`DECODE_RETRY_LIMIT`] times. + Ok(accepted) => { + if accepted > 0 { + info!( + "Watcher recorded {accepted} of {offered} cross-zone deliveries from peer {} block {}", + hex::encode(peer_zone), + block.header.block_id + ); + } else { + debug!( + "Watcher already held every cross-zone delivery in peer {} block {}", + hex::encode(peer_zone), + block.header.block_id + ); + } + true + } + // Includes the pending list being full, which is why this holds the + // floor rather than dropping the block: the slot stays re-readable and + // the peer's messages wait instead of being lost. + Err(err) => { + error!( + "Failed to record the {offered} cross-zone deliveries in peer {} block {}: {err}", hex::encode(peer_zone), - block.header.block_id, - index - ), - Err(err) => error!("Watcher failed to enqueue inbox dispatch: {err}"), + block.header.block_id + ); + false } } } + +#[cfg(test)] +mod tests { + use common::test_utils::produce_dummy_block; + use futures::stream; + use lee::{ + PublicTransaction, + public_transaction::{Message, WitnessSet}, + }; + use logos_blockchain_core::mantle::ops::channel::{MsgId, inscribe::Inscription}; + use logos_blockchain_zone_sdk::ZoneBlock; + use ping_core::{SenderInstruction, ping_record_pda}; + use storage::sequencer::{DB_META_PENDING_CROSS_ZONE_DISPATCHES_KEY, RocksDBIO}; + use tempfile::TempDir; + + use super::*; + + const SELF_ZONE: [u8; 32] = [1; 32]; + const PEER_ZONE: [u8; 32] = [2; 32]; + + fn peer_context() -> PeerContext { + PeerContext { + peer_zone: PEER_ZONE, + self_zone: SELF_ZONE, + allowed_targets: vec![programs::ping_receiver().id()], + expected_pubkey: None, + } + } + + /// A store backed by a temp dir. The dir is returned so it outlives the db. + fn store() -> (TempDir, RocksDBIO) { + let dir = tempfile::tempdir().expect("temp dir"); + let genesis = produce_dummy_block(0, None, vec![]); + let dbio = RocksDBIO::create(dir.path(), &genesis, &lee::V03State::new()).expect("db"); + (dir, dbio) + } + + /// A `ping_sender` emission addressed to `SELF_ZONE`. + fn emission() -> LeeTransaction { + let receiver_id = programs::ping_receiver().id(); + let send = SenderInstruction::Send { + outbox_program_id: programs::cross_zone_outbox().id(), + target_zone: SELF_ZONE, + target_program_id: receiver_id, + target_accounts: vec![ping_record_pda(receiver_id).into_value()], + payload: b"hi".to_vec(), + ordinal: 0, + }; + let message = Message::try_new(programs::ping_sender().id(), vec![], vec![], send) + .expect("emission serializes"); + LeeTransaction::Public(PublicTransaction::new( + message, + WitnessSet::from_raw_parts(vec![]), + )) + } + + fn peer_msg(data: Vec, slot: u64) -> (ZoneMessage, Slot) { + ( + ZoneMessage::Block(ZoneBlock { + id: MsgId::from([0; 32]), + data: Inscription::try_from(data).expect("test inscription is within bounds"), + }), + Slot::from(slot), + ) + } + + /// A stream item carrying block `block_id` with one emission for this zone. + fn peer_block_msg(block_id: u64, slot: u64) -> (ZoneMessage, Slot) { + let block = produce_dummy_block(block_id, None, vec![emission()]); + peer_msg(borsh::to_vec(&block).expect("block serializes"), slot) + } + + fn undecodable_msg(slot: u64) -> (ZoneMessage, Slot) { + peer_msg(b"not a block".to_vec(), slot) + } + + /// The message keys recorded so far, in insertion order. + fn recorded_keys(dbio: &RocksDBIO) -> Vec<[u8; 32]> { + dbio.get_pending_cross_zone_dispatches() + .expect("pending dispatches readable") + .into_iter() + .map(|record| record.message_key) + .collect() + } + + /// Makes every later pending-dispatch read fail, standing in for any store + /// failure between reading a peer block and the delivery being durable. + /// Recording reads the list before it writes it, so a value that will not + /// decode is enough. + fn break_the_dispatch_store(dbio: &RocksDBIO) { + let cf = dbio + .db + .cf_handle(storage::CF_META_NAME) + .expect("meta column family"); + let key = borsh::to_vec(&DB_META_PENDING_CROSS_ZONE_DISPATCHES_KEY).expect("key encodes"); + dbio.db + .put_cf(&cf, key, b"not a pending dispatch list") + .expect("write"); + } + + /// Drives the state machine over a sequence of pass outcomes, with the read + /// position after each, and returns the state it lands in. + fn run_passes(passes: &[(PassOutcome, Option)]) -> WatcherState { + let mut state = WatcherState::default(); + for (outcome, cursor) in passes { + state.after_pass(*outcome, cursor.map(Slot::from)); + } + state + } + + fn retry_limit() -> usize { + usize::try_from(DECODE_RETRY_LIMIT).expect("retry limit fits in usize") + } + + fn stall(slot: u64, cursor: Option) -> (PassOutcome, Option) { + (PassOutcome::Undecodable(Slot::from(slot)), cursor) + } + + #[test] + fn a_slot_is_skipped_only_after_the_retry_limit() { + let limit = retry_limit(); + let almost = vec![stall(4, Some(3)); limit - 1]; + assert_eq!( + run_passes(&almost).skip, + SkipPolicy::DeliverAll, + "a slot must not be given up on before the limit" + ); + + let enough = vec![stall(4, Some(3)); limit]; + assert_eq!( + run_passes(&enough).skip, + SkipPolicy::Skipping(Slot::from(4)) + ); + } + + #[test] + fn the_floor_stays_frozen_for_the_rest_of_the_run_after_a_skip() { + // Twenty failures at slot 4, then the pass that reads past it, then + // clean passes: the floor must never be persistable again, or the skip + // survives the next restart and those messages are gone for good. + let mut passes = vec![stall(4, Some(3)); retry_limit()]; + passes.push((PassOutcome::Drained, Some(9))); + passes.push((PassOutcome::Drained, Some(12))); + let state = run_passes(&passes); + + assert_eq!(state.skip, SkipPolicy::FloorFrozen); + assert!(!state.skip.persists_floor()); + assert_eq!(state.stalled, None); + } + + #[test] + fn a_stream_that_ended_before_the_stalled_slot_does_not_reset_the_count() { + // The zone-sdk ends a stream on a fetch failure exactly as it does on + // catching up. Treating that as a clean pass would reset the retry count + // for ever, and the watcher would never escape a slot it cannot decode. + let mut passes = vec![stall(4, Some(3)); 5]; + passes.push((PassOutcome::Drained, Some(3))); + let state = run_passes(&passes); + assert_eq!( + state.stalled, + Some((Slot::from(4), 5)), + "the count survives a pass that never reached the stalled slot" + ); + + // Reading past it is what actually clears the stall. + let mut read_past = vec![stall(4, Some(3)); 5]; + read_past.push((PassOutcome::Drained, Some(7))); + assert_eq!(run_passes(&read_past).stalled, None); + } + + #[test] + fn a_failed_handoff_does_not_spend_the_decode_budget() { + // A store or mempool failure is ours, not the peer's. Counting it here + // would read past a block that decodes perfectly well. + let passes = vec![(PassOutcome::Undelivered(Slot::from(4)), Some(3)); retry_limit() * 2]; + let state = run_passes(&passes); + assert_eq!(state.skip, SkipPolicy::DeliverAll); + assert_eq!(state.stalled, None); + } + + #[test] + fn a_truncated_pass_does_not_disarm_a_skip_before_it_is_used() { + // Arming a skip clears `stalled`, so a `Drained` pass that never reached + // the bad slot passes the stall check vacuously. Downgrading on that + // would disarm the skip before it read past anything, and the slot would + // have to be given up on again from scratch, so a peer endpoint that is + // flaky around one bad slot would never be read past. + let mut passes = vec![stall(4, Some(3)); retry_limit()]; + passes.push((PassOutcome::Drained, Some(3))); + let state = run_passes(&passes); + assert_eq!( + state.skip, + SkipPolicy::Skipping(Slot::from(4)), + "a pass that ended before the skipped slot must leave the skip armed" + ); + + // The pass that actually gets past it is what downgrades. + let mut used = vec![stall(4, Some(3)); retry_limit()]; + used.push((PassOutcome::Drained, Some(7))); + assert_eq!(run_passes(&used).skip, SkipPolicy::FloorFrozen); + } + + #[test] + fn a_stall_at_a_new_slot_starts_its_own_count() { + let passes = vec![stall(4, Some(3)), stall(4, Some(3)), stall(9, Some(8))]; + assert_eq!(run_passes(&passes).stalled, Some((Slot::from(9), 1))); + } + + #[test] + fn a_run_that_skipped_once_never_moves_its_floor_again() { + // The state that makes a skip recoverable: after the bad slot is read + // past, later passes decode cleanly, and the floor still must not move + // over the gap or the skip survives the next restart. + assert_eq!( + SkipPolicy::Skipping(Slot::from(4)).after_clean_pass(), + SkipPolicy::FloorFrozen + ); + assert_eq!( + SkipPolicy::FloorFrozen.after_clean_pass(), + SkipPolicy::FloorFrozen + ); + assert!(!SkipPolicy::FloorFrozen.persists_floor()); + assert_eq!(SkipPolicy::FloorFrozen.skip_slot(), None); + + // A run that has never skipped keeps moving. + assert_eq!( + SkipPolicy::DeliverAll.after_clean_pass(), + SkipPolicy::DeliverAll + ); + assert!(SkipPolicy::DeliverAll.persists_floor()); + } + + #[tokio::test] + async fn watcher_persists_its_cursor_as_it_consumes() { + let (_dir, dbio) = store(); + let mut cursor = None; + + let outcome = consume_peer_stream( + stream::iter(vec![peer_block_msg(1, 0), peer_block_msg(2, 1)]), + &peer_context(), + &dbio, + &mut cursor, + SkipPolicy::DeliverAll, + ) + .await; + + assert_eq!(outcome, PassOutcome::Drained); + assert_eq!(cursor, Some(Slot::from(1))); + assert_eq!( + get_cross_zone_peer_floor(&dbio, PEER_ZONE).unwrap(), + Some(Slot::from(1)), + "the cursor must be durable, not just in memory" + ); + assert_eq!( + recorded_keys(&dbio), + vec![message_key(&PEER_ZONE, 1, 0), message_key(&PEER_ZONE, 2, 0)] + ); + } + + #[tokio::test] + async fn watcher_records_every_delivery_it_reads() { + let (_dir, dbio) = store(); + let mut cursor = None; + + consume_peer_stream( + stream::iter(vec![peer_block_msg(1, 0)]), + &peer_context(), + &dbio, + &mut cursor, + SkipPolicy::DeliverAll, + ) + .await; + + // The read cursor is durable, so once it advances this peer block is + // never re-read. The record is the whole of what survives that: block + // production drains it, and it outlives a restart. It is dropped when + // the delivery itself becomes irreversible, not when it is included. + let records = dbio.get_pending_cross_zone_dispatches().unwrap(); + assert_eq!(records.len(), 1, "the delivery must be recorded"); + assert_eq!( + records[0].message_key, + message_key(&PEER_ZONE, 1, 0), + "the record is keyed by the message it delivers, so a replay is not double-tracked" + ); + assert!( + borsh::from_slice::(&records[0].transaction).is_ok(), + "the recorded bytes must decode, or the drain silently skips them" + ); + assert_eq!( + records[0].failed_attempts, 0, + "a delivery that has never been attempted starts with a clean count" + ); + } + + #[tokio::test] + async fn a_delivery_that_cannot_be_recorded_holds_the_floor() { + let (_dir, dbio) = store(); + break_the_dispatch_store(&dbio); + let mut cursor = None; + + let outcome = consume_peer_stream( + stream::iter(vec![peer_block_msg(1, 0)]), + &peer_context(), + &dbio, + &mut cursor, + SkipPolicy::DeliverAll, + ) + .await; + + // The floor is durable and nothing re-reads a slot below it, so a pass + // that failed to record must not let it move, or the delivery is lost + // rather than retried. + assert_eq!(outcome, PassOutcome::Undelivered(Slot::from(0))); + assert_eq!( + get_cross_zone_peer_floor(&dbio, PEER_ZONE).unwrap(), + None, + "the slot must stay re-readable" + ); + } + + #[tokio::test] + async fn watcher_resumes_from_the_persisted_cursor_without_rereading() { + let (_dir, dbio) = store(); + let mut cursor = None; + + consume_peer_stream( + stream::iter(vec![peer_block_msg(1, 0), peer_block_msg(2, 1)]), + &peer_context(), + &dbio, + &mut cursor, + SkipPolicy::DeliverAll, + ) + .await; + assert_eq!(recorded_keys(&dbio).len(), 2); + + // Restart: a fresh watcher seeds its cursor from the store rather than + // starting at `None`, which is what stops it re-reading the peer channel + // from genesis. + let resumed = get_cross_zone_peer_floor(&dbio, PEER_ZONE).unwrap(); + assert_eq!(resumed, Some(Slot::from(1))); + + // The sdk resumes the stream at cursor + 1, so only block 3 arrives. + let mut resumed_cursor = resumed; + consume_peer_stream( + stream::iter(vec![peer_block_msg(3, 2)]), + &peer_context(), + &dbio, + &mut resumed_cursor, + SkipPolicy::DeliverAll, + ) + .await; + + assert_eq!( + recorded_keys(&dbio), + vec![ + message_key(&PEER_ZONE, 1, 0), + message_key(&PEER_ZONE, 2, 0), + message_key(&PEER_ZONE, 3, 0) + ], + "only the unread block is recorded on the second pass" + ); + assert_eq!( + get_cross_zone_peer_floor(&dbio, PEER_ZONE).unwrap(), + Some(Slot::from(2)) + ); + } + + #[tokio::test] + async fn watcher_does_not_persist_past_an_undecodable_block() { + let (_dir, dbio) = store(); + let mut cursor = None; + + let outcome = consume_peer_stream( + stream::iter(vec![ + peer_block_msg(1, 0), + undecodable_msg(1), + peer_block_msg(3, 2), + ]), + &peer_context(), + &dbio, + &mut cursor, + SkipPolicy::DeliverAll, + ) + .await; + + // A durable cursor makes this load-bearing: advancing past the bad block + // would drop its messages permanently rather than until the next restart. + assert_eq!(outcome, PassOutcome::Undecodable(Slot::from(1))); + assert_eq!( + get_cross_zone_peer_floor(&dbio, PEER_ZONE).unwrap(), + Some(Slot::from(0)) + ); + assert_eq!( + recorded_keys(&dbio), + vec![message_key(&PEER_ZONE, 1, 0)], + "the block after the failure is unread" + ); + } + + #[tokio::test] + async fn watcher_does_not_persist_inside_a_partially_failed_slot() { + // One slot can carry several messages. Persisting after each message + // would store a cursor the retry resumes past, so the message that + // failed is never re-read and its delivery is lost for good. + let (_dir, dbio) = store(); + let mut cursor = None; + + let outcome = consume_peer_stream( + stream::iter(vec![peer_block_msg(1, 4), undecodable_msg(4)]), + &peer_context(), + &dbio, + &mut cursor, + SkipPolicy::DeliverAll, + ) + .await; + + assert_eq!(outcome, PassOutcome::Undecodable(Slot::from(4))); + assert_eq!(cursor, None, "slot 4 is re-read whole on the next pass"); + assert_eq!(get_cross_zone_peer_floor(&dbio, PEER_ZONE).unwrap(), None); + assert_eq!(recorded_keys(&dbio), vec![message_key(&PEER_ZONE, 1, 0)]); + } + + #[tokio::test] + async fn watcher_reads_past_a_slot_it_has_given_up_on() { + let (_dir, dbio) = store(); + let mut cursor = None; + + let outcome = consume_peer_stream( + stream::iter(vec![ + peer_block_msg(1, 0), + undecodable_msg(1), + peer_block_msg(3, 2), + ]), + &peer_context(), + &dbio, + &mut cursor, + SkipPolicy::Skipping(Slot::from(1)), + ) + .await; + + assert_eq!(outcome, PassOutcome::Drained, "the pass drains"); + assert_eq!( + recorded_keys(&dbio), + vec![message_key(&PEER_ZONE, 1, 0), message_key(&PEER_ZONE, 3, 0)], + "only the skipped block goes unrecorded" + ); + + // The cursor moves so later blocks are still read, but the durable floor + // does not follow it past the gap. + assert_eq!( + cursor, + Some(Slot::from(2)), + "the pass keeps reading forward" + ); + assert_eq!( + get_cross_zone_peer_floor(&dbio, PEER_ZONE).unwrap(), + None, + "the floor must not move past a slot this node could not decode" + ); + } + + #[tokio::test] + async fn a_restart_re_reads_a_skipped_slot() { + let (_dir, dbio) = store(); + let mut cursor = None; + + // Slot 0 is recorded, slot 1 is undecodable and eventually skipped, slot + // 2 is recorded on top of the gap. + consume_peer_stream( + stream::iter(vec![peer_block_msg(1, 0)]), + &peer_context(), + &dbio, + &mut cursor, + SkipPolicy::DeliverAll, + ) + .await; + consume_peer_stream( + stream::iter(vec![undecodable_msg(1), peer_block_msg(3, 2)]), + &peer_context(), + &dbio, + &mut cursor, + SkipPolicy::Skipping(Slot::from(1)), + ) + .await; + assert_eq!(recorded_keys(&dbio).len(), 2); + + // A fresh watcher seeds from the floor, so slot 1 comes back around + // rather than being skipped for the life of the store. That is what + // makes a decoder fix recover the messages instead of a store reset. + let resumed = get_cross_zone_peer_floor(&dbio, PEER_ZONE).unwrap(); + assert_eq!(resumed, Some(Slot::from(0))); + + let mut resumed_cursor = resumed; + consume_peer_stream( + stream::iter(vec![peer_block_msg(2, 1), peer_block_msg(3, 2)]), + &peer_context(), + &dbio, + &mut resumed_cursor, + SkipPolicy::DeliverAll, + ) + .await; + + // Three records, not four: the block at slot 2 was recorded on the + // earlier pass and the re-read does not double-track it, while the block + // at slot 1, skipped before, is recorded for the first time. + assert_eq!( + recorded_keys(&dbio), + vec![ + message_key(&PEER_ZONE, 1, 0), + message_key(&PEER_ZONE, 3, 0), + message_key(&PEER_ZONE, 2, 0) + ], + "the previously skipped block must be recorded after a restart, and nothing re-recorded" + ); + assert_eq!( + get_cross_zone_peer_floor(&dbio, PEER_ZONE).unwrap(), + Some(Slot::from(2)), + "with the gap filled the floor moves again" + ); + } +} diff --git a/lez/sequencer/core/src/lib.rs b/lez/sequencer/core/src/lib.rs index 5ed0e50c..854e6e3e 100644 --- a/lez/sequencer/core/src/lib.rs +++ b/lez/sequencer/core/src/lib.rs @@ -16,6 +16,7 @@ use common::{ transaction::{LeeTransaction, clock_invocation}, }; use config::{GenesisAction, SequencerConfig}; +use cross_zone_inbox_core::CrossZoneMessage; use futures::StreamExt as _; use itertools::Itertools as _; use lee::{AccountId, PublicTransaction, public_transaction::Message}; @@ -33,12 +34,16 @@ use num_bigint::BigUint; pub use storage::error::DbError; use storage::sequencer::{ RocksDBIO, StoreUpdate, - sequencer_cells::{PendingDepositEventRecord, WithdrawalReconciliationKey, ZoneAnchorRecord}, + sequencer_cells::{ + PendingCrossZoneDispatchRecord, PendingDepositEventRecord, WithdrawalReconciliationKey, + ZoneAnchorRecord, + }, }; use crate::{ block_publisher::{BlockPublisherTrait, MsgId, ZoneSdkPublisher}, block_store::SequencerStore, + task_group::{StoreRelease, TaskGroup}, }; pub mod block_publisher; @@ -48,6 +53,23 @@ pub mod cross_zone_watcher; #[cfg(feature = "mock")] pub mod mock; +pub mod task_group; + +/// Failed production attempts before a cross-zone dispatch is given up on. +/// +/// One attempt per block, so this is tens of seconds of retrying. Enough for a +/// failure that is not the message's fault to clear, short enough that a message +/// which will never execute stops being retried. +const RETIRE_DISPATCH_AFTER_FAILURES: u32 = 3; + +/// Cross-zone deliveries one block may carry. +/// +/// Each one costs a guest execution whether it succeeds or fails, and what +/// queues them up is chosen by peer zones. Without a bound, a backlog decides +/// how long a block takes to build and leaves no room for user transactions, +/// since store-drained work is taken before the mempool. The rest wait one +/// block; nothing is dropped. +const MAX_DISPATCHES_PER_BLOCK: usize = 16; /// The origin of a transaction. #[derive(Clone, Copy)] @@ -71,6 +93,10 @@ pub struct SequencerCore { mempool: MemPool<(TransactionOrigin, LeeTransaction)>, sequencer_config: SequencerConfig, block_publisher: BP, + /// Cross-zone watchers, stopped when this sequencer is dropped. They hold a + /// store handle, so leaving them running would keep the `RocksDB` lock held + /// and make the home directory unopenable by a restarting sequencer. + watchers: TaskGroup, } impl SequencerCore { @@ -202,14 +228,17 @@ impl SequencerCore { // Cross-zone messaging: start a watcher per configured peer. The inbox // config account is seeded into genesis state in `build_genesis_state`. - if let Some(cross_zone) = &config.cross_zone { - cross_zone_watcher::spawn_watchers( - &config.bedrock_config, - cross_zone, - config.block_create_timeout, - &mempool_handle, - ); - } + let watchers = config + .cross_zone + .as_ref() + .map_or_else(TaskGroup::default, |cross_zone| { + cross_zone_watcher::spawn_watchers( + &config.bedrock_config, + cross_zone, + config.block_create_timeout, + &store.dbio(), + ) + }); // Before producing, verify our local state still belongs to the chain // the channel serves and replay any channel blocks we are missing // (e.g. from other sequencers). @@ -267,6 +296,7 @@ impl SequencerCore { mempool, sequencer_config: config, block_publisher, + watchers, }; (sequencer_core, mempool_handle) @@ -423,6 +453,12 @@ impl SequencerCore { .context("Failed to read stored block")? { Some(stored) if stored.header.hash == block_hash => { + // Already applied, but the channel serving it is what makes + // it irreversible, so its deliveries are settled and their + // records are owed nothing. Without this a restart leaves a + // record for every delivery it already published, and + // nothing downstream would ever remove them. + settle_reconstructed_deliveries(store, &stored); store .set_zone_anchor(&record) .context("Failed to persist zone anchor")?; @@ -464,6 +500,9 @@ impl SequencerCore { .iter() .filter_map(extract_bridge_deposit_id) .collect(); + // The same for the deliveries it carries: the inbox has seen them, so + // the drain would skip them anyway, and the records are owed nothing. + let finalized_dispatch_keys = settled_dispatch_keys(&store.dbio(), block); // The tip meta stays pinned to the head tip even when the reconstructed // block lands below it, and the anchor only advances if the block @@ -477,6 +516,7 @@ impl SequencerCore { head_tip: head_tip.as_ref(), final_snapshot: final_meta.as_ref().map(|meta| (chain.final_state(), meta)), remove_deposit_records: &finalized_deposit_ids, + remove_dispatch_records: &finalized_dispatch_keys, zone_anchor: Some(&record), ..StoreUpdate::new(chain.head_state()) }) @@ -638,8 +678,42 @@ impl SequencerCore { fn build_block_from_mempool(&mut self) -> Result { let now = Instant::now(); - // Build on the head: its tip is the parent, its state the validation base. - let (prev_block_hash, new_block_height, mut working_state) = { + // Decoded outside the chain lock, and read before it is taken: the usual + // case is no delivery records at all, and decoding is the expensive part. + // One that does not decode is dropped rather than kept, since nothing + // will ever turn those bytes into a block transaction. + let mut settled = Vec::new(); + let recorded_dispatches: Vec<_> = self + .store + .dbio() + .get_pending_cross_zone_dispatches() + .context("Failed to load pending cross-zone dispatches")? + .into_iter() + .filter_map( + |record| match borsh::from_slice::(&record.transaction) { + Ok(tx) => { + let message = extract_cross_zone_dispatch(&tx); + Some((record.message_key, message, tx)) + } + Err(err) => { + warn!( + "Dropping pending cross-zone dispatch {} that does not decode: {err:#}", + hex::encode(record.message_key) + ); + settled.push(record.message_key); + None + } + }, + ) + .collect(); + + // Build on the head: its tip is the parent, its state the validation + // base. + // + // The delivery records are classified in here rather than after, so the + // final state can be read by reference. Cloning it cost a full state + // copy on every block of every zone, cross-zone or not. + let (prev_block_hash, new_block_height, mut working_state, pending_dispatches) = { let chain = self.chain.lock().expect("chain state mutex poisoned"); let tip = chain.head_tip(); let height = tip.as_ref().map_or(GENESIS_BLOCK_ID, |head| { @@ -648,9 +722,43 @@ impl SequencerCore { .expect("block id should not overflow") }); let prev = tip.map_or(HashType([0; 32]), |head| head.hash); - (prev, height, chain.head_state().clone()) + + // Three outcomes per record. Already in the final state means the + // delivery is irreversible, so the record is dropped; that is the + // only thing that removes a record the watcher re-added after its + // delivery had already settled, which it does whenever it re-reads a + // slot it has consumed. Already in the head state but not the final + // one means the delivery is on this chain but could still orphan, so + // the record is skipped and kept. Otherwise it goes in this block. + let mut pending: VecDeque = VecDeque::new(); + for (key, message, tx) in recorded_dispatches { + match message { + Some(message) if dispatch_already_delivered(chain.final_state(), &message) => { + settled.push(key); + } + Some(message) if dispatch_already_delivered(chain.head_state(), &message) => {} + _ if pending.len() >= MAX_DISPATCHES_PER_BLOCK => {} + _ => pending.push_back(tx), + } + } + + (prev, height, chain.head_state().clone(), pending) }; + if !settled.is_empty() + && let Err(err) = self + .store + .dbio() + .drop_settled_cross_zone_dispatches(&settled) + { + // Only bookkeeping: the deliveries themselves are irreversible, and + // the next turn tries again. + warn!( + "Failed to drop {} settled delivery record(s): {err:#}", + settled.len() + ); + } + let mut valid_transactions = Vec::new(); let mut withdrawals = Vec::new(); @@ -664,7 +772,7 @@ impl SequencerCore { // build on — it was minted by us or by a peer whose block we adopted. // An orphan reverts the receipt with the block, so the next turn // re-mints without any bookkeeping of our own. - let mut pending_deposits: VecDeque = self + let pending_deposits: VecDeque = self .store .get_pending_deposit_events() .context("Failed to load pending deposit events")? @@ -691,10 +799,13 @@ impl SequencerCore { let clock_tx = clock_invocation(new_block_timestamp); let clock_lee_tx = LeeTransaction::Public(clock_tx.clone()); - // Pending deposit mints first, then user work. `from_store` is not the - // same as a `Sequencer` origin — the cross-zone watcher pushes those - // into the mempool too. - while let Some((origin, tx, from_store)) = pending_deposits + // Everything drained from the store first, then user work. `from_store` + // is not the same as a `Sequencer` origin: it says the transaction has a + // record behind it and so needs no requeue, where the origin only says + // it was not submitted by a user. + let mut pending_from_store = pending_deposits; + pending_from_store.extend(pending_dispatches); + while let Some((origin, tx, from_store)) = pending_from_store .pop_front() .map(|tx| (TransactionOrigin::Sequencer, tx, true)) .or_else(|| self.mempool.pop().map(|(origin, tx)| (origin, tx, false))) @@ -719,12 +830,39 @@ impl SequencerCore { .len(); if block_size > max_block_size { + // Would a block carrying nothing but this still be too big? Then + // it does not fit in any block and deferring it defers it for + // ever. A store-drained transaction is at the head of the queue + // every turn, so breaking here would stop production reaching + // anything behind it, including the whole mempool, permanently. + // Count it against the delivery instead so it is given up on. + // + // Measured on its own rather than from `block_size`, which also + // counts whatever this block already holds: a transaction that + // merely does not fit *today* is the ordinary deferral below. + if from_store + && !self.fits_in_an_empty_block( + &tx, + &clock_lee_tx, + new_block_height, + prev_block_hash, + new_block_timestamp, + )? + { + error!( + "Sequencer-drained transaction {tx_hash} cannot fit in any block under the \ + {max_block_size} byte limit; giving up on it rather than stalling production", + ); + self.count_dispatch_failure(&tx); + continue; + } + warn!( "Transaction with hash {tx_hash} deferred to next block: \ block size {block_size} bytes would exceed limit of {max_block_size} bytes", ); - // A deposit mint needs no requeue: its record stays unfulfilled - // in the store and is drained again on the next turn. + // Anything drained from the store needs no requeue: its record + // stays there and is drained again on the next turn. if !from_store { self.mempool.push_front((origin, tx)); } @@ -740,6 +878,11 @@ impl SequencerCore { &mut withdrawals, ) { valid_transactions.push(tx); + } else { + // A failed transaction is simply left out of the block, except a + // dispatch: that one is re-fed from the store every turn, so one + // that can never execute would fail on every block for ever. + self.count_dispatch_failure(&tx); } if valid_transactions.len() >= self.sequencer_config.max_num_tx_in_block { @@ -828,6 +971,93 @@ impl SequencerCore { &self.block_publisher } + /// Whether a block carrying nothing but `tx` and the clock would be within + /// the size limit. + /// + /// Distinguishes "does not fit in this block" from "does not fit in any + /// block". The first is an ordinary deferral; the second, for a transaction + /// the store re-feeds every turn, is a permanent stall unless it is given up + /// on. + fn fits_in_an_empty_block( + &self, + tx: &LeeTransaction, + clock_tx: &LeeTransaction, + block_id: u64, + prev_block_hash: HashType, + timestamp: u64, + ) -> Result { + let alone = HashableBlockData { + block_id, + transactions: vec![tx.clone(), clock_tx.clone()], + prev_block_hash, + timestamp, + }; + let size = borsh::to_vec(&alone) + .context("Failed to serialize block for size check")? + .len(); + let max = usize::try_from(self.sequencer_config.max_block_size.as_u64()) + .expect("`max_block_size` should fit into usize"); + Ok(size <= max) + } + + /// Counts one failed production attempt against `tx` if it is a cross-zone + /// delivery, giving up on it once too many accumulate. + /// + /// A delivery's payload and target accounts are chosen on the peer zone and + /// validated by nobody in between, so one can fail for good; but a failure + /// can equally be a property of the moment, so give up only after several. + /// Giving up drops the record, which is also what keeps a peer from growing + /// the pending list with deliveries that can never execute. + fn count_dispatch_failure(&self, tx: &LeeTransaction) { + let Some(message) = extract_cross_zone_dispatch(tx) else { + return; + }; + let key = cross_zone_inbox_core::message_key( + &message.src_zone, + message.src_block_id, + message.src_tx_index, + ); + match self + .store + .dbio() + .record_dispatch_failure(key, RETIRE_DISPATCH_AFTER_FAILURES) + { + Ok(true) => error!( + "Giving up on cross-zone delivery {} after {RETIRE_DISPATCH_AFTER_FAILURES} failed attempts; it will not be retried", + hex::encode(key) + ), + Ok(false) => warn!( + "Cross-zone delivery {} failed to execute, will retry next block", + hex::encode(key) + ), + Err(err) => error!( + "Failed to count the failed attempt for cross-zone delivery {}: {err:#}", + hex::encode(key) + ), + } + } + + /// A weak reference to this sequencer's store, for a shutdown path that + /// needs to observe the database actually closing rather than infer it. + #[must_use] + pub fn store_release(&self) -> StoreRelease { + StoreRelease::new(&self.store.dbio()) + } + + /// Every background task that holds this sequencer's store handle. + /// + /// Taken before the core is shared, so a shutdown path can wait for them + /// without owning the core. Until all of them have stopped the `RocksDB` + /// lock is still held and the home directory cannot be reopened, which is + /// what a restart does. + #[must_use] + pub fn background_tasks(&self) -> Vec { + vec![ + self.watchers.clone(), + self.block_publisher.background_tasks(), + ] + } + /// Whether this sequencer is currently authorized to write to the channel. #[must_use] pub fn is_our_turn(&self) -> bool { @@ -857,6 +1087,29 @@ fn deposit_already_minted(state: &lee::V03State, deposit_op_id: HashType) -> boo .is_some_and(|receipt| *receipt != lee::Account::default()) } +/// Whether a cross-zone delivery is already on the chain we are building on. +/// +/// The inbox records every delivered message key in a seen shard and no-ops a +/// replay, so that shard is the same kind of answer the deposit receipt gives: +/// state, not bookkeeping. An orphan reverts the entry with the block, so the +/// next turn re-delivers with nothing of ours to unwind. +fn dispatch_already_delivered(state: &lee::V03State, message: &CrossZoneMessage) -> bool { + let shard_id = cross_zone_inbox_core::inbox_seen_shard_account_id( + programs::cross_zone_inbox().id(), + &message.src_zone, + message.src_block_id, + ); + state.get_account_by_id_ref(shard_id).is_some_and(|shard| { + cross_zone_inbox_core::SeenShard::from_bytes(shard.data.as_ref()).is_ok_and(|seen| { + seen.contains(&cross_zone_inbox_core::message_key( + &message.src_zone, + message.src_block_id, + message.src_tx_index, + )) + }) + }) +} + /// Feed one channel delta into the follow state and mirror it to the store: /// revert orphaned, then apply and persist adopted and finalized blocks. /// Production builds on this same head. Wired to the publisher via @@ -975,6 +1228,14 @@ fn apply_follow_update( .filter_map(extract_bridge_deposit_id) .collect(); + // The same for cross-zone deliveries, keyed by message key: a record + // goes once its own delivery is irreversible, never because another + // block finalized at its height. + let finalized_dispatch_keys: Vec<[u8; 32]> = irreversible + .iter() + .flat_map(|block| settled_dispatch_keys(dbio, block)) + .collect(); + // A persist 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 cancellation halts the node. @@ -987,6 +1248,7 @@ fn apply_follow_update( finalized_up_to: last_finalized, new_deposit_events: &deposit_records, remove_deposit_records: &finalized_deposit_ids, + remove_dispatch_records: &finalized_dispatch_keys, consumed_withdrawals: &consumed_withdrawals, ..StoreUpdate::new(chain.head_state()) }) @@ -1226,6 +1488,104 @@ fn is_sequencer_only_tx(tx: &LeeTransaction) -> bool { if is_sequencer_only_program(tx.message().program_id)) } +/// The cross-zone message an inbox dispatch delivers, or `None` if `tx` is not +/// a dispatch. +#[must_use] +fn extract_cross_zone_dispatch(tx: &LeeTransaction) -> Option { + let LeeTransaction::Public(tx) = tx else { + return None; + }; + + let message = tx.message(); + if message.program_id != programs::cross_zone_inbox().id() { + return None; + } + + match risc0_zkvm::serde::from_slice::( + &message.instruction_data, + ) { + Ok(cross_zone_inbox_core::Instruction::Dispatch(msg)) => Some(msg), + Ok(cross_zone_inbox_core::Instruction::InitConfig(_)) | Err(_) => None, + } +} + +/// The content-addressed key of the message an inbox dispatch delivers. +/// +/// A delivery in an irreversible block settles its pending record, so the record +/// is dropped by identity rather than by the height it happened to land at. +#[must_use] +fn extract_cross_zone_dispatch_key(tx: &LeeTransaction) -> Option<[u8; 32]> { + extract_cross_zone_dispatch(tx).map(|msg| { + cross_zone_inbox_core::message_key(&msg.src_zone, msg.src_block_id, msg.src_tx_index) + }) +} + +/// The keys of the deliveries `block` carries, reporting any whose transaction +/// is not the one we recorded for that key. +/// +/// The key covers `(src_zone, src_block_id, src_tx_index)` and nothing about the +/// payload, and so does the inbox's own replay check, so a sequencer that +/// publishes a dispatch with the right key and a forged payload settles our +/// correct record along with it. The forgery is caught downstream by the +/// indexer, which re-derives every delivery and halts, but the local record is +/// the last copy of what we believed and it is about to be dropped either way. +/// Saying so in the log is what makes the halt diagnosable. +fn settled_dispatch_keys(dbio: &RocksDBIO, block: &Block) -> Vec<[u8; 32]> { + let recorded = dbio.get_pending_cross_zone_dispatches().unwrap_or_default(); + let (keys, forged) = classify_settled_deliveries(&recorded, block); + for key in forged { + error!( + "Cross-zone delivery {} settled with a transaction that is not the one this node recorded for that key. The message key does not cover the payload, so a peer's sequencer can publish a different delivery under it.", + hex::encode(key) + ); + } + keys +} + +/// Splits the deliveries `block` carries into every settled key, and the subset +/// whose transaction is not the one `recorded` holds for that key. +/// +/// Separated from the logging so the detection is testable: a forged delivery +/// leaves no trace in state that differs from an honest one, precisely because +/// the key does not cover the payload. +fn classify_settled_deliveries( + recorded: &[PendingCrossZoneDispatchRecord], + block: &Block, +) -> (Vec<[u8; 32]>, Vec<[u8; 32]>) { + let mut keys = Vec::new(); + let mut forged = Vec::new(); + for tx in &block.body.transactions { + let Some(key) = extract_cross_zone_dispatch_key(tx) else { + continue; + }; + let mismatched = recorded + .iter() + .find(|record| record.message_key == key) + .is_some_and(|record| { + borsh::to_vec(tx).is_ok_and(|encoded| encoded != record.transaction) + }); + if mismatched { + forged.push(key); + } + keys.push(key); + } + (keys, forged) +} + +/// Drops the records of deliveries carried by a reconstructed block. +/// +/// A persist failure is only logged: the deliveries are already irreversible, so +/// the worst case is a record the next drain drops instead. +fn settle_reconstructed_deliveries(store: &SequencerStore, block: &Block) { + let keys = settled_dispatch_keys(&store.dbio(), block); + if keys.is_empty() { + return; + } + if let Err(err) = store.dbio().drop_settled_cross_zone_dispatches(&keys) { + warn!("Failed to settle reconstructed delivery records: {err:#}"); + } +} + #[must_use] fn extract_bridge_deposit_id(tx: &LeeTransaction) -> Option { let LeeTransaction::Public(tx) = tx else { diff --git a/lez/sequencer/core/src/task_group.rs b/lez/sequencer/core/src/task_group.rs new file mode 100644 index 00000000..8572a62f --- /dev/null +++ b/lez/sequencer/core/src/task_group.rs @@ -0,0 +1,137 @@ +//! A set of background tasks that can be stopped and waited on. + +use std::sync::{Arc, Mutex, MutexGuard, PoisonError, Weak}; + +use log::warn; +use storage::sequencer::RocksDBIO; +use tokio::task::JoinHandle; + +/// Background tasks owned by one component, stoppable on demand and stopped +/// anyway when the last handle goes away. +/// +/// `JoinHandle::abort` only *requests* cancellation, and dropping a handle +/// detaches rather than cancels, so neither on its own says when a task has +/// actually stopped. That matters because these tasks hold a store handle: +/// until they are gone the `RocksDB` lock is still held and a restarting +/// sequencer cannot reopen its home directory. [`TaskGroup::shutdown`] is the +/// answer to "have they stopped yet"; the `Drop` below stays as the best-effort +/// path for panics and tests that never call it. +/// +/// Cloneable so the owner can keep it (tying task lifetime to its own) while a +/// shutdown path elsewhere holds a clone. +#[derive(Clone, Default)] +pub struct TaskGroup(Arc); + +#[derive(Default)] +struct TaskGroupInner(Mutex>>); + +/// A weak handle to the store, for observing when it is finally closed. +/// +/// Every strong reference lives inside a task or a server that shutdown stops, +/// but the last drop runs on whichever thread owned it, not on the one awaiting +/// shutdown. Watching the count is the difference between knowing the database +/// file is closed and assuming it from another crate's drop order. +pub struct StoreRelease(Weak); + +impl StoreRelease { + #[must_use] + pub fn new(store: &Arc) -> Self { + Self(Arc::downgrade(store)) + } + + /// How many holders are left. Zero means the store is closed. + #[must_use] + pub fn holders(&self) -> usize { + self.0.strong_count() + } +} + +impl Drop for TaskGroupInner { + fn drop(&mut self) { + for task in Self::take(&self.0) { + task.abort(); + } + } +} + +impl TaskGroupInner { + /// Empties the handle list, so a second shutdown (or a drop after one) is a + /// no-op rather than a second abort. + fn handles(handles: &Mutex>>) -> MutexGuard<'_, Vec>> { + handles.lock().unwrap_or_else(PoisonError::into_inner) + } + + fn take(handles: &Mutex>>) -> Vec> { + std::mem::take(&mut *Self::handles(handles)) + } +} + +impl TaskGroup { + /// Takes ownership of already-spawned tasks. + #[must_use] + pub fn new(handles: Vec>) -> Self { + Self(Arc::new(TaskGroupInner(Mutex::new(handles)))) + } + + /// Whether any task has ended on its own. + /// + /// These tasks run for the lifetime of the sequencer, so a finished one is a + /// task that panicked, and whatever it was doing is not happening any more. + #[must_use] + pub fn any_finished(&self) -> bool { + TaskGroupInner::handles(&self.0.0) + .iter() + .any(JoinHandle::is_finished) + } + + /// Stops every task and waits for it to finish. + /// + /// Returns only once the runtime has dropped each task's future, so whatever + /// they held (a store handle, a network client) is released by the time this + /// returns. Cancellation is the expected outcome, so it is not reported; a + /// panic is, since it means the task died on its own terms earlier. + pub async fn shutdown(&self) { + let handles = TaskGroupInner::take(&self.0.0); + for handle in handles { + handle.abort(); + if let Err(err) = handle.await + && err.is_panic() + { + warn!("Background task panicked before shutdown: {err}"); + } + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[tokio::test] + async fn a_task_that_ends_on_its_own_is_visible() { + let group = TaskGroup::new(vec![tokio::spawn(async {})]); + // A watcher only ends by panicking, so "finished" is the signal that a + // peer's deliveries have stopped happening. + tokio::task::yield_now().await; + assert!(group.any_finished()); + + let running = TaskGroup::new(vec![tokio::spawn(std::future::pending())]); + assert!(!running.any_finished()); + } + + #[tokio::test] + async fn shutdown_ends_a_task_that_would_never_end_on_its_own() { + let group = TaskGroup::new(vec![tokio::spawn(std::future::pending())]); + + // The watchers and the drive task are infinite loops, so awaiting one + // without cancelling it first hangs here for ever. + tokio::time::timeout(std::time::Duration::from_secs(5), group.shutdown()) + .await + .expect("shutdown must not hang on a task that never finishes by itself"); + + // Shutting down twice is a no-op rather than a second abort. + tokio::time::timeout(std::time::Duration::from_secs(5), group.shutdown()) + .await + .expect("a second shutdown must return immediately"); + } +} diff --git a/lez/sequencer/core/src/tests.rs b/lez/sequencer/core/src/tests.rs index 4da4059d..a7df9bad 100644 --- a/lez/sequencer/core/src/tests.rs +++ b/lez/sequencer/core/src/tests.rs @@ -4,7 +4,7 @@ use std::{pin::pin, time::Duration}; use common::{ HashType, - block::{BedrockStatus, HashableBlockData}, + block::{BedrockStatus, Block, HashableBlockData}, test_utils::sequencer_sign_key_for_testing, transaction::{LeeTransaction, clock_invocation}, }; @@ -29,23 +29,32 @@ use logos_blockchain_core::mantle::{ }; use logos_blockchain_zone_sdk::sequencer::DepositInfo; use mempool::MemPoolHandle; -use storage::sequencer::sequencer_cells::PendingDepositEventRecord; +use ping_core::{ReceiverInstruction, ping_record_pda}; +use storage::sequencer::sequencer_cells::{ + PendingCrossZoneDispatchRecord, PendingDepositEventRecord, +}; use tempfile::tempdir; use testnet_initial_state::{initial_pub_accounts_private_keys, initial_public_user_accounts}; use crate::{ - TransactionOrigin, apply_follow_update, + MAX_DISPATCHES_PER_BLOCK, RETIRE_DISPATCH_AFTER_FAILURES, TransactionOrigin, + apply_follow_update, block_publisher::FollowUpdate, block_store::SequencerStore, - build_bridge_deposit_tx_from_event, build_genesis_state, - config::{BedrockConfig, GenesisAction, SequencerConfig}, - deposit_already_minted, is_sequencer_only_program, + build_bridge_deposit_tx_from_event, build_genesis_state, classify_settled_deliveries, + config::{BedrockConfig, CrossZoneConfig, CrossZonePeer, GenesisAction, SequencerConfig}, + deposit_already_minted, dispatch_already_delivered, extract_cross_zone_dispatch, + extract_cross_zone_dispatch_key, is_sequencer_only_program, mock::{SequencerCoreWithMockClients, mock_checkpoint}, resubmittable_txs, }; mod reconstruction; +/// The peer zone a cross-zone test receives from. Distinct from the test +/// channel id (`[0; 32]`), which the inbox guest rejects as a source. +const PEER_ZONE: [u8; 32] = [0xbe_u8; 32]; + #[derive(borsh::BorshSerialize)] struct DepositMetadataForEncoding { recipient_id: lee::AccountId, @@ -162,6 +171,80 @@ fn tx_is_bridge_deposit( ) } +/// A config that receives `ping_receiver` messages from [`PEER_ZONE`], so +/// `build_genesis_state` seeds the inbox config PDA and a delivery has an +/// allowlist to pass. +fn cross_zone_test_config() -> SequencerConfig { + SequencerConfig { + cross_zone: Some(CrossZoneConfig { + peers: vec![CrossZonePeer { + channel_id: PEER_ZONE, + allowed_targets: vec![programs::ping_receiver().id()], + expected_block_signing_pubkey: None, + }], + }), + ..setup_sequencer_config() + } +} + +/// A `ping_receiver::Record` instruction as risc0 words, little-endian: the wire +/// form an emitter on the peer zone puts in the message payload. +fn ping_payload(payload: &[u8]) -> Vec { + risc0_zkvm::serde::to_vec(&ReceiverInstruction::Record { + payload: payload.to_vec(), + }) + .expect("ping instruction serializes") + .iter() + .flat_map(|word| word.to_le_bytes()) + .collect() +} + +/// The dispatch transaction for a message at index 0 of [`PEER_ZONE`] block +/// `src_block_id`. Built through the same builder the watcher uses, so a change +/// to the encoding shows up here rather than passing silently. +fn dispatch_tx(src_block_id: u64, payload: Vec) -> LeeTransaction { + let receiver_id = programs::ping_receiver().id(); + LeeTransaction::Public(cross_zone::build_dispatch_from_emission( + PEER_ZONE, + src_block_id, + 0, + programs::ping_sender().id(), + receiver_id, + &[ping_record_pda(receiver_id).into_value()], + payload, + )) +} + +/// The pending record the watcher would leave behind for that dispatch. +fn dispatch_record(src_block_id: u64, payload: Vec) -> PendingCrossZoneDispatchRecord { + let tx = dispatch_tx(src_block_id, payload); + PendingCrossZoneDispatchRecord::recorded( + cross_zone_inbox_core::message_key(&PEER_ZONE, src_block_id, 0), + borsh::to_vec(&tx).expect("dispatch encodes"), + ) +} + +/// The message keys of the deliveries a block carries. +fn dispatches_in(block: &Block) -> Vec<[u8; 32]> { + block + .body + .transactions + .iter() + .filter_map(extract_cross_zone_dispatch_key) + .collect() +} + +/// The pending dispatch records a sequencer still holds. +fn pending_dispatches( + sequencer: &SequencerCoreWithMockClients, +) -> Vec { + sequencer + .store + .dbio() + .get_pending_cross_zone_dispatches() + .expect("pending dispatches readable") +} + #[tokio::test] async fn start_from_config() { let config = setup_sequencer_config(); @@ -477,6 +560,363 @@ async fn a_replayed_deposit_mint_no_ops_in_the_guest() { ); } +#[tokio::test] +async fn recorded_dispatches_are_drained_from_the_store_on_production() { + let payload = b"hello-cross-zone".to_vec(); + let record = dispatch_record(7, ping_payload(&payload)); + let key = record.message_key; + + let (mut sequencer, _mempool_handle) = + SequencerCoreWithMockClients::start_from_config(cross_zone_test_config()).await; + assert_eq!( + sequencer + .store + .dbio() + .add_pending_cross_zone_dispatches(vec![record]) + .unwrap(), + 1 + ); + + // The delivery never goes through the mempool: the record is the queue, and + // production drains it. That is what makes the window between the watcher's + // durable read cursor and a block carrying the dispatch survivable. + assert!( + sequencer.mempool.pop().is_none(), + "deliveries are drained from the store, never queued in the mempool" + ); + + let block_id = sequencer.produce_new_block().await.unwrap(); + let block = sequencer + .store + .get_block_at_id(block_id) + .unwrap() + .expect("produced block is stored"); + assert_eq!( + dispatches_in(&block), + vec![key], + "the drained delivery should be included in the produced block" + ); + + let record_id = ping_record_pda(programs::ping_receiver().id()); + assert_eq!( + sequencer.with_state(|state| state.get_account_by_id(record_id).data.into_inner()), + payload, + "the dispatch must reach its target program, not just sit in the block" + ); + + // The record stays until the delivery finalizes; re-delivery is prevented by + // the inbox seen-set now in head state, not by any marker on the record. + assert_eq!( + pending_dispatches(&sequencer) + .iter() + .map(|record| record.message_key) + .collect::>(), + vec![key], + "the record remains until the delivery becomes irreversible" + ); +} + +#[tokio::test] +async fn a_delivered_dispatch_is_skipped_on_the_next_turn() { + // The seen-set is what replaces the submitted mark: the drain asks the state + // it is building on whether the inbox has already taken this message, so a + // record that outlives its delivery costs one skipped drain, not a replay. + let record = dispatch_record(11, ping_payload(b"once")); + let key = record.message_key; + + let (mut sequencer, _mempool_handle) = + SequencerCoreWithMockClients::start_from_config(cross_zone_test_config()).await; + sequencer + .store + .dbio() + .add_pending_cross_zone_dispatches(vec![record]) + .unwrap(); + + let first = sequencer.produce_new_block().await.unwrap(); + let second = sequencer.produce_new_block().await.unwrap(); + + let delivered_in = |block_id: u64| { + dispatches_in( + &sequencer + .store + .get_block_at_id(block_id) + .unwrap() + .expect("produced block is stored"), + ) + }; + assert_eq!(delivered_in(first), vec![key]); + assert!( + delivered_in(second).is_empty(), + "the inbox seen-set must keep the drain from re-delivering" + ); + + let message = extract_cross_zone_dispatch(&dispatch_tx(11, ping_payload(b"once"))) + .expect("the dispatch carries a cross-zone message"); + assert!( + sequencer.with_state(|state| dispatch_already_delivered(state, &message)), + "the seen shard in head state is what the skip reads" + ); +} + +#[tokio::test] +async fn a_dispatch_that_never_executes_is_given_up_on_after_repeated_failures() { + // A payload that is not `u32`-aligned: the inbox guest rejects it outright, + // so this is a delivery that can never execute however often it is retried. + // Its content is chosen on the peer zone and validated by nobody in between, + // so without a give-up policy it would fail on every block for ever. + let record = dispatch_record(13, b"odd".to_vec()); + + let (mut sequencer, _mempool_handle) = + SequencerCoreWithMockClients::start_from_config(cross_zone_test_config()).await; + sequencer + .store + .dbio() + .add_pending_cross_zone_dispatches(vec![record]) + .unwrap(); + + for attempt in 1..RETIRE_DISPATCH_AFTER_FAILURES { + let block_id = sequencer.produce_new_block().await.unwrap(); + let block = sequencer + .store + .get_block_at_id(block_id) + .unwrap() + .expect("produced block is stored"); + assert!( + dispatches_in(&block).is_empty(), + "a dispatch that fails to execute must not reach the block" + ); + + let records = pending_dispatches(&sequencer); + assert_eq!(records.len(), 1); + assert_eq!( + records[0].failed_attempts, attempt, + "the counter advances once per block, not once per process start" + ); + } + + // The attempt at the limit gives up on it, and giving up drops the record. + // Anything else leaves an entry no later block can ever remove, which is how + // a peer that can make deliveries fail would grow this list without bound. + sequencer.produce_new_block().await.unwrap(); + assert!( + pending_dispatches(&sequencer).is_empty(), + "giving up on a delivery must drop its record, not flag it" + ); + + // And nothing re-feeds it, so it stops costing a guest execution per block. + let block_id = sequencer.produce_new_block().await.unwrap(); + let block = sequencer.store.get_block_at_id(block_id).unwrap().unwrap(); + assert!(dispatches_in(&block).is_empty()); + assert!(pending_dispatches(&sequencer).is_empty()); +} + +#[tokio::test] +async fn a_redelivered_record_is_dropped_once_its_delivery_is_irreversible() { + // The watcher persists its floor only at slot boundaries, so a crash inside + // a slot makes the next run re-read it and re-record deliveries that have + // already settled. Their keys are in the inbox seen-set for good, so no + // future block will ever carry them and the settlement path cannot reach + // them. The drain dropping them is the only thing that does. + let record = dispatch_record(29, ping_payload(b"again")); + let key = record.message_key; + + let (mut sequencer, mempool_handle) = + SequencerCoreWithMockClients::start_from_config(cross_zone_test_config()).await; + sequencer + .store + .dbio() + .add_pending_cross_zone_dispatches(vec![record.clone()]) + .unwrap(); + + let block_id = sequencer.produce_new_block().await.unwrap(); + let delivery_block = sequencer.store.get_block_at_id(block_id).unwrap().unwrap(); + assert_eq!(dispatches_in(&delivery_block), vec![key]); + + apply_follow_update( + &sequencer.store.dbio(), + &sequencer.chain(), + &mempool_handle, + FollowUpdate { + finalized: vec![(MsgId::from(delivery_block.header.hash.0), delivery_block)], + ..empty_follow_update() + }, + ); + assert!(pending_dispatches(&sequencer).is_empty()); + + // The watcher re-reads the slot and records it again. + sequencer + .store + .dbio() + .add_pending_cross_zone_dispatches(vec![record]) + .unwrap(); + assert_eq!(pending_dispatches(&sequencer).len(), 1); + + let block_id = sequencer.produce_new_block().await.unwrap(); + let block = sequencer.store.get_block_at_id(block_id).unwrap().unwrap(); + assert!( + dispatches_in(&block).is_empty(), + "the delivery is already on the chain, so it must not be delivered again" + ); + assert!( + pending_dispatches(&sequencer).is_empty(), + "a record whose delivery is already irreversible must be dropped, not kept for ever" + ); +} + +#[tokio::test] +async fn a_delivery_still_reversible_keeps_its_record() { + // The counterpart to the test above, and the reason the drain checks two + // states rather than one. In head but not yet final means the delivery can + // still orphan, so skipping it is right but dropping its record would lose + // the delivery when it does. + let record = dispatch_record(31, ping_payload(b"pending")); + let key = record.message_key; + + let (mut sequencer, _mempool_handle) = + SequencerCoreWithMockClients::start_from_config(cross_zone_test_config()).await; + sequencer + .store + .dbio() + .add_pending_cross_zone_dispatches(vec![record]) + .unwrap(); + + sequencer.produce_new_block().await.unwrap(); + sequencer.produce_new_block().await.unwrap(); + + assert_eq!( + pending_dispatches(&sequencer) + .iter() + .map(|record| record.message_key) + .collect::>(), + vec![key], + "nothing has finalized, so the record must survive in case the block orphans" + ); +} + +#[test] +fn a_settled_delivery_that_is_not_the_one_we_recorded_is_reported() { + // The message key covers (src_zone, src_block_id, src_tx_index) and nothing + // about the payload, and so does the inbox's own replay check. So a peer's + // sequencer can publish a delivery under a key we hold with a payload we + // never saw, and it settles our correct record along with it. The indexer + // catches the forgery and halts; this record is the last local copy of what + // we believed, so the mismatch has to be reported before it is dropped. + let honest = dispatch_record(53, ping_payload(b"honest")); + let key = honest.message_key; + let forged = dispatch_tx(53, ping_payload(b"forged")); + assert_eq!( + extract_cross_zone_dispatch_key(&forged), + Some(key), + "the forged delivery must share the key, or it proves nothing" + ); + + let block = common::test_utils::produce_dummy_block(2, None, vec![forged]); + let (keys, mismatched) = classify_settled_deliveries(std::slice::from_ref(&honest), &block); + assert_eq!(keys, vec![key], "the record is settled either way"); + assert_eq!( + mismatched, + vec![key], + "a delivery that differs from the one recorded under that key must be reported" + ); + + // The honest case must stay quiet, or the report is noise. + let honest_block = common::test_utils::produce_dummy_block( + 2, + None, + vec![dispatch_tx(53, ping_payload(b"honest"))], + ); + let (keys, mismatched) = classify_settled_deliveries(&[honest], &honest_block); + assert_eq!(keys, vec![key]); + assert!(mismatched.is_empty()); +} + +#[tokio::test] +async fn a_delivery_too_large_for_any_block_does_not_stall_production() { + // A store-drained transaction is at the head of the queue every turn, so one + // that cannot fit in any block would defer itself for ever and, because the + // deferral breaks the loop, stop production ever reaching the mempool behind + // it. The peer chooses the payload, so this is theirs to trigger. + let record = dispatch_record(41, ping_payload(&[7_u8; 8192])); + + let mut config = cross_zone_test_config(); + config.max_block_size = bytesize::ByteSize::kib(4); + let (mut sequencer, mempool_handle) = + SequencerCoreWithMockClients::start_from_config(config).await; + sequencer + .store + .dbio() + .add_pending_cross_zone_dispatches(vec![record]) + .unwrap(); + + let user_tx = common::test_utils::create_transaction_native_token_transfer( + initial_public_user_accounts()[0].account_id, + 0, + initial_public_user_accounts()[1].account_id, + 10, + &create_signing_key_for_account1(), + ); + mempool_handle + .push((TransactionOrigin::User, user_tx.clone())) + .await + .unwrap(); + + // Production must get past it to the mempool in the very first block. + let block_id = sequencer.produce_new_block().await.unwrap(); + let block = sequencer.store.get_block_at_id(block_id).unwrap().unwrap(); + assert!( + block.body.transactions.contains(&user_tx), + "an oversized drained delivery must not stop production reaching the mempool" + ); + assert!(dispatches_in(&block).is_empty()); + + // And it is given up on rather than retried for ever. + for _ in 1..RETIRE_DISPATCH_AFTER_FAILURES { + sequencer.produce_new_block().await.unwrap(); + } + assert!( + pending_dispatches(&sequencer).is_empty(), + "a delivery that fits in no block must be given up on" + ); +} + +#[tokio::test] +async fn a_delivery_backlog_is_spread_across_blocks() { + // Each delivery costs a guest execution and peers decide how many queue up, + // so an unbounded drain would let a backlog decide how long a block takes to + // build and leave no room for user work, since store-drained transactions + // are taken before the mempool. + let backlog = MAX_DISPATCHES_PER_BLOCK + 3; + let records: Vec<_> = (0..backlog) + .map(|index| { + let src_block_id = 100 + u64::try_from(index).expect("test index fits"); + dispatch_record(src_block_id, ping_payload(b"backlog")) + }) + .collect(); + + let mut config = cross_zone_test_config(); + config.max_num_tx_in_block = backlog + 10; + let (mut sequencer, _mempool_handle) = + SequencerCoreWithMockClients::start_from_config(config).await; + sequencer + .store + .dbio() + .add_pending_cross_zone_dispatches(records) + .unwrap(); + + let block_id = sequencer.produce_new_block().await.unwrap(); + let block = sequencer.store.get_block_at_id(block_id).unwrap().unwrap(); + assert_eq!( + dispatches_in(&block).len(), + MAX_DISPATCHES_PER_BLOCK, + "one block must not carry an unbounded number of deliveries" + ); + + // Deferred, not dropped: the rest go in the next block. + let block_id = sequencer.produce_new_block().await.unwrap(); + let block = sequencer.store.get_block_at_id(block_id).unwrap().unwrap(); + assert_eq!(dispatches_in(&block).len(), 3); +} + #[test] fn transaction_pre_check_pass() { let tx = common::test_utils::produce_dummy_empty_transaction(); @@ -1774,6 +2214,98 @@ async fn follow_finalized_own_block_moves_final_tier_and_marks_store() { assert!(matches!(stored.bedrock_status, BedrockStatus::Finalized)); } +#[tokio::test] +async fn follow_finalized_delivery_drops_its_pending_record() { + // The record exists to bridge the gap between the watcher's durable read + // cursor and a block that carries the delivery. Once that block is + // irreversible the delivery cannot be lost any more, so the record is owed + // nothing and goes with the same update that made the block irreversible. + let record = dispatch_record(17, ping_payload(b"settled")); + let key = record.message_key; + + let (mut sequencer, mempool_handle) = + SequencerCoreWithMockClients::start_from_config(cross_zone_test_config()).await; + sequencer + .store + .dbio() + .add_pending_cross_zone_dispatches(vec![record]) + .unwrap(); + + let block_id = sequencer.produce_new_block().await.unwrap(); + let delivery_block = sequencer.store.get_block_at_id(block_id).unwrap().unwrap(); + assert_eq!(dispatches_in(&delivery_block), vec![key]); + assert_eq!( + pending_dispatches(&sequencer).len(), + 1, + "including the delivery is not enough to settle its record" + ); + + apply_follow_update( + &sequencer.store.dbio(), + &sequencer.chain(), + &mempool_handle, + FollowUpdate { + finalized: vec![(MsgId::from(delivery_block.header.hash.0), delivery_block)], + ..empty_follow_update() + }, + ); + + assert!( + pending_dispatches(&sequencer).is_empty(), + "a delivery in an irreversible block settles its record" + ); +} + +#[tokio::test] +async fn a_parked_finalized_block_does_not_drop_a_dispatch_record() { + // Keyed by message key, not by height: a finalized block the final tier + // parks never became irreversible, so nothing it happens to sit above may + // settle a record. Dropping one here would lose the delivery for good, since + // the watcher's floor has already moved past the peer block it came from. + let record = dispatch_record(19, ping_payload(b"parked")); + let key = record.message_key; + let delivery = dispatch_tx(19, ping_payload(b"parked")); + + let (mut sequencer, mempool_handle) = + SequencerCoreWithMockClients::start_from_config(cross_zone_test_config()).await; + sequencer + .store + .dbio() + .add_pending_cross_zone_dispatches(vec![record]) + .unwrap(); + + let tx = common::test_utils::produce_dummy_empty_transaction(); + mempool_handle + .push((TransactionOrigin::User, tx)) + .await + .unwrap(); + sequencer.produce_new_block().await.unwrap(); + + // A skip-ahead block carrying the same delivery: not in head and linking to + // nothing we hold, so the final tier parks it instead of applying it. + let parked = + common::test_utils::produce_dummy_block(9, Some(HashType([44; 32])), vec![delivery]); + + apply_follow_update( + &sequencer.store.dbio(), + &sequencer.chain(), + &mempool_handle, + FollowUpdate { + finalized: vec![(MsgId::from([9; 32]), parked)], + ..empty_follow_update() + }, + ); + + assert_eq!( + pending_dispatches(&sequencer) + .iter() + .map(|record| record.message_key) + .collect::>(), + vec![key], + "a parked finalized block must not drop its delivery's record" + ); +} + #[tokio::test] async fn follow_finalized_backfill_block_is_applied_and_marked_finalized() { let config = setup_sequencer_config(); diff --git a/lez/sequencer/core/src/tests/reconstruction.rs b/lez/sequencer/core/src/tests/reconstruction.rs index 47d84c06..1a23775a 100644 --- a/lez/sequencer/core/src/tests/reconstruction.rs +++ b/lez/sequencer/core/src/tests/reconstruction.rs @@ -586,6 +586,146 @@ async fn reconstruction_reconciles_already_finished_deposit() { ); } +/// A cross-zone delivery whose record is still pending locally, but whose block +/// arrives already finalized on the channel. Reconstruction must settle the +/// record on the way through: the delivery is permanently reflected in the +/// reconstructed state (the inbox seen shard), so the next production neither +/// re-delivers it nor leaves a record nothing will ever drop. +#[tokio::test] +async fn reconstructed_delivery_settles_its_pending_record() { + let payload = b"reconstructed".to_vec(); + let record = dispatch_record(23, ping_payload(&payload)); + let key = record.message_key; + + // Sequencer A produces the block that carries the delivery. + let config_a = cross_zone_test_config(); + let (mut seq_a, _mempool_a) = + SequencerCoreWithMockClients::start_from_config(config_a.clone()).await; + seq_a + .block_store() + .dbio() + .add_pending_cross_zone_dispatches(vec![record.clone()]) + .unwrap(); + seq_a.produce_new_block().await.unwrap(); + + let tip_a = seq_a.block_store().latest_block_meta().unwrap().unwrap(); + let messages = channel_from_store(seq_a.block_store(), 10); + let tip_slot = messages.last().unwrap().1; + let channel_id = config_a.bedrock_config.channel_id; + + // Sequencer B holds the same record, as its own watcher would after reading + // the peer block, and reconstructs A's chain from a fresh store. + let (mut seq_b, _mempool_b) = + SequencerCoreWithMockClients::start_from_config(cross_zone_test_config()).await; + assert_eq!( + seq_b + .block_store() + .dbio() + .add_pending_cross_zone_dispatches(vec![record]) + .unwrap(), + 1 + ); + + let mock_b = MockBlockPublisher::with_canned_channel(channel_id, Some(tip_slot), messages); + SequencerCore::::verify_and_reconstruct( + &mock_b, + &seq_b.store, + &seq_b.chain, + true, + ) + .await + .expect("reconstruct"); + + let tip_b = seq_b.block_store().latest_block_meta().unwrap().unwrap(); + assert_eq!(tip_b.id, tip_a.id); + assert_eq!(tip_b.hash, tip_a.hash); + assert!( + seq_b + .block_store() + .dbio() + .get_pending_cross_zone_dispatches() + .unwrap() + .is_empty(), + "reconstruction must settle the record of a delivery it replayed" + ); + + // The delivery landed exactly once, and the next turn does not re-emit it. + let record_id = ping_record_pda(programs::ping_receiver().id()); + assert_eq!( + seq_b.with_state(|state| state.get_account_by_id(record_id).data.into_inner()), + payload, + "the reconstructed delivery must reach its target program" + ); + seq_b.produce_new_block().await.unwrap(); + let produced = seq_b + .block_store() + .get_block_at_id(tip_b.id + 1) + .unwrap() + .expect("produced block present"); + assert!( + !dispatches_in(&produced).contains(&key), + "the reconstructed delivery must not be re-emitted" + ); +} + +/// A delivery this node published itself, served back by the channel at or below +/// its own tip. That path verifies the block matches and returns early, so it is +/// reached on every restart. It must still settle the delivery's record: the +/// channel serving the block is what makes it irreversible, and nothing later +/// will ever put that key in a block again. +#[tokio::test] +async fn a_verified_own_block_settles_its_delivery_records() { + let record = dispatch_record(37, ping_payload(b"verified")); + let key = record.message_key; + + let (mut seq, _mempool) = + SequencerCoreWithMockClients::start_from_config(cross_zone_test_config()).await; + seq.block_store() + .dbio() + .add_pending_cross_zone_dispatches(vec![record]) + .unwrap(); + + let block_id = seq.produce_new_block().await.unwrap(); + let block = seq + .block_store() + .get_block_at_id(block_id) + .unwrap() + .unwrap(); + assert_eq!(dispatches_in(&block), vec![key]); + assert_eq!( + seq.block_store() + .dbio() + .get_pending_cross_zone_dispatches() + .unwrap() + .len(), + 1, + "producing the block is not what settles the record" + ); + + // The channel serves our own chain back, tip included. + let messages = channel_from_store(seq.block_store(), 10); + let tip_slot = messages.last().unwrap().1; + let mock = MockBlockPublisher::with_canned_channel( + seq.sequencer_config.bedrock_config.channel_id, + Some(tip_slot), + messages, + ); + SequencerCore::::verify_and_reconstruct( + &mock, &seq.store, &seq.chain, true, + ) + .await + .expect("reconstruct"); + + assert!( + seq.block_store() + .dbio() + .get_pending_cross_zone_dispatches() + .unwrap() + .is_empty(), + "a delivery the channel confirms must not leave a record nothing can remove" + ); +} + #[tokio::test] async fn committed_local_against_missing_channel_fails_without_anchor() { // A sequencer that has committed blocks — a non-genesis tip plus a persisted diff --git a/lez/sequencer/service/src/lib.rs b/lez/sequencer/service/src/lib.rs index a8aa856e..ba5b68ec 100644 --- a/lez/sequencer/service/src/lib.rs +++ b/lez/sequencer/service/src/lib.rs @@ -12,7 +12,11 @@ use sequencer_core::SequencerCore; #[cfg(feature = "standalone")] use sequencer_core::SequencerCoreWithMockClients as SequencerCore; pub use sequencer_core::config::*; -use sequencer_core::{TransactionOrigin, block_publisher::BlockPublisherTrait as _}; +use sequencer_core::{ + TransactionOrigin, + block_publisher::BlockPublisherTrait as _, + task_group::{StoreRelease, TaskGroup}, +}; use sequencer_service_rpc::RpcServer as _; use tokio::{sync::Mutex, task::JoinHandle}; use tokio_util::sync::CancellationToken; @@ -26,12 +30,19 @@ const REQUEST_BODY_MAX_SIZE: ByteSize = ByteSize::mib(10); /// Implements `Drop` to ensure all tasks are aborted and the RPC server is stopped when dropped. pub struct SequencerHandle { addr: SocketAddr, - /// Option because of `Drop` which forbids to simply move out of `self` in `stopped()`. - server_handle: Option, + server_handle: ServerHandle, main_loop_handle: JoinHandle>, /// 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, + /// 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 { @@ -40,29 +51,73 @@ impl SequencerHandle { server_handle: ServerHandle, main_loop_handle: JoinHandle>, driver_cancellation: CancellationToken, + background_tasks: Vec, + store: StoreRelease, ) -> Self { Self { addr, - server_handle: Some(server_handle), + 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}"); + } + + for tasks in &self.background_tasks { + tasks.shutdown().await; + } + + if let Err(err) = self.server_handle.stop() { + error!("An error occurred while stopping Sequencer RPC server: {err}"); + } + self.server_handle.clone().stopped().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; + } + /// Wait for any of the sequencer tasks to fail and return the error. #[expect( clippy::integer_division_remainder_used, reason = "Generated by select! macro, can't be easily rewritten to avoid this lint" )] - pub async fn failed(mut self) -> Result { + pub async fn failed(&mut self) -> Result { let Self { addr: _, server_handle, main_loop_handle, driver_cancellation, - } = &mut self; + background_tasks: _, + store: _, + } = self; - let server_handle = server_handle.take().expect("Server handle is set"); + // 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")) @@ -89,11 +144,16 @@ impl SequencerHandle { server_handle, main_loop_handle, driver_cancellation, + background_tasks, + store: _, } = self; - let stopped = server_handle.as_ref().is_none_or(ServerHandle::is_stopped) + let stopped = server_handle.is_stopped() || main_loop_handle.is_finished() - || driver_cancellation.is_cancelled(); + || 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 } @@ -110,20 +170,46 @@ impl Drop for SequencerHandle { server_handle, main_loop_handle, driver_cancellation: _, + background_tasks: _, + store: _, } = self; main_loop_handle.abort(); - let Some(handle) = server_handle else { - return; - }; - - if let Err(err) = handle.stop() { + 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 { let block_timeout = config.block_create_timeout; let max_block_size = config.max_block_size; @@ -134,6 +220,11 @@ pub async fn run(config: SequencerConfig, listen_addr: SocketAddr) -> Result Result Result<()> { home, } = Args::parse(); - // TODO: handle this cancellation token more gracefully within Sequencer service - // similar to how we do in Indexer let cancellation_token = listen_for_shutdown_signal(); let mut config = sequencer_service::SequencerConfig::from_path(&config_path)?; if let Some(home) = home { config.home = home; } - let sequencer_handle = + let mut sequencer_handle = sequencer_service::run(config, SocketAddr::new(listen_address, port)).await?; tokio::select! { @@ -60,21 +59,50 @@ async fn main() -> Result<()> { } } + // Stop the watchers, the publisher's drive task, the block loop and the RPC + // server, and wait for each. Dropping the handle only asks; the store stays + // open for an unbounded stretch after that, so a restart can find its own + // home directory locked, and a watcher can be killed between recording a + // delivery and handing it over. + sequencer_handle.shutdown().await; + info!("Sequencer shutdown complete"); Ok(()) } +/// Cancelled on Ctrl-C or `SIGTERM`. +/// +/// `SIGTERM` is what a container runtime sends first, so without it every +/// orchestrated stop is the ungraceful path. +#[expect( + clippy::integer_division_remainder_used, + reason = "Generated by select! macro, can't be easily rewritten to avoid this lint" +)] fn listen_for_shutdown_signal() -> CancellationToken { let cancellation_token = CancellationToken::new(); let cancellation_token_clone = cancellation_token.clone(); tokio::spawn(async move { - if let Err(err) = tokio::signal::ctrl_c().await { - error!("Failed to listen for Ctrl-C signal: {err}"); - return; + let mut terminate = match signal(SignalKind::terminate()) { + Ok(terminate) => terminate, + Err(err) => { + error!("Failed to listen for SIGTERM: {err}"); + return; + } + }; + + tokio::select! { + result = tokio::signal::ctrl_c() => match result { + Ok(()) => info!("Received Ctrl-C signal"), + Err(err) => { + error!("Failed to listen for Ctrl-C signal: {err}"); + return; + } + }, + _ = terminate.recv() => info!("Received SIGTERM"), } - info!("Received Ctrl-C signal"); + cancellation_token_clone.cancel(); }); diff --git a/lez/storage/src/sequencer/mod.rs b/lez/storage/src/sequencer/mod.rs index 28a9f65e..a559085a 100644 --- a/lez/storage/src/sequencer/mod.rs +++ b/lez/storage/src/sequencer/mod.rs @@ -1,4 +1,8 @@ -use std::{collections::BTreeMap, path::Path, sync::Arc}; +use std::{ + collections::BTreeMap, + path::Path, + sync::{Arc, Mutex, MutexGuard, PoisonError}, +}; use borsh::{BorshDeserialize, BorshSerialize}; use common::{ @@ -18,7 +22,9 @@ use crate::{ sequencer::sequencer_cells::{ FinalBlockMetaCellOwned, FinalBlockMetaCellRef, FinalLeeStateCellOwned, FinalLeeStateCellRef, LEEStateCellOwned, LEEStateCellRef, LastFinalizedBlockIdCell, - LatestBlockMetaCellOwned, LatestBlockMetaCellRef, PendingDepositEventRecord, + LatestBlockMetaCellOwned, LatestBlockMetaCellRef, PeerFloorCellOwned, PeerFloorCellRef, + PeerZoneKey, PendingCrossZoneDispatchRecord, PendingCrossZoneDispatchesCellOwned, + PendingCrossZoneDispatchesCellRef, PendingDepositEventRecord, PendingDepositEventsCellOwned, PendingDepositEventsCellRef, UnseenWithdrawCountCell, WithdrawalReconciliationKey, ZoneAnchorCell, ZoneAnchorRecord, ZoneSdkCheckpointCellOwned, ZoneSdkCheckpointCellRef, @@ -40,9 +46,25 @@ pub const DB_META_ZONE_CURSOR_KEY: &str = "zone_cursor"; /// Key base for storing queued deposit events that were not yet /// fulfilled on L2. pub const DB_META_PENDING_DEPOSIT_EVENTS_KEY: &str = "pending_deposit_events"; +/// Key base for storing a cross-zone watcher's delivery floor on one peer +/// channel (opaque bytes). Keyed per peer zone. +pub const DB_META_CROSS_ZONE_PEER_FLOOR_KEY: &str = "cross_zone_peer_floor"; +/// Key base for storing cross-zone deliveries the watcher has recorded but +/// which are not yet known to be irreversibly delivered. +pub const DB_META_PENDING_CROSS_ZONE_DISPATCHES_KEY: &str = "pending_cross_zone_dispatches"; + /// Key base for counting unseen L2 withdraw intents. pub const DB_META_UNSEEN_WITHDRAW_COUNT_KEY: &str = "unseen_withdraw_count"; +/// How many cross-zone deliveries may be pending at once. +/// +/// The whole list is a single value, read on every block and rewritten on every +/// change, and what fills it is chosen by peer zones rather than by us. Refusing +/// to record past this bound turns "a peer decides how large our store gets" +/// into "a peer's messages wait", since a watcher that cannot record holds its +/// delivery floor and reads the slot again later. +pub const MAX_PENDING_CROSS_ZONE_DISPATCHES: usize = 4096; + /// Key base for storing the LEE state. pub const DB_LEE_STATE_KEY: &str = "lee_state"; /// Key base for storing the LEE state at the last L1-finalized block. @@ -123,6 +145,8 @@ pub struct StoreUpdate<'update> { pub new_deposit_events: &'update [PendingDepositEventRecord], /// Deposit op ids whose mint finalized: their pending records are dropped. pub remove_deposit_records: &'update [HashType], + /// Message keys whose delivery finalized: their pending records are dropped. + pub remove_dispatch_records: &'update [[u8; 32]], /// L1 withdraw events to reconcile against the local unseen counters. pub consumed_withdrawals: &'update [WithdrawalReconciliationKey], /// L2 withdraw intents this update raises, awaiting their L1 event. @@ -146,6 +170,7 @@ impl<'update> StoreUpdate<'update> { finalized_up_to: None, new_deposit_events: &[], remove_deposit_records: &[], + remove_dispatch_records: &[], consumed_withdrawals: &[], new_withdraw_intents: &[], zone_anchor: None, @@ -165,8 +190,21 @@ pub struct StoreUpdateOutcome { pub unmatched_withdrawals: Vec, } +#[expect( + clippy::partial_pub_fields, + reason = "the pending-record lock is an implementation detail and must stay private" +)] pub struct RocksDBIO { pub db: DBWithThreadMode, + /// Serializes the read-modify-write cycles over the pending cross-zone + /// dispatch list. + /// + /// The list is a single value holding the whole `Vec`, and three tasks + /// rewrite it: the watcher recording a delivery, the production loop + /// counting a failed attempt, and the publisher's drive task settling + /// finalized deliveries. Rocksdb makes the write atomic, not the cycle, so + /// without this the writer that read first silently drops the others. + pending_records: Mutex<()>, } impl DBIO for RocksDBIO { @@ -176,6 +214,18 @@ impl DBIO for RocksDBIO { } impl RocksDBIO { + /// Held across a pending-record read-modify-write. See + /// [`RocksDBIO::pending_records`]. + /// + /// A poisoned lock is recovered rather than propagated: the records behind + /// it are a plain `Vec` that a panicking writer cannot leave half-written, + /// since the write is a single rocksdb put. + fn lock_pending_records(&self) -> MutexGuard<'_, ()> { + self.pending_records + .lock() + .unwrap_or_else(PoisonError::into_inner) + } + pub fn open(path: &Path) -> DbResult { let db_opts = Options::default(); Self::open_inner(path, &db_opts) @@ -289,7 +339,10 @@ impl RocksDBIO { additional_info: Some("Failed to open or create DB".to_owned()), })?; - let dbio = Self { db }; + let dbio = Self { + db, + pending_records: Mutex::new(()), + }; Ok(dbio) } @@ -539,6 +592,185 @@ impl RocksDBIO { Ok(accepted) } + /// One cross-zone watcher's delivery floor on `peer_zone`'s channel, or + /// `None` before it has delivered anything from that peer. + pub fn get_cross_zone_peer_floor_bytes( + &self, + peer_zone: PeerZoneKey, + ) -> DbResult>> { + Ok(self + .get_opt::(peer_zone)? + .map(|cell| cell.0)) + } + + pub fn put_cross_zone_peer_floor_bytes( + &self, + peer_zone: PeerZoneKey, + bytes: &[u8], + ) -> DbResult<()> { + self.put(&PeerFloorCellRef(bytes), peer_zone) + } + + pub fn get_pending_cross_zone_dispatches( + &self, + ) -> DbResult> { + Ok(self + .get_opt::(())? + .map_or_else(Vec::new, |cell| cell.0)) + } + + fn put_pending_cross_zone_dispatches( + &self, + records: &[PendingCrossZoneDispatchRecord], + ) -> DbResult<()> { + self.put(&PendingCrossZoneDispatchesCellRef(records), ()) + } + + fn put_pending_cross_zone_dispatches_batch( + &self, + records: &[PendingCrossZoneDispatchRecord], + batch: &mut WriteBatch, + ) -> DbResult<()> { + self.put_batch(&PendingCrossZoneDispatchesCellRef(records), (), batch) + } + + /// Records every delivery one peer block carries, in a single write. + /// + /// Returns how many were new. Ones already recorded are skipped, so a slot + /// the watcher re-reads is not double-tracked. + /// + /// Batched rather than one call per delivery because the whole list is one + /// value: recording a block's messages one at a time rewrites the list once + /// per message, which is quadratic in a block that carries many. + /// + /// Fails without writing anything if the list would exceed + /// [`MAX_PENDING_CROSS_ZONE_DISPATCHES`]. The caller's floor then stays put + /// and the slot is read again later, which is the difference between + /// backpressure and an unbounded list a peer controls the size of. + pub fn add_pending_cross_zone_dispatches( + &self, + dispatches: Vec, + ) -> DbResult { + if dispatches.is_empty() { + return Ok(0); + } + + let _pending = self.lock_pending_records(); + let mut records = self.get_pending_cross_zone_dispatches()?; + let before = records.len(); + + for dispatch in dispatches { + if records + .iter() + .any(|record| record.message_key == dispatch.message_key) + { + continue; + } + records.push(dispatch); + } + + let accepted = records.len().saturating_sub(before); + if accepted == 0 { + return Ok(0); + } + if records.len() > MAX_PENDING_CROSS_ZONE_DISPATCHES { + return Err(DbError::db_interaction_error(format!( + "Refusing to hold more than {MAX_PENDING_CROSS_ZONE_DISPATCHES} pending cross-zone deliveries; {before} already pending" + ))); + } + + self.put_pending_cross_zone_dispatches(&records)?; + Ok(accepted) + } + + /// Counts a failed production attempt against a delivery, dropping its + /// record once it reaches `retire_at`. Returns whether it was dropped. + /// + /// Dropped rather than flagged: a retired record is one the drain will never + /// turn into a block transaction again, so nothing would ever remove it, and + /// a peer that can make deliveries fail could grow the list without bound. + /// The delivery is given up on either way; this way the cost is a log line + /// rather than a permanent entry. + /// + /// A delivery with no record is already retired as far as this is concerned: + /// there is nothing left to count against. + pub fn record_dispatch_failure(&self, message_key: [u8; 32], retire_at: u32) -> DbResult { + let _pending = self.lock_pending_records(); + let mut records = self.get_pending_cross_zone_dispatches()?; + let Some(position) = records + .iter() + .position(|record| record.message_key == message_key) + else { + return Ok(true); + }; + + let attempts = { + let record = &mut records[position]; + record.failed_attempts = record.failed_attempts.saturating_add(1); + record.failed_attempts + }; + let retired = attempts >= retire_at; + if retired { + records.remove(position); + } + self.put_pending_cross_zone_dispatches(&records)?; + Ok(retired) + } + + /// Drops the records of deliveries that are settled for good, outside any + /// store update. + /// + /// The settlement path in [`Self::store_update`] catches a delivery as its + /// block becomes irreversible. This catches the ones that path cannot: a + /// record re-added after its delivery had already settled, which the watcher + /// does whenever it re-reads a slot it has already consumed. Nothing would + /// ever put such a key in a block again, so without this it stays for ever. + pub fn drop_settled_cross_zone_dispatches(&self, message_keys: &[[u8; 32]]) -> DbResult { + if message_keys.is_empty() { + return Ok(0); + } + + let _pending = self.lock_pending_records(); + let to_remove: std::collections::HashSet<&[u8; 32]> = message_keys.iter().collect(); + let mut records = self.get_pending_cross_zone_dispatches()?; + let before = records.len(); + records.retain(|record| !to_remove.contains(&record.message_key)); + let removed = before.saturating_sub(records.len()); + + if removed > 0 { + self.put_pending_cross_zone_dispatches(&records)?; + } + Ok(removed) + } + + /// Drops the pending records of deliveries that just became irreversible, + /// staged into `batch` so they go with the update that made them so. + /// + /// Removal only, unlike [`Self::stage_pending_deposit_events`]: a delivery is + /// recorded by the watcher through + /// [`Self::add_pending_cross_zone_dispatch`], on its own task and outside + /// any store update, so nothing ever adds one here. + fn stage_removed_dispatches( + &self, + remove_keys: &[[u8; 32]], + batch: &mut WriteBatch, + ) -> DbResult { + if remove_keys.is_empty() { + return Ok(0); + } + + let to_remove: std::collections::HashSet<&[u8; 32]> = remove_keys.iter().collect(); + let mut records = self.get_pending_cross_zone_dispatches()?; + let before = records.len(); + records.retain(|record| !to_remove.contains(&record.message_key)); + let removed = before.saturating_sub(records.len()); + + if removed > 0 { + self.put_pending_cross_zone_dispatches_batch(&records, batch)?; + } + Ok(removed) + } + /// Stages the unseen-withdraw decrements for one update into `batch`, /// returning one entry per occurrence that matched no local counter. /// @@ -875,6 +1107,7 @@ impl RocksDBIO { /// one, most carry nothing else) must not drag a full state serialization /// with it. pub fn store_update(&self, update: &StoreUpdate<'_>) -> DbResult { + let _pending = self.lock_pending_records(); let StoreUpdate { checkpoint, blocks, @@ -884,6 +1117,7 @@ impl RocksDBIO { finalized_up_to, new_deposit_events, remove_deposit_records, + remove_dispatch_records, consumed_withdrawals, new_withdraw_intents, zone_anchor, @@ -940,6 +1174,7 @@ impl RocksDBIO { remove_deposit_records, &mut batch, )?; + self.stage_removed_dispatches(remove_dispatch_records, &mut batch)?; let unmatched_withdrawals = self.stage_consumed_withdrawals(consumed_withdrawals, &mut batch)?; self.stage_new_withdraw_intents(new_withdraw_intents, &mut batch)?; diff --git a/lez/storage/src/sequencer/sequencer_cells.rs b/lez/storage/src/sequencer/sequencer_cells.rs index 8a4223c0..9f3ae9eb 100644 --- a/lez/storage/src/sequencer/sequencer_cells.rs +++ b/lez/storage/src/sequencer/sequencer_cells.rs @@ -8,7 +8,8 @@ use crate::{ error::DbError, sequencer::{ CF_LEE_STATE_NAME, DB_FINAL_BLOCK_META_KEY, DB_FINAL_LEE_STATE_KEY, DB_LEE_STATE_KEY, - DB_META_LAST_FINALIZED_BLOCK_ID, DB_META_LATEST_BLOCK_META_KEY, + DB_META_CROSS_ZONE_PEER_FLOOR_KEY, DB_META_LAST_FINALIZED_BLOCK_ID, + DB_META_LATEST_BLOCK_META_KEY, DB_META_PENDING_CROSS_ZONE_DISPATCHES_KEY, DB_META_PENDING_DEPOSIT_EVENTS_KEY, DB_META_UNSEEN_WITHDRAW_COUNT_KEY, DB_META_ZONE_CURSOR_KEY, DB_META_ZONE_SDK_CHECKPOINT_KEY, }, @@ -245,6 +246,82 @@ pub struct PendingDepositEventRecord { pub metadata: Vec, } +/// A cross-zone delivery the watcher has read off a peer block but which is not +/// yet known to be irreversibly delivered. +/// +/// The watcher's delivery floor is durable, so once it advances past a peer +/// block that block is never re-read. This record is what stands in its place: +/// block production drains it every turn, and it survives a restart. Mirrors +/// [`PendingDepositEventRecord`], which solves the same problem for deposits, +/// and like it carries no "submitted" mark: the record is dropped when the +/// delivery itself finalizes, and re-including one meanwhile is harmless +/// because the inbox no-ops a replay on chain. +#[derive(Debug, Clone, PartialEq, Eq, BorshSerialize, BorshDeserialize)] +pub struct PendingCrossZoneDispatchRecord { + /// Content-addressed replay key of the delivered message, and this record's + /// identity. + pub message_key: [u8; 32], + /// The borsh-encoded dispatch transaction, so production can re-feed it + /// without re-reading the peer channel. + pub transaction: Vec, + /// Production attempts that ended in an execution failure. + /// + /// A dispatch's payload and target accounts are chosen on the peer zone and + /// validated by nobody in between, so one can fail for good. A failure can + /// equally be a property of the moment, so a single one is not enough to + /// give up on a delivery. Once too many accumulate the record is dropped + /// rather than flagged, since a delivery nothing will retry is also a + /// delivery nothing would ever remove. + pub failed_attempts: u32, +} + +impl PendingCrossZoneDispatchRecord { + /// A delivery the watcher has just read: never attempted. + #[must_use] + pub const fn recorded(message_key: [u8; 32], transaction: Vec) -> Self { + Self { + message_key, + transaction, + failed_attempts: 0, + } + } +} + +#[derive(BorshDeserialize)] +pub struct PendingCrossZoneDispatchesCellOwned(pub Vec); + +impl SimpleStorableCell for PendingCrossZoneDispatchesCellOwned { + type KeyParams = (); + + const CELL_NAME: &'static str = DB_META_PENDING_CROSS_ZONE_DISPATCHES_KEY; + const CF_NAME: &'static str = CF_META_NAME; +} + +impl SimpleReadableCell for PendingCrossZoneDispatchesCellOwned {} + +#[derive(BorshSerialize)] +pub struct PendingCrossZoneDispatchesCellRef<'records>( + pub &'records [PendingCrossZoneDispatchRecord], +); + +impl SimpleStorableCell for PendingCrossZoneDispatchesCellRef<'_> { + type KeyParams = (); + + const CELL_NAME: &'static str = DB_META_PENDING_CROSS_ZONE_DISPATCHES_KEY; + const CF_NAME: &'static str = CF_META_NAME; +} + +impl SimpleWritableCell for PendingCrossZoneDispatchesCellRef<'_> { + fn value_constructor(&self) -> DbResult> { + borsh::to_vec(&self).map_err(|err| { + DbError::borsh_cast_message( + err, + Some("Failed to serialize pending cross-zone dispatches cell".to_owned()), + ) + }) + } +} + #[derive(BorshDeserialize)] pub struct PendingDepositEventsCellOwned(pub Vec); @@ -278,6 +355,72 @@ impl SimpleWritableCell for PendingDepositEventsCellRef<'_> { } } +/// Identifies which peer channel a cross-zone watcher cursor belongs to. The +/// 32-byte peer channel id doubles as the peer's zone id. +pub type PeerZoneKey = [u8; 32]; + +/// Opaque bytes for one peer's cross-zone read cursor. As with the zone-sdk +/// checkpoint, the caller owns the encoding, since the cursor type derives serde +/// rather than borsh. +#[derive(BorshDeserialize)] +pub struct PeerFloorCellOwned(pub Vec); + +impl SimpleStorableCell for PeerFloorCellOwned { + type KeyParams = PeerZoneKey; + + const CELL_NAME: &'static str = DB_META_CROSS_ZONE_PEER_FLOOR_KEY; + const CF_NAME: &'static str = CF_META_NAME; + + /// Folds the peer zone into the key so each peer keeps its own cursor. + fn key_constructor(peer_zone: Self::KeyParams) -> DbResult> { + borsh::to_vec(&(Self::CELL_NAME, peer_zone)).map_err(|err| { + DbError::borsh_cast_message( + err, + Some(format!( + "Failed to serialize {:?} key params", + Self::CELL_NAME + )), + ) + }) + } +} + +impl SimpleReadableCell for PeerFloorCellOwned {} + +#[derive(BorshSerialize)] +pub struct PeerFloorCellRef<'bytes>(pub &'bytes [u8]); + +impl SimpleStorableCell for PeerFloorCellRef<'_> { + type KeyParams = PeerZoneKey; + + const CELL_NAME: &'static str = DB_META_CROSS_ZONE_PEER_FLOOR_KEY; + const CF_NAME: &'static str = CF_META_NAME; + + /// Folds the peer zone into the key so each peer keeps its own cursor. + fn key_constructor(peer_zone: Self::KeyParams) -> DbResult> { + borsh::to_vec(&(Self::CELL_NAME, peer_zone)).map_err(|err| { + DbError::borsh_cast_message( + err, + Some(format!( + "Failed to serialize {:?} key params", + Self::CELL_NAME + )), + ) + }) + } +} + +impl SimpleWritableCell for PeerFloorCellRef<'_> { + fn value_constructor(&self) -> DbResult> { + borsh::to_vec(&self).map_err(|err| { + DbError::borsh_cast_message( + err, + Some("Failed to serialize cross-zone peer floor cell".to_owned()), + ) + }) + } +} + #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] pub struct WithdrawalReconciliationKey { pub amount: u64, diff --git a/lez/storage/src/sequencer/tests.rs b/lez/storage/src/sequencer/tests.rs index b50a5591..8f0f1ee6 100644 --- a/lez/storage/src/sequencer/tests.rs +++ b/lez/storage/src/sequencer/tests.rs @@ -37,6 +37,17 @@ fn deposit_record(seed: u8) -> PendingDepositEventRecord { } } +fn dispatch_record(seed: u8) -> PendingCrossZoneDispatchRecord { + PendingCrossZoneDispatchRecord::recorded([seed; 32], vec![seed; 4]) +} + +/// A distinct message key per index, for filling the pending list. +fn key_from_index(index: usize) -> [u8; 32] { + let mut key = [0_u8; 32]; + key[..8].copy_from_slice(&u64::try_from(index).expect("test index fits").to_le_bytes()); + key +} + fn stored_balance(dbio: &RocksDBIO) -> u128 { dbio.get_lee_state() .unwrap() @@ -399,6 +410,171 @@ fn finalized_deposit_records_are_removed_by_op_id() { assert_eq!(stored, vec![second]); } +#[test] +fn dispatch_records_round_trip_and_dedupe_by_message_key() { + let temp_dir = tempdir().unwrap(); + let (dbio, _genesis) = dbio_with_genesis(temp_dir.path()); + + let record = dispatch_record(1); + assert_eq!( + dbio.add_pending_cross_zone_dispatches(vec![record.clone()]) + .unwrap(), + 1 + ); + // The watcher re-reads a slot it stalled on, so the same delivery arrives + // again; recording it twice would double-count its failed attempts. + assert_eq!( + dbio.add_pending_cross_zone_dispatches(vec![record.clone(), dispatch_record(2)]) + .unwrap(), + 1, + "only the delivery not already held is newly recorded" + ); + + assert_eq!( + dbio.get_pending_cross_zone_dispatches().unwrap(), + vec![record, dispatch_record(2)] + ); +} + +#[test] +fn recording_past_the_cap_writes_nothing() { + let temp_dir = tempdir().unwrap(); + let (dbio, _genesis) = dbio_with_genesis(temp_dir.path()); + + // What fills this list is chosen by peer zones, so the bound is what stops a + // peer deciding how large our store gets. Refusing the whole write leaves + // the watcher's floor where it is, so the slot is read again later and + // nothing is lost. + let full: Vec<_> = (0..MAX_PENDING_CROSS_ZONE_DISPATCHES) + .map(|seed| PendingCrossZoneDispatchRecord::recorded(key_from_index(seed), vec![0_u8; 4])) + .collect(); + assert_eq!( + dbio.add_pending_cross_zone_dispatches(full).unwrap(), + MAX_PENDING_CROSS_ZONE_DISPATCHES + ); + + let over = PendingCrossZoneDispatchRecord::recorded( + key_from_index(MAX_PENDING_CROSS_ZONE_DISPATCHES), + vec![0_u8; 4], + ); + assert!( + dbio.add_pending_cross_zone_dispatches(vec![over]).is_err(), + "recording past the cap must fail so the caller holds its floor" + ); + assert_eq!( + dbio.get_pending_cross_zone_dispatches().unwrap().len(), + MAX_PENDING_CROSS_ZONE_DISPATCHES, + "a refused write must leave the list untouched" + ); + + // Re-offering only what is already held is not growth, so it still succeeds. + assert_eq!( + dbio.add_pending_cross_zone_dispatches(vec![PendingCrossZoneDispatchRecord::recorded( + key_from_index(0), + vec![0_u8; 4] + )]) + .unwrap(), + 0 + ); +} + +#[test] +fn settled_dispatch_records_are_dropped_outside_an_update() { + let temp_dir = tempdir().unwrap(); + let (dbio, _genesis) = dbio_with_genesis(temp_dir.path()); + + // The watcher re-reads a slot it already consumed and re-records a delivery + // that settled long ago. Its key will never appear in a future block, so the + // store-update path cannot reach it and this is the only thing that does. + let first = dispatch_record(1); + let second = dispatch_record(2); + dbio.add_pending_cross_zone_dispatches(vec![first.clone(), second.clone()]) + .unwrap(); + + assert_eq!( + dbio.drop_settled_cross_zone_dispatches(&[first.message_key]) + .unwrap(), + 1 + ); + assert_eq!( + dbio.get_pending_cross_zone_dispatches().unwrap(), + vec![second] + ); + + // Dropping one that is already gone is a no-op, not an error. + assert_eq!( + dbio.drop_settled_cross_zone_dispatches(&[first.message_key]) + .unwrap(), + 0 + ); +} + +#[test] +fn finalized_dispatch_records_are_removed_by_message_key() { + let temp_dir = tempdir().unwrap(); + let (dbio, _genesis) = dbio_with_genesis(temp_dir.path()); + + let first = dispatch_record(1); + let second = dispatch_record(2); + dbio.add_pending_cross_zone_dispatches(vec![first.clone(), second.clone()]) + .unwrap(); + + // Only the finalized delivery's key is dropped. Two deliveries can sit in + // the same block, so a record must go by its own identity rather than by + // anything about the height its delivery landed at. + dbio.store_update(&StoreUpdate { + remove_dispatch_records: &[first.message_key], + ..StoreUpdate::new(&state_with_balance(100)) + }) + .unwrap(); + + assert_eq!( + dbio.get_pending_cross_zone_dispatches().unwrap(), + vec![second] + ); +} + +#[test] +fn record_dispatch_failure_drops_the_record_at_the_limit() { + let temp_dir = tempdir().unwrap(); + let (dbio, _genesis) = dbio_with_genesis(temp_dir.path()); + + let record = dispatch_record(1); + let key = record.message_key; + let survivor = dispatch_record(2); + dbio.add_pending_cross_zone_dispatches(vec![record, survivor.clone()]) + .unwrap(); + + assert!(!dbio.record_dispatch_failure(key, 3).unwrap()); + assert_eq!( + dbio.get_pending_cross_zone_dispatches().unwrap()[0].failed_attempts, + 1, + "a failure short of the limit is counted, not given up on" + ); + assert!(!dbio.record_dispatch_failure(key, 3).unwrap()); + assert!( + dbio.record_dispatch_failure(key, 3).unwrap(), + "the third failure is the one it is given up on" + ); + + // Dropped rather than flagged: a delivery the drain will never feed into a + // block again is one nothing would ever remove, so flagging it would let a + // peer that can make deliveries fail grow the list without bound. + assert_eq!( + dbio.get_pending_cross_zone_dispatches().unwrap(), + vec![survivor], + "giving up on a delivery drops its record and leaves the others alone" + ); + + // A key with no record reads as given up on: there is nothing left to count + // against, and nothing will feed it into a block. + assert!( + dbio.record_dispatch_failure(key, 3).unwrap(), + "a failure against a dropped delivery must not re-create its record" + ); + assert_eq!(dbio.get_pending_cross_zone_dispatches().unwrap().len(), 1); +} + #[test] fn repeated_withdrawal_key_in_one_update_folds_once_per_occurrence() { let temp_dir = tempdir().unwrap(); diff --git a/test_fixtures/src/lib.rs b/test_fixtures/src/lib.rs index 596354ab..9077feaf 100644 --- a/test_fixtures/src/lib.rs +++ b/test_fixtures/src/lib.rs @@ -217,7 +217,7 @@ impl Drop for TestContext { temp_wallet_dir: _, } = self; - let sequencer_handle = sequencer_handle + let mut sequencer_handle = sequencer_handle .take() .expect("Sequencer handle should be present in TestContext drop"); if !sequencer_handle.is_healthy() {