From 5b87891230bbd656c5b7ca41ab63146ec2dae764 Mon Sep 17 00:00:00 2001 From: moudyellaz Date: Thu, 13 Aug 2026 21:46:19 +0200 Subject: [PATCH] feat(cross-zone): extract one peer-block acceptance policy and adopt it in the watcher --- Cargo.lock | 2 + lez/common/src/block.rs | 12 + lez/cross_zone/Cargo.toml | 6 + lez/cross_zone/src/acceptance.rs | 472 +++++++++++++++++++ lez/cross_zone/src/lib.rs | 8 + lez/cross_zone/src/test_utils.rs | 51 ++ lez/sequencer/core/Cargo.toml | 1 + lez/sequencer/core/src/cross_zone_watcher.rs | 396 +++------------- lez/storage/src/sequencer/sequencer_cells.rs | 18 +- 9 files changed, 615 insertions(+), 351 deletions(-) create mode 100644 lez/cross_zone/src/acceptance.rs create mode 100644 lez/cross_zone/src/test_utils.rs diff --git a/Cargo.lock b/Cargo.lock index 06d7be1ad..7ffd2693e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1979,7 +1979,9 @@ name = "cross_zone" version = "0.1.0" dependencies = [ "bridge_lock_core", + "common", "cross_zone_inbox_core", + "hex", "lee", "lee_core", "ping_core", diff --git a/lez/common/src/block.rs b/lez/common/src/block.rs index 53a2f0337..d5d244b66 100644 --- a/lez/common/src/block.rs +++ b/lez/common/src/block.rs @@ -22,6 +22,18 @@ impl From<&Block> for BlockMeta { } } +/// The last peer block accepted onto a cross-zone peer chain, and the link the +/// next one has to carry. +/// +/// `block_hash` is the recomputed hash, not `header.hash` as read: the +/// signature does not cover that field, so a signed block may carry a bogus one +/// and break the link against the peer's next honest block. +#[derive(Debug, Clone, Copy, PartialEq, Eq, BorshSerialize, BorshDeserialize)] +pub struct PeerChainTip { + pub block_id: u64, + pub block_hash: HashType, +} + #[derive(Debug, Clone)] /// Our own hasher. /// Currently it is SHA256 hasher wrapper. May change in a future. diff --git a/lez/cross_zone/Cargo.toml b/lez/cross_zone/Cargo.toml index 25341bda7..2a0d9e1ab 100644 --- a/lez/cross_zone/Cargo.toml +++ b/lez/cross_zone/Cargo.toml @@ -7,13 +7,19 @@ license = { workspace = true } [lints] workspace = true +[features] +# Chain-builder helpers shared by the watcher's and verifier's tests. +test-utils = [] + [dependencies] lee.workspace = true lee_core.workspace = true programs.workspace = true +common.workspace = true cross_zone_inbox_core.workspace = true bridge_lock_core.workspace = true ping_core.workspace = true wrapped_token_core.workspace = true serde.workspace = true risc0-zkvm.workspace = true +hex.workspace = true diff --git a/lez/cross_zone/src/acceptance.rs b/lez/cross_zone/src/acceptance.rs new file mode 100644 index 000000000..eaa177aae --- /dev/null +++ b/lez/cross_zone/src/acceptance.rs @@ -0,0 +1,472 @@ +//! The one peer-block acceptance policy. +//! +//! The sequencer's watcher and the indexer's verifier both admit peer blocks +//! through here, so they cannot disagree about which block holds an id: a +//! disagreement makes the verifier re-derive a delivery against a block the +//! watcher never delivered from, and ingestion halts. + +use std::fmt::{self, Display, Formatter}; + +use common::{ + HashType, + block::{Block, PeerChainTip}, +}; +use cross_zone_inbox_core::ZoneId; +use lee::{GENESIS_BLOCK_ID, PublicKey}; + +/// Consecutive passes a reader spends stuck on one slot before it says so as +/// something more than the per-pass failure. It never reads past the slot. +/// +/// The cadence bounds log volume only: at one pass per poll interval, 5 passes +/// is roughly seconds to tens of seconds depending on each side's interval. +pub const STUCK_SLOT_ALERT_PASSES: u32 = 5; + +/// Where a screened peer block sits relative to the chain pinned by `tip`. +#[derive(Debug, PartialEq, Eq)] +pub enum Link { + /// The next block on the peer's chain, carrying its recomputed hash. + Next(HashType), + /// At or below the tip, so already accepted from. The ordinary shape of a + /// re-read slot. `equivocates` is set when the block claims the tip's own + /// id under a different hash, so callers can say so; below the tip there is + /// no held hash to compare against and it is never set. + AlreadySeen { equivocates: bool }, + /// Not on the chain the tip pins, so not acceptable. Callers read on: the + /// peer's own next block still links to the tip, and treating this as + /// terminal would hand the peer a way to stop its deliveries permanently. + OffChain(OffChain), +} + +/// Why a block is not on the chain the tip pins. +#[derive(Debug, PartialEq, Eq)] +pub enum OffChain { + /// The next id off the tip, but linking to some other predecessor. + DoesNotLink { + block_id: u64, + tip_id: u64, + links_to: HashType, + expected: HashType, + }, + /// Read with no stored tip, and not the peer's genesis. Anchoring on + /// whatever arrived first would let the peer pick the id, and burn every + /// replay key below it with one block. + NotTheGenesis { block_id: u64 }, + /// An id above the next one on the chain. + SkipsAhead { block_id: u64, next: u64 }, +} + +impl Display for OffChain { + fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result { + match *self { + Self::DoesNotLink { + block_id, + tip_id, + links_to, + expected, + } => write!( + f, + "block {block_id} does not follow block {tip_id}: it links to {links_to} rather than {expected}" + ), + Self::NotTheGenesis { block_id } => write!( + f, + "block {block_id} is the first one read, but with no stored chain tip acceptance has to start at the peer's genesis block {GENESIS_BLOCK_ID}" + ), + Self::SkipsAhead { block_id, next } => write!( + f, + "block {block_id} skips past {next}, which is either a hole in what this node read or an id claimed ahead of the peer's chain" + ), + } + } +} + +/// Why a peer block was refused before any chain placement. +/// +/// The channel authorizes who may write, not what they may claim. The hash +/// check is unconditional: the signature does not cover `header.hash`, and the +/// chain link compares hashes, so an unchecked one lets a peer assert links it +/// never built. The pinned key, checked only when one is configured, is what +/// says the peer's own sequencer produced the block; it subsumes nothing here, +/// since a correctly signed block may still carry a bogus hash. +#[derive(Debug, PartialEq, Eq)] +pub enum ScreenRefusal { + /// `header.hash` is not the hash of the block's contents. + HashMismatch { + block_id: u64, + declared: HashType, + recomputed: HashType, + }, + /// The block is not signed by the pinned block-signing key. + KeyMismatch { block_id: u64 }, +} + +impl Display for ScreenRefusal { + fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result { + match *self { + Self::HashMismatch { + block_id, + declared, + recomputed, + } => write!( + f, + "block {block_id} carries header hash {declared} but its contents hash to {recomputed}" + ), + Self::KeyMismatch { block_id } => write!( + f, + "block {block_id} is not signed by the pinned block-signing key" + ), + } + } +} + +/// The pass-to-pass stall of one peer reader: the slot it is stuck on and how +/// many consecutive passes it has spent there. Keyed by slot so a failure at a +/// new slot does not inherit an older slot's count. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct StallState { + stalled: Option<(S, u32)>, +} + +impl Default for StallState { + fn default() -> Self { + Self { stalled: None } + } +} + +impl StallState { + /// Folds one pass in: `stuck_on` is the slot the pass ended inside, `None` + /// for a pass that ended cleanly. Returns the slot the reader is stuck on + /// and how long it has been stuck, so the caller can say so on the + /// [`alerts_at`] cadence. + /// + /// `read_to` is the read position after the pass, and 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 it a flaky peer endpoint resets the count for ever and a + /// reader stuck for hours never says so. + pub fn after_pass(&mut self, stuck_on: Option, read_to: Option) -> Option<(S, u32)> { + let Some(slot) = stuck_on else { + if self.passed_the_stall(read_to) { + self.stalled = None; + } + return None; + }; + let attempts = match self.stalled { + Some((held, attempts)) if held == slot => attempts.saturating_add(1), + _ => 1, + }; + self.stalled = Some((slot, attempts)); + self.stalled + } + + /// Whether the read position is now past whatever the reader was stuck on. + /// Vacuously true when it was not stuck. + fn passed_the_stall(self, read_to: Option) -> bool { + self.stalled + .is_none_or(|(stuck_on, _)| read_to.is_some_and(|slot| slot >= stuck_on)) + } +} + +/// Whether a reader stuck for `attempts` passes should say so on this one. +/// +/// Every [`STUCK_SLOT_ALERT_PASSES`], not on the crossing alone: a stall that +/// never clears would otherwise be reported once and then look resolved for as +/// long as it lasts. Not every pass, since that is one line per block time. +#[must_use] +pub const fn alerts_at(attempts: u32) -> bool { + attempts > 0 && attempts.is_multiple_of(STUCK_SLOT_ALERT_PASSES) +} + +/// The one report both sides log for a differing block at an id the accepted +/// chain already holds. +#[must_use] +pub fn equivocation_report( + peer_zone: &ZoneId, + block_id: u64, + holding: HashType, + refusing: HashType, +) -> String { + format!( + "Peer zone {} equivocated at block {block_id}: holding {holding}, refusing {refusing}. Nothing at or above block {block_id} can be delivered from until that peer inscribes a block continuing the run verified from its genesis.", + hex::encode(peer_zone) + ) +} + +/// Whether `block` continues the peer chain pinned by `tip`. +/// +/// `recomputed` is the hash [`screen_peer_block`] returned for it, so a tip +/// stored off [`Link::Next`] pins contents rather than a declared field. +/// +/// This is what closes the id suppression. A delivered message's replay key +/// covers `(src_zone, src_block_id, src_tx_index)` and nothing else, so a peer +/// that can get a block accepted under an id of its choosing burns the key an +/// honest block would later use, and the inbox then no-ops the real message as +/// a replay. Off a hash link ids are only claimable in order, so the only id +/// within reach is the one the peer is about to publish anyway. +#[must_use] +pub fn link_to_tip(tip: Option<&PeerChainTip>, block: &Block, recomputed: HashType) -> Link { + let block_id = block.header.block_id; + let Some(tip) = tip else { + return if block_id == GENESIS_BLOCK_ID { + Link::Next(recomputed) + } else { + Link::OffChain(OffChain::NotTheGenesis { block_id }) + }; + }; + + if block_id <= tip.block_id { + return Link::AlreadySeen { + equivocates: block_id == tip.block_id && recomputed != tip.block_hash, + }; + } + let next = tip.block_id.saturating_add(1); + if block_id > next { + return Link::OffChain(OffChain::SkipsAhead { block_id, next }); + } + if block.header.prev_block_hash != tip.block_hash { + return Link::OffChain(OffChain::DoesNotLink { + block_id, + tip_id: tip.block_id, + links_to: block.header.prev_block_hash, + expected: tip.block_hash, + }); + } + Link::Next(recomputed) +} + +/// Whether a block read off a peer's channel may be considered for the chain at +/// all, returning the recomputed hash every later placement has to use. +pub fn screen_peer_block( + block: &Block, + expected_pubkey: Option<&PublicKey>, +) -> Result { + let recomputed = block.recompute_hash(); + if recomputed != block.header.hash { + return Err(ScreenRefusal::HashMismatch { + block_id: block.header.block_id, + declared: block.header.hash, + recomputed, + }); + } + if expected_pubkey.is_some_and(|key| !block.is_signed_by(key)) { + return Err(ScreenRefusal::KeyMismatch { + block_id: block.header.block_id, + }); + } + Ok(recomputed) +} + +#[cfg(test)] +mod tests { + use common::test_utils::produce_dummy_block; + + use super::*; + use crate::test_utils::linked_chain_to; + + /// The peer's block at `block_id`, on its one honest chain. + fn chain_block(block_id: u64) -> Block { + linked_chain_to(block_id, |_| vec![]) + .pop() + .expect("chain reaches block_id") + } + + /// The hash the block after `block_id` has to link to. + fn chain_hash(block_id: u64) -> HashType { + chain_block(block_id).header.hash + } + + /// The tip a reader holds after accepting up to `block_id`. + fn tip_at(block_id: u64) -> PeerChainTip { + PeerChainTip { + block_id, + block_hash: chain_hash(block_id), + } + } + + fn screened(block: &Block) -> HashType { + screen_peer_block(block, None).expect("honest block passes screening") + } + + /// Runs the stall machine over `(stuck_on, read_to)` pass results. + fn run_stalls(passes: &[(Option, Option)]) -> StallState { + let mut state = StallState::default(); + for (stuck_on, read_to) in passes { + state.after_pass(*stuck_on, *read_to); + } + state + } + + #[test] + fn only_the_next_block_off_the_tip_links() { + let tip = tip_at(2); + + let next = chain_block(3); + assert_eq!( + link_to_tip(Some(&tip), &next, screened(&next)), + Link::Next(chain_hash(3)), + "the block that continues the chain is the one accepted" + ); + + // The #677 suppression. The peer's chain is public, so the version that + // matters is the block linking correctly and lying only about the id: + // one with no link at all is caught by the same check and proves + // nothing about this one. + for ahead in [ + produce_dummy_block(5, Some(chain_hash(2)), vec![]), + produce_dummy_block(5, None, vec![]), + ] { + assert_eq!( + link_to_tip(Some(&tip), &ahead, screened(&ahead)), + Link::OffChain(OffChain::SkipsAhead { + block_id: 5, + next: 3 + }) + ); + } + + // Right id, wrong ancestry: the peer forked at our tip, or reset it. + let forked = produce_dummy_block(3, Some(HashType([9; 32])), vec![]); + assert!(matches!( + link_to_tip(Some(&tip), &forked, screened(&forked)), + Link::OffChain(OffChain::DoesNotLink { .. }) + )); + } + + #[test] + fn a_reader_with_no_tip_starts_at_the_peers_genesis() { + let genesis = chain_block(GENESIS_BLOCK_ID); + assert_eq!( + link_to_tip(None, &genesis, screened(&genesis)), + Link::Next(chain_hash(GENESIS_BLOCK_ID)) + ); + // Anchoring on whatever arrived first is the whole attack: the peer + // would pick the id, and every key below it with one block. + let mid_chain = chain_block(GENESIS_BLOCK_ID + 1); + assert_eq!( + link_to_tip(None, &mid_chain, screened(&mid_chain)), + Link::OffChain(OffChain::NotTheGenesis { + block_id: GENESIS_BLOCK_ID + 1 + }) + ); + } + + #[test] + fn a_differing_block_at_the_tip_id_reports_equivocation() { + // Two blocks claiming one id collapse to one replay key on chain, so + // accepting both delivers one message twice. The re-read of the block + // the tip holds is ordinary; only a differing hash is worth a report. + let tip = tip_at(2); + + let held = chain_block(2); + assert_eq!( + link_to_tip(Some(&tip), &held, screened(&held)), + Link::AlreadySeen { equivocates: false } + ); + + let differing = produce_dummy_block(2, Some(HashType([9; 32])), vec![]); + assert_eq!( + link_to_tip(Some(&tip), &differing, screened(&differing)), + Link::AlreadySeen { equivocates: true } + ); + + // Below the tip there is no held hash to compare against. + let below = produce_dummy_block(1, Some(HashType([9; 32])), vec![]); + assert_eq!( + link_to_tip(Some(&tip), &below, screened(&below)), + Link::AlreadySeen { equivocates: false } + ); + } + + #[test] + fn a_block_whose_header_hash_is_not_its_contents_is_refused() { + // A correctly signed block can still carry any value in `header.hash`. + let mut tampered = chain_block(3); + tampered.header.hash = HashType([9; 32]); + assert!(matches!( + screen_peer_block(&tampered, None), + Err(ScreenRefusal::HashMismatch { block_id: 3, .. }) + )); + + // And the hash verdict comes first, whatever else is wrong. + let other = lee::PublicKey::try_new([42; 32]).expect("test key"); + assert!(matches!( + screen_peer_block(&tampered, Some(&other)), + Err(ScreenRefusal::HashMismatch { .. }) + )); + } + + #[test] + fn a_block_not_signed_by_the_pinned_key_is_refused() { + let signer = lee::PublicKey::new_from_private_key( + &lee::PrivateKey::try_new([37; 32]).expect("test key"), + ); + let block = chain_block(GENESIS_BLOCK_ID); + assert_eq!( + screen_peer_block(&block, Some(&signer)), + Ok(chain_hash(GENESIS_BLOCK_ID)), + "produce_dummy_block signs with this key, so the pin must accept it" + ); + + let other = lee::PublicKey::try_new([42; 32]).expect("test key"); + assert!(matches!( + screen_peer_block(&block, Some(&other)), + Err(ScreenRefusal::KeyMismatch { + block_id: GENESIS_BLOCK_ID + }) + )); + } + + #[test] + fn a_stuck_slot_is_counted_but_never_read_past() { + // Counting is only how loud to be about a slot a reader is stuck on; + // nothing here ever moves a cursor. + let passes = vec![(Some(4), Some(3)); 3]; + assert_eq!(run_stalls(&passes).stalled, Some((4, 3))); + + let long = vec![ + (Some(4), Some(3)); + usize::try_from(STUCK_SLOT_ALERT_PASSES).expect("alert threshold fits") * 2 + ]; + assert_eq!( + run_stalls(&long).stalled, + Some((4, STUCK_SLOT_ALERT_PASSES.saturating_mul(2))), + "a slot is retried for as long as it stays stuck" + ); + } + + #[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 count for + // ever, and a reader stuck for hours would never say so. + let mut passes = vec![(Some(4), Some(3)); 5]; + passes.push((None, Some(3))); + assert_eq!( + run_stalls(&passes).stalled, + Some((4, 5)), + "the count survives a pass that never reached the stalled slot" + ); + + // Getting past it is what actually clears the stall. + let mut read_past = vec![(Some(4), Some(3)); 5]; + read_past.push((None, Some(7))); + assert_eq!(run_stalls(&read_past).stalled, None); + } + + #[test] + fn a_stall_at_a_new_slot_starts_its_own_count() { + let passes = [(Some(4), Some(3)), (Some(4), Some(3)), (Some(9), Some(8))]; + assert_eq!(run_stalls(&passes).stalled, Some((9, 1))); + } + + #[test] + fn a_stall_says_so_on_a_cadence_rather_than_once() { + // Reporting only on the crossing leaves a reader that never recovers + // looking resolved. + assert!(!alerts_at(0)); + assert!(!alerts_at(1)); + assert!(!alerts_at(STUCK_SLOT_ALERT_PASSES - 1)); + assert!(alerts_at(STUCK_SLOT_ALERT_PASSES)); + assert!(!alerts_at(STUCK_SLOT_ALERT_PASSES + 1)); + assert!(alerts_at(STUCK_SLOT_ALERT_PASSES * 3)); + } +} diff --git a/lez/cross_zone/src/lib.rs b/lez/cross_zone/src/lib.rs index 58d9344dc..7593b78e8 100644 --- a/lez/cross_zone/src/lib.rs +++ b/lez/cross_zone/src/lib.rs @@ -9,6 +9,10 @@ //! own block-reading, emission-extraction, delivery-building, and trust model; a //! shared trait is best lifted from that first real adapter, not from this one. +pub use acceptance::{ + Link, OffChain, STUCK_SLOT_ALERT_PASSES, ScreenRefusal, StallState, alerts_at, + equivocation_report, link_to_tip, screen_peer_block, +}; pub use cross_zone_inbox_core::{CrossZoneConfig, CrossZonePeer}; use cross_zone_inbox_core::{ CrossZoneMessage, InboxConfig, Instruction, ZoneId, inbox_config_account_id, @@ -20,6 +24,10 @@ use lee_core::{ }; use serde::Serialize; +pub mod acceptance; +#[cfg(any(test, feature = "test-utils"))] +pub mod test_utils; + /// The cross-zone emission fields a watcher or verifier reads off a source /// transaction, common to every emitter program. pub struct Emission { diff --git a/lez/cross_zone/src/test_utils.rs b/lez/cross_zone/src/test_utils.rs new file mode 100644 index 000000000..7789be4eb --- /dev/null +++ b/lez/cross_zone/src/test_utils.rs @@ -0,0 +1,51 @@ +//! Chain-builder helpers shared by the watcher's and verifier's tests, so both +//! sides exercise the acceptance policy against identically built peer chains. + +use common::{block::Block, test_utils::produce_dummy_block, transaction::LeeTransaction}; +use cross_zone_inbox_core::ZoneId; +use lee::{ + GENESIS_BLOCK_ID, PublicTransaction, + public_transaction::{Message, WitnessSet}, +}; +use lee_core::program::ProgramId; +use ping_core::{SenderInstruction, ping_record_pda, receiver_config_account_id}; + +/// The peer's hash-linked chain from its genesis up to and including `last`, +/// each block carrying the transactions `txs_at(block_id)` returns. Empty when +/// `last` is below [`GENESIS_BLOCK_ID`]. +pub fn linked_chain_to(last: u64, txs_at: impl Fn(u64) -> Vec) -> Vec { + let mut blocks: Vec = Vec::new(); + for block_id in GENESIS_BLOCK_ID..=last { + let prev = blocks.last().map(|block| block.header.hash); + blocks.push(produce_dummy_block(block_id, prev, txs_at(block_id))); + } + blocks +} + +/// A `ping_sender` emission aimed at `target_zone` and `target_program_id`, +/// carrying `payload`. The sender lets its caller name any target, which is why +/// routes pin the pair rather than the target alone. +#[must_use] +pub fn ping_emission( + target_zone: ZoneId, + target_program_id: ProgramId, + payload: &[u8], +) -> LeeTransaction { + let receiver_id = programs::ping_receiver().id(); + let send = SenderInstruction::Send { + target_zone, + target_program_id, + target_accounts: vec![ + receiver_config_account_id(receiver_id).into_value(), + ping_record_pda(receiver_id).into_value(), + ], + payload: payload.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![]), + )) +} diff --git a/lez/sequencer/core/Cargo.toml b/lez/sequencer/core/Cargo.toml index a2d8a21c3..d31c0314c 100644 --- a/lez/sequencer/core/Cargo.toml +++ b/lez/sequencer/core/Cargo.toml @@ -54,6 +54,7 @@ testnet = [] mock = [] [dev-dependencies] +cross_zone = { workspace = true, features = ["test-utils"] } futures.workspace = true test_programs.workspace = true lee = { workspace = true, features = ["test-utils"] } diff --git a/lez/sequencer/core/src/cross_zone_watcher.rs b/lez/sequencer/core/src/cross_zone_watcher.rs index 9a910df10..4d6f30147 100644 --- a/lez/sequencer/core/src/cross_zone_watcher.rs +++ b/lez/sequencer/core/src/cross_zone_watcher.rs @@ -2,11 +2,12 @@ use std::{sync::Arc, time::Duration}; use common::{HashType, block::Block, transaction::LeeTransaction}; use cross_zone::{ - EmissionSource, build_dispatch_from_emission, extract_emission, is_sequencer_only_program, + EmissionSource, Link, StallState, alerts_at, build_dispatch_from_emission, equivocation_report, + extract_emission, is_sequencer_only_program, link_to_tip, screen_peer_block, }; use cross_zone_inbox_core::message_key; use futures::{Stream, StreamExt as _}; -use lee::{GENESIS_BLOCK_ID, PublicKey}; +use lee::PublicKey; use log::{debug, error, warn}; use logos_blockchain_core::mantle::ops::channel::ChannelId; use logos_blockchain_zone_sdk::{ @@ -25,15 +26,6 @@ use crate::{ task_group::TaskGroup, }; -/// Consecutive passes a watcher spends stuck on one slot before it says so as -/// something more than the per-pass failure. -/// -/// 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; anything still stuck after it wants -/// someone to look. -const STUCK_SLOT_ALERT_PASSES: u32 = 20; - /// The per-peer settings one watcher pass needs. struct PeerContext { peer_zone: [u8; 32], @@ -46,7 +38,7 @@ struct PeerContext { /// All of them hold the delivery floor at the last slot the watcher consumed /// whole, bar [`PassOutcome::Drained`] and [`PassOutcome::Stranded`], so the /// next pass re-reads from there. Only a block that will not deserialize ends a -/// pass; [`link_against`] says why one that decodes never does. +/// pass; [`link_to_tip`] says why one that decodes never does. #[derive(Clone, Copy, Debug, PartialEq, Eq)] enum PassOutcome { /// The stream drained, having delivered from at least one block or found @@ -71,10 +63,7 @@ enum PassOutcome { /// The pass-to-pass state of one watcher. #[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] struct WatcherState { - /// The slot it is stuck on and how many consecutive 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)>, + stall: StallState, /// Consecutive passes that placed nothing while skipping blocks. Not keyed /// by slot: the peer keeps producing, so every such pass ends at a new slot /// and a slot-keyed count would reset to one for ever. @@ -82,44 +71,19 @@ struct WatcherState { } impl WatcherState { - /// Folds one pass's outcome in, returning the slot the watcher is stuck on - /// and how long it has been stuck, so the caller can say so. - /// - /// `cursor` is the read position after the pass, and 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 it a flaky peer endpoint resets the count for ever and a watcher - /// stuck for hours never says so. + /// Folds one pass's outcome into the stall and stranded counts, returning + /// the slot the watcher is stuck on and how long it has been stuck. fn after_pass(&mut self, outcome: PassOutcome, cursor: Option) -> Option<(Slot, u32)> { - let slot = match outcome { - PassOutcome::Drained | PassOutcome::Stranded => { - if self.passed_the_stall(cursor) { - self.stalled = None; - } - self.stranded = match outcome { - PassOutcome::Stranded => self.stranded.saturating_add(1), - PassOutcome::Drained - | PassOutcome::Undecodable(_) - | PassOutcome::Undelivered(_) => 0, - }; - return None; - } - PassOutcome::Undecodable(slot) | PassOutcome::Undelivered(slot) => slot, + match outcome { + PassOutcome::Stranded => self.stranded = self.stranded.saturating_add(1), + PassOutcome::Drained => self.stranded = 0, + PassOutcome::Undecodable(_) | PassOutcome::Undelivered(_) => {} + } + let stuck_on = match outcome { + PassOutcome::Undecodable(slot) | PassOutcome::Undelivered(slot) => Some(slot), + PassOutcome::Drained | PassOutcome::Stranded => None, }; - - let attempts = match self.stalled { - Some((stuck_on, attempts)) if stuck_on == slot => attempts.saturating_add(1), - _ => 1, - }; - self.stalled = Some((slot, attempts)); - self.stalled - } - - /// 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)) + self.stall.after_pass(stuck_on, cursor) } } @@ -132,96 +96,6 @@ struct Resume { clear_floor: bool, } -/// Where a peer block sits relative to the chain this watcher has delivered -/// from. -#[derive(Debug, PartialEq, Eq)] -enum Link { - /// The next block on the peer's chain, carrying its recomputed hash. - Next(HashType), - /// At or below the tip, so already delivered from. The ordinary shape of a - /// re-read slot, and how an equivocating second block at one id is refused. - AlreadySeen, - /// Not on the chain this watcher is following, so not deliverable. Read on: - /// the peer's own next block still links to the tip, and treating this as - /// terminal would hand the peer a way to stop its deliveries permanently. - OffChain(String), -} - -/// Whether `block` continues the peer chain pinned by `tip`. -/// -/// This is what closes the suppression. A delivered message's replay key covers -/// `(src_zone, src_block_id, src_tx_index)` and nothing else, so a peer that can -/// get a block delivered under an id of its choosing burns the key an honest -/// block would later use, and the inbox then no-ops the real message as a -/// replay. Off a hash link ids are only claimable in order, so the only id -/// within reach is the one the peer is about to publish anyway. -/// -/// Nothing but [`Link::Next`] is ever delivered from, and nothing but a block -/// that will not decode stops the pass. A peer can inscribe anything it likes on -/// its own channel, so a block this watcher cannot place is read past rather -/// than treated as the end of the chain: the peer's own next honest block still -/// links to the tip. -fn link_against( - tip: Option, - block: &Block, - expected_pubkey: Option<&PublicKey>, -) -> Link { - // The channel authorizes who may write, not what they may claim, so the - // pinned key is what says this node's own sequencer produced the block. - if expected_pubkey.is_some_and(|key| !block.is_signed_by(key)) { - return Link::OffChain("block-signing key does not match the pinned key".to_owned()); - } - - let recomputed = block.recompute_hash(); - if recomputed != block.header.hash { - // The signature does not cover this field, so a correctly signed block - // may still carry a bogus one, and the peer's own next block links - // against the recomputed value rather than this one. - return Link::OffChain(format!( - "block {} carries header hash {} but its contents hash to {recomputed}", - block.header.block_id, block.header.hash - )); - } - - let Some(tip) = tip else { - return if block.header.block_id == GENESIS_BLOCK_ID { - Link::Next(recomputed) - } else { - Link::OffChain(format!( - "block {} is the first one read, but a watcher with no stored chain tip has to start at the peer's genesis block {GENESIS_BLOCK_ID}", - block.header.block_id - )) - }; - }; - - if block.header.block_id <= tip.block_id { - return Link::AlreadySeen; - } - if block.header.block_id > tip.block_id.saturating_add(1) { - return Link::OffChain(format!( - "block {} skips past {}, which is either a hole in what this node read or an id claimed ahead of the peer's chain", - block.header.block_id, - tip.block_id.saturating_add(1) - )); - } - if block.header.prev_block_hash != tip.block_hash { - return Link::OffChain(format!( - "block {} does not follow block {} we delivered from: it links to {} rather than {}", - block.header.block_id, tip.block_id, block.header.prev_block_hash, tip.block_hash - )); - } - Link::Next(recomputed) -} - -/// Whether a watcher stuck for `attempts` passes should say so on this one. -/// -/// Every [`STUCK_SLOT_ALERT_PASSES`], not on the crossing alone: a stall that -/// never clears would otherwise be reported once and then look resolved for as -/// long as it lasts. Not every pass, since that is one line per block time. -const fn alerts_at(attempts: u32) -> bool { - attempts > 0 && attempts.is_multiple_of(STUCK_SLOT_ALERT_PASSES) -} - /// Where a starting watcher resumes reading a peer's channel. /// /// A store holding a floor but no tip predates chain pinning, and its next block @@ -452,13 +326,42 @@ where hex::encode(peer.peer_zone), block.header.block_id ); - match link_against(*tip, &block, peer.expected_pubkey.as_ref()) { - Link::AlreadySeen => { - debug!( - "Watcher ignoring peer {} block {}: at or below the block it has already delivered from", - hex::encode(peer.peer_zone), - block.header.block_id + // Nothing but [`Link::Next`] is ever delivered from, and + // nothing but a block that will not decode stops the pass. A + // peer can inscribe anything it likes on its own channel, so a + // block this watcher cannot place is read past rather than + // treated as the end of the chain: the peer's own next honest + // block still links to the tip. + let link = match screen_peer_block(&block, peer.expected_pubkey.as_ref()) { + Ok(recomputed) => link_to_tip(tip.as_ref(), &block, recomputed), + Err(refusal) => { + skipped = skipped.saturating_add(1); + warn!( + "Watcher not delivering from peer {} block at slot {slot:?}: {refusal}. Reading on; the peer's next block that continues the chain still delivers.", + hex::encode(peer.peer_zone) ); + continue; + } + }; + match link { + Link::AlreadySeen { equivocates } => { + if equivocates && let Some(held) = *tip { + error!( + "{}", + equivocation_report( + &peer.peer_zone, + block.header.block_id, + held.block_hash, + block.header.hash + ) + ); + } else { + debug!( + "Watcher ignoring peer {} block {}: at or below the block it has already delivered from", + hex::encode(peer.peer_zone), + block.header.block_id + ); + } } Link::OffChain(reason) => { skipped = skipped.saturating_add(1); @@ -546,8 +449,8 @@ fn advance_cursor(dbio: &RocksDBIO, peer_zone: [u8; 32], cursor: &mut Option LeeTransaction { - let receiver_id = programs::ping_receiver().id(); - let send = SenderInstruction::Send { - target_zone: SELF_ZONE, - target_program_id, - target_accounts: vec![ - receiver_config_account_id(receiver_id).into_value(), - 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![]), - )) + ping_emission(SELF_ZONE, target_program_id, b"hi") } fn peer_msg(data: Vec, slot: u64) -> (ZoneMessage, Slot) { @@ -730,14 +611,9 @@ mod tests { /// The peer's chain from its genesis up to and including `block_id`, each /// block linked to the one before it and carrying one emission for this - /// zone. Empty below [`GENESIS_BLOCK_ID`]. + /// zone. fn chain_to(block_id: u64) -> Vec { - let mut blocks: Vec = Vec::new(); - for id in GENESIS_BLOCK_ID..=block_id { - let prev = blocks.last().map(|block| block.header.hash); - blocks.push(produce_dummy_block(id, prev, vec![emission()])); - } - blocks + linked_chain_to(block_id, |_| vec![emission()]) } /// The peer's block at `block_id`. @@ -826,60 +702,6 @@ mod tests { state } - fn stall(slot: u64, cursor: Option) -> (PassOutcome, Option) { - (PassOutcome::Undecodable(Slot::from(slot)), cursor) - } - - #[test] - fn a_stuck_slot_is_counted_but_never_read_past() { - // The watcher used to give up on a slot and read past it. Counting is - // now only how loud to be about one it is stuck on. - let passes = vec![stall(4, Some(3)); 3]; - assert_eq!(run_passes(&passes).stalled, Some((Slot::from(4), 3))); - - let long = vec![ - stall(4, Some(3)); - usize::try_from(STUCK_SLOT_ALERT_PASSES).expect("alert threshold fits") * 2 - ]; - assert_eq!( - run_passes(&long).stalled, - Some((Slot::from(4), STUCK_SLOT_ALERT_PASSES.saturating_mul(2))), - "a slot is retried for as long as it stays stuck" - ); - } - - #[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 count for - // ever, and a watcher stuck for hours would never say so. - 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" - ); - - // Getting 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_stall_says_so_on_a_cadence_rather_than_once() { - // Reporting only on the crossing leaves a watcher that never recovers - // looking resolved, which is the failure this whole commit is about. - assert!(!alerts_at(0)); - assert!(!alerts_at(1)); - assert!(!alerts_at(STUCK_SLOT_ALERT_PASSES - 1)); - assert!(alerts_at(STUCK_SLOT_ALERT_PASSES)); - assert!(!alerts_at(STUCK_SLOT_ALERT_PASSES + 1)); - assert!(alerts_at(STUCK_SLOT_ALERT_PASSES * 3)); - } - #[test] fn passes_that_place_nothing_while_skipping_blocks_are_counted() { // A tip that stops tracking the peer is silent by construction: every @@ -903,12 +725,6 @@ mod tests { assert_eq!(run_passes(&[(PassOutcome::Drained, Some(4))]).stranded, 0); } - #[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 every_way_of_ending_early_keeps_the_slot_coming_back() { // Undecodable and undelivered differ in whose problem they are, not in @@ -918,106 +734,14 @@ mod tests { PassOutcome::Undecodable(Slot::from(4)), PassOutcome::Undelivered(Slot::from(4)), ] { + let mut state = WatcherState::default(); assert_eq!( - run_passes(&[(outcome, Some(3))]).stalled, + state.after_pass(outcome, Some(Slot::from(3))), Some((Slot::from(4), 1)) ); } } - #[test] - fn only_the_next_block_off_the_tip_links() { - let tip = Some(tip_at(2)); - - assert_eq!( - link_against(tip, &chain_block(3), None), - Link::Next(chain_hash(3)), - "the block that continues the chain is the one delivered from" - ); - - // The #677 suppression. The peer's chain is public, so the version that - // matters is the block linking correctly and lying only about the id: - // one with no link at all is caught by the check below and proves - // nothing about this one. - assert!(matches!( - link_against( - tip, - &produce_dummy_block(5, Some(chain_hash(2)), vec![emission()]), - None - ), - Link::OffChain(_) - )); - assert!(matches!( - link_against(tip, &produce_dummy_block(5, None, vec![emission()]), None), - Link::OffChain(_) - )); - - // Two blocks claiming one id collapse to one key on chain, so - // delivering from both delivers one message twice. - assert_eq!(link_against(tip, &chain_block(2), None), Link::AlreadySeen); - assert_eq!( - link_against( - tip, - &produce_dummy_block(2, Some(HashType([9; 32])), vec![emission()]), - None - ), - Link::AlreadySeen - ); - - // Right id, wrong ancestry: the peer forked at our tip, or reset it. - assert!(matches!( - link_against( - tip, - &produce_dummy_block(3, Some(HashType([9; 32])), vec![emission()]), - None - ), - Link::OffChain(_) - )); - } - - #[test] - fn a_watcher_with_no_tip_starts_at_the_peers_genesis() { - assert_eq!( - link_against(None, &chain_block(GENESIS_BLOCK_ID), None), - Link::Next(chain_hash(GENESIS_BLOCK_ID)) - ); - // Anchoring on whatever arrived first is the whole attack: the peer - // would pick the id, and every key below it with one block. - assert!(matches!( - link_against(None, &chain_block(GENESIS_BLOCK_ID + 1), None), - Link::OffChain(_) - )); - } - - #[test] - fn a_block_whose_header_hash_is_not_its_contents_is_off_chain() { - // A correctly signed block can still carry any value in `header.hash`. - let mut tampered = chain_block(3); - tampered.header.hash = HashType([9; 32]); - assert!(matches!( - link_against(Some(tip_at(2)), &tampered, None), - Link::OffChain(_) - )); - } - - #[test] - fn a_block_not_signed_by_the_pinned_key_is_not_delivered_from() { - let signer = lee::PublicKey::new_from_private_key( - &lee::PrivateKey::try_new([37; 32]).expect("test key"), - ); - assert_eq!( - link_against(None, &chain_block(GENESIS_BLOCK_ID), Some(&signer)), - Link::Next(chain_hash(GENESIS_BLOCK_ID)), - "produce_dummy_block signs with this key, so the pin must accept it" - ); - - let other = lee::PublicKey::try_new([42; 32]).expect("test key"); - assert!(matches!( - link_against(None, &chain_block(GENESIS_BLOCK_ID), Some(&other)), - Link::OffChain(_) - )); - } - #[test] fn a_floor_without_a_tip_resumes_from_the_peers_genesis() { // A store written before chain pinning. The floor is cleared rather diff --git a/lez/storage/src/sequencer/sequencer_cells.rs b/lez/storage/src/sequencer/sequencer_cells.rs index 8cbfe012d..91a0ae6a9 100644 --- a/lez/storage/src/sequencer/sequencer_cells.rs +++ b/lez/storage/src/sequencer/sequencer_cells.rs @@ -1,4 +1,5 @@ use borsh::{BorshDeserialize, BorshSerialize}; +pub use common::block::PeerChainTip; use common::{HashType, block::BlockMeta}; use lee::V03State; @@ -542,21 +543,8 @@ impl SimpleWritableCell for PeerFloorCellRef<'_> { } } -/// The last peer block a cross-zone watcher delivered from, and the link the -/// next one has to carry. -/// -/// `block_hash` is the recomputed hash, not `header.hash` as read: the -/// signature does not cover that field, so a signed block may carry a bogus one -/// and break the link against the peer's next honest block. -/// -/// Durable, not in-memory: a watcher that re-anchored on restart would accept a -/// block claiming any id. -#[derive(Debug, Clone, Copy, PartialEq, Eq, BorshSerialize, BorshDeserialize)] -pub struct PeerChainTip { - pub block_id: u64, - pub block_hash: HashType, -} - +/// The watcher's [`PeerChainTip`], durable rather than in-memory: a watcher +/// that re-anchored on restart would accept a block claiming any id. #[derive(Debug, BorshSerialize, BorshDeserialize)] pub struct PeerTipCell(pub PeerChainTip);