feat(cross-zone): extract one peer-block acceptance policy and adopt it in the watcher

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",

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.

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

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<S> {
+    stalled: Option<(S, u32)>,
+}
+
+impl<S> Default for StallState<S> {
+    fn default() -> Self {
+        Self { stalled: None }
+    }
+}
+
+impl<S: Copy + PartialEq + PartialOrd> StallState<S> {
+    /// 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<S>, read_to: Option<S>) -> 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<S>) -> 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<HashType, ScreenRefusal> {
+    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<u64>, Option<u64>)]) -> StallState<u64> {
+        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));
+    }
+}

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 {

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<LeeTransaction>) -> Vec<Block> {
+    let mut blocks: Vec<Block> = 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![]),
+    ))
+}

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"] }

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<Slot>,
     /// 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<Slot>) -> 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<Slot>) -> 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<PeerChainTip>,
-    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<Slo
 /// into a stall: the record is the only thing standing between a durable read
 /// position and a lost message.
 ///
-/// `block_hash` is the value [`link_against`] recomputed from the block's own
-/// contents, not `block.header.hash`, which the signature does not cover.
+/// `block_hash` is the value [`screen_peer_block`] recomputed from the block's
+/// own contents, not `block.header.hash`, which the signature does not cover.
 fn record_block_deliveries(
     block: &Block,
     block_hash: HashType,
@@ -658,14 +561,10 @@ fn record_block_deliveries(
 #[cfg(test)]
 mod tests {
     use common::test_utils::produce_dummy_block;
+    use cross_zone::test_utils::{linked_chain_to, ping_emission};
     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, receiver_config_account_id};
     use storage::sequencer::{DB_META_PENDING_CROSS_ZONE_DISPATCHES_KEY, RocksDBIO};
     use tempfile::TempDir;
 
@@ -695,27 +594,9 @@ mod tests {
         emission_to(programs::ping_receiver().id())
     }
 
-    /// A `ping_sender` emission aimed at `target_program_id`. The sender lets its
-    /// caller name any target, which is exactly why the route has to pin the
-    /// pair rather than the target alone.
+    /// A `ping_sender` emission aimed at `target_program_id`.
     fn emission_to(target_program_id: lee_core::program::ProgramId) -> 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<u8>, 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<Block> {
-        let mut blocks: Vec<Block> = 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<u64>) -> (PassOutcome, Option<u64>) {
-        (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

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);