Merge pull request #649 from logos-blockchain/moudy/fix-verifier-forgery-halt

fix(cross-zone): gate the forgery test on a verified peer-chain prefix

Cargo.lock

@@ -4155,6 +4155,7 @@ dependencies = [
  "storage",
  "tempfile",
  "testnet_initial_state",
+ "thiserror 2.0.18",
  "tokio",
  "url",
 ]

lez/indexer/core/Cargo.toml

@@ -37,6 +37,7 @@ async-stream.workspace = true
 tokio.workspace = true
 risc0-zkvm.workspace = true
 hex.workspace = true
+thiserror.workspace = true
 
 [dev-dependencies]
 tempfile.workspace = true

lez/indexer/core/src/cross_zone_verifier.rs

@@ -4,18 +4,18 @@ use std::{
     time::Duration,
 };
 
-use anyhow::{Result, bail};
+use anyhow::anyhow;
 use common::{block::Block, transaction::LeeTransaction};
 use cross_zone::{build_dispatch_from_emission, extract_emission};
 use cross_zone_inbox_core::{
     CrossZoneMessage, Instruction as InboxInstruction, MessageKey, ZoneId, message_key,
 };
-use futures::StreamExt as _;
-use lee::PublicKey;
+use futures::{Stream, StreamExt as _};
+use lee::{GENESIS_BLOCK_ID, PublicKey};
 use log::{debug, error, info};
 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 tokio::sync::RwLock;
 
@@ -25,39 +25,157 @@ use crate::config::IndexerConfig;
 /// so a stuck wait is observable without rejecting a legitimate message.
 const LAG_LOG_INTERVAL: Duration = Duration::from_secs(30);
 
+/// How long to wait for a referenced peer block before giving up on this pass.
+/// Generous, since ordinary L1 finality lag delays a peer block by minutes.
+/// Expiry is not a rejection: the caller retries the same block, so the cost of a
+/// premature expiry is one repeated pass.
+const PEER_BLOCK_WAIT_TIMEOUT: Duration = Duration::from_secs(300);
+
+/// How long each wait iteration sleeps. Also the unit the elapsed counter is
+/// advanced by, so `waited` counts sleeps rather than wall time and, since a
+/// sleep can overshoot, understates it.
+const PEER_BLOCK_POLL_INTERVAL: Duration = Duration::from_secs(1);
+
+/// Consecutive passes a peer reader re-reads the same undecodable slot before
+/// giving up and reading past it.
+const DECODE_RETRY_LIMIT: u32 = 3;
+
+/// Why a cross-zone dispatch could not be verified.
+///
+/// A forgery is terminal and must stop the block applying; an unavailable peer
+/// block is transient and must be retried, or a lagging peer reader would
+/// permanently halt ingestion.
+#[derive(Debug, thiserror::Error)]
+#[non_exhaustive]
+pub enum CrossZoneVerifyError {
+    /// The dispatch does not match the peer's finalized chain.
+    #[error("{0:#}")]
+    Forged(anyhow::Error),
+    /// The referenced peer block has not been read yet.
+    #[error(
+        "peer zone {} block {block_id} still unavailable after {waited:?}",
+        hex::encode(zone)
+    )]
+    PeerUnavailable {
+        zone: ZoneId,
+        block_id: u64,
+        waited: Duration,
+    },
+}
+
+/// One peer zone's cached blocks, plus how far this reader has read them as an
+/// unbroken hash-linked run from the peer's genesis.
+#[derive(Default)]
+struct PeerChain {
+    blocks: HashMap<u64, Block>,
+    /// Highest id such that every block from [`GENESIS_BLOCK_ID`] up to it has
+    /// been read and each links to its predecessor. `None` until genesis is read.
+    ///
+    /// This, not `max(blocks.keys())`, is what the forgery test gates on: a peer
+    /// picks its own `block_id`s, and an id that does not continue the run
+    /// cannot advance the run. It bounds how far this reader has read, not that
+    /// the chain is authentic: the link is self-asserted, since `header.hash` is
+    /// not recomputed on decode and the reader does not apply the pinned-key
+    /// check that [`crate::cross_zone_verifier::CrossZoneVerifier::rederive`]
+    /// applies to the block a dispatch actually names.
+    verified_prefix: Option<u64>,
+}
+
+impl PeerChain {
+    /// The id that would extend the verified run.
+    const fn next_expected(&self) -> u64 {
+        match self.verified_prefix {
+            Some(prefix) => prefix.saturating_add(1),
+            None => GENESIS_BLOCK_ID,
+        }
+    }
+
+    /// Extends the verified run as far as the cached blocks allow.
+    fn extend_prefix(&mut self) {
+        while let Some(next) = self.blocks.get(&self.next_expected()) {
+            let links = match self.verified_prefix {
+                Some(prefix) => self
+                    .blocks
+                    .get(&prefix)
+                    .is_some_and(|prev| prev.header.hash == next.header.prev_block_hash),
+                // Genesis has no predecessor to link to.
+                None => true,
+            };
+            if !links {
+                return;
+            }
+            self.verified_prefix = Some(next.header.block_id);
+        }
+    }
+}
+
+/// What one consistent look at the peer cache says about a referenced block.
+enum PeerLookup {
+    Cached(Box<Block>),
+    /// Inside the verified run but not held, so it is not on the peer chain.
+    InsideRun,
+    /// The reader has not verified this far yet.
+    Behind,
+}
+
 /// Cache of finalized peer-zone blocks, filled by per-peer reader tasks and read
 /// by the verifier to re-derive cross-zone dispatch transactions.
 #[derive(Clone, Default)]
 struct PeerBlocks {
-    chains: Arc<RwLock<HashMap<ZoneId, HashMap<u64, Block>>>>,
+    chains: Arc<RwLock<HashMap<ZoneId, PeerChain>>>,
 }
 
 impl PeerBlocks {
     async fn insert(&self, zone: ZoneId, block: Block) {
-        self.chains
-            .write()
-            .await
-            .entry(zone)
-            .or_default()
-            .insert(block.header.block_id, block);
+        let mut chains = self.chains.write().await;
+        let chain = chains.entry(zone).or_default();
+        chain.blocks.insert(block.header.block_id, block);
+        chain.extend_prefix();
     }
 
+    /// Resolves `block_id` under a single read lock.
+    ///
+    /// Answering "is it cached?" and "is it inside the verified run?" under two
+    /// separate locks races with the peer reader: an insert landing between them
+    /// reads as absent-and-inside-the-run, which is the forgery signal, for a
+    /// block that is in fact cached. That is the normal steady state, a waiting
+    /// verifier and the block it waits for arriving, so it must be one look.
+    async fn resolve(&self, zone: ZoneId, block_id: u64) -> PeerLookup {
+        let chains = self.chains.read().await;
+        let Some(chain) = chains.get(&zone) else {
+            return PeerLookup::Behind;
+        };
+        if let Some(block) = chain.blocks.get(&block_id) {
+            return PeerLookup::Cached(Box::new(block.clone()));
+        }
+        if chain
+            .verified_prefix
+            .is_some_and(|prefix| prefix >= block_id)
+        {
+            PeerLookup::InsideRun
+        } else {
+            PeerLookup::Behind
+        }
+    }
+
+    #[cfg(test)]
     async fn get(&self, zone: ZoneId, block_id: u64) -> Option<Block> {
         self.chains
             .read()
             .await
             .get(&zone)
-            .and_then(|chain| chain.get(&block_id).cloned())
+            .and_then(|chain| chain.blocks.get(&block_id).cloned())
     }
 
-    /// The highest block id this reader has finalized for `zone`, or `None` if it
-    /// has read nothing yet.
-    async fn highest_seen(&self, zone: ZoneId) -> Option<u64> {
+    /// How far this reader has read `zone` as an unbroken run from genesis, or
+    /// `None` if it has not read the peer's genesis block yet.
+    #[cfg(test)]
+    async fn verified_prefix(&self, zone: ZoneId) -> Option<u64> {
         self.chains
             .read()
             .await
             .get(&zone)
-            .and_then(|chain| chain.keys().copied().max())
+            .and_then(|chain| chain.verified_prefix)
     }
 }
 
@@ -115,8 +233,12 @@ impl CrossZoneVerifier {
         })
     }
 
-    /// Verifies every cross-zone dispatch in a block, returning `Err` on the first
-    /// forged dispatch (the caller halts ingestion) or the keys to mark seen.
+    /// Verifies every cross-zone dispatch in a block, returning the keys to mark
+    /// seen, or the reason verification could not complete.
+    ///
+    /// [`CrossZoneVerifyError::Forged`] means the caller must halt ingestion;
+    /// [`CrossZoneVerifyError::PeerUnavailable`] means the caller must hold its
+    /// read cursor and retry, since the block is not yet judged either way.
     ///
     /// The caller MUST record the returned keys via [`Self::record_seen`] only
     /// after the block applies, so the seen-set mirrors the inbox's on-chain
@@ -124,7 +246,10 @@ impl CrossZoneVerifier {
     /// forged dispatch reuse it to skip re-derivation while the inbox delivers the
     /// forgery. A key already seen is a replay the inbox no-ops, so it is accepted
     /// without re-derivation rather than halting on a legitimate re-delivery.
-    pub async fn verify_block(&self, block: &Block) -> Result<Vec<MessageKey>> {
+    pub async fn verify_block(
+        &self,
+        block: &Block,
+    ) -> Result<Vec<MessageKey>, CrossZoneVerifyError> {
         let mut verified = Vec::new();
         for tx in &block.body.transactions {
             let Some(msg) = Self::decode_dispatch(tx) else {
@@ -144,12 +269,12 @@ impl CrossZoneVerifier {
 
             let expected = self.rederive(&msg).await?;
             if LeeTransaction::Public(expected) != *tx {
-                bail!(
+                return Err(CrossZoneVerifyError::Forged(anyhow!(
                     "forged cross-zone dispatch from zone {} block {} tx {}: re-derivation mismatch",
                     hex::encode(msg.src_zone),
                     msg.src_block_id,
                     msg.src_tx_index
-                );
+                )));
             }
 
             info!(
@@ -194,7 +319,10 @@ impl CrossZoneVerifier {
 
     /// Re-derives the dispatch transaction the watcher should have injected for
     /// `msg`, reading the source emission from the peer's finalized block.
-    async fn rederive(&self, msg: &CrossZoneMessage) -> Result<lee::PublicTransaction> {
+    async fn rederive(
+        &self,
+        msg: &CrossZoneMessage,
+    ) -> Result<lee::PublicTransaction, CrossZoneVerifyError> {
         let peer_block = self
             .wait_for_peer_block(msg.src_zone, msg.src_block_id)
             .await?;
@@ -204,35 +332,43 @@ impl CrossZoneVerifier {
         if let Some(expected) = self.peer_pubkeys.get(&msg.src_zone)
             && !peer_block.is_signed_by(expected)
         {
-            bail!(
+            return Err(CrossZoneVerifyError::Forged(anyhow!(
                 "forged cross-zone dispatch: peer zone {} block {} is not signed by the pinned block-signing key",
                 hex::encode(msg.src_zone),
                 msg.src_block_id
-            );
+            )));
         }
 
+        // Everything below is a property of the peer block just read, so a
+        // mismatch is the dispatch lying about it, not a transient condition.
         let emission_tx = peer_block
             .body
             .transactions
             .get(usize::try_from(msg.src_tx_index).expect("u32 index fits in usize"))
             .ok_or_else(|| {
-                anyhow::anyhow!(
+                CrossZoneVerifyError::Forged(anyhow!(
                     "src_tx_index {} out of range in peer block",
                     msg.src_tx_index
-                )
+                ))
             })?;
 
         let LeeTransaction::Public(emission_tx) = emission_tx else {
-            bail!("peer emission transaction is not public");
+            return Err(CrossZoneVerifyError::Forged(anyhow!(
+                "peer emission transaction is not public"
+            )));
         };
         let message = emission_tx.message();
         let emission =
             extract_emission(message.program_id, &message.instruction_data).ok_or_else(|| {
-                anyhow::anyhow!("peer transaction at src_tx_index is not a recognized emitter")
+                CrossZoneVerifyError::Forged(anyhow!(
+                    "peer transaction at src_tx_index is not a recognized emitter"
+                ))
             })?;
 
         if emission.target_zone != self.self_zone {
-            bail!("peer emission targets a different zone");
+            return Err(CrossZoneVerifyError::Forged(anyhow!(
+                "peer emission targets a different zone"
+            )));
         }
 
         Ok(build_dispatch_from_emission(
@@ -248,26 +384,42 @@ impl CrossZoneVerifier {
 
     /// Resolves the referenced peer block, distinguishing forgery from lag.
     ///
-    /// If the block is cached, return it. If our peer reader has already
-    /// finalized past `block_id` and we still do not have it, the reference is to
-    /// a block that does not exist on the peer chain, a forgery, so reject now.
-    async fn wait_for_peer_block(&self, zone: ZoneId, block_id: u64) -> Result<Block> {
+    /// A `block_id` inside the run verified from the peer's genesis (see
+    /// [`PeerChain::verified_prefix`]) that we do not hold does not exist on the
+    /// peer chain, so reject it. Otherwise the reader has not reached it, so
+    /// wait, and give up after [`PEER_BLOCK_WAIT_TIMEOUT`] rather than block
+    /// ingestion forever. A reference to a block the peer will never produce
+    /// stalls rather than being rejected, since that is indistinguishable from a
+    /// peer that has not produced it yet; either way it is never applied.
+    async fn wait_for_peer_block(
+        &self,
+        zone: ZoneId,
+        block_id: u64,
+    ) -> Result<Block, CrossZoneVerifyError> {
         let mut waited = Duration::ZERO;
         loop {
-            if let Some(block) = self.peers.get(zone, block_id).await {
-                return Ok(block);
+            match self.peers.resolve(zone, block_id).await {
+                PeerLookup::Cached(block) => return Ok(*block),
+                // A backstop, not the live path: every id inside the run is
+                // cached by construction. Bounding the cache must preserve that
+                // or track a floor alongside the prefix, since an evicted block
+                // is not a forged one and reporting it as forged would halt a
+                // legitimate dispatch.
+                PeerLookup::InsideRun => {
+                    return Err(CrossZoneVerifyError::Forged(anyhow!(
+                        "forged cross-zone reference: peer zone {} chain is verified past block {} but it is absent",
+                        hex::encode(zone),
+                        block_id
+                    )));
+                }
+                PeerLookup::Behind => {}
             }
-            if self
-                .peers
-                .highest_seen(zone)
-                .await
-                .is_some_and(|h| h >= block_id)
-            {
-                bail!(
-                    "forged cross-zone reference: peer zone {} finalized past block {} but it is absent",
-                    hex::encode(zone),
-                    block_id
-                );
+            if waited >= PEER_BLOCK_WAIT_TIMEOUT {
+                return Err(CrossZoneVerifyError::PeerUnavailable {
+                    zone,
+                    block_id,
+                    waited,
+                });
             }
             if !waited.is_zero() && waited.as_secs().is_multiple_of(LAG_LOG_INTERVAL.as_secs()) {
                 info!(
@@ -277,12 +429,21 @@ impl CrossZoneVerifier {
                     waited.as_secs()
                 );
             }
-            tokio::time::sleep(Duration::from_secs(1)).await;
-            waited = waited.saturating_add(Duration::from_secs(1));
+            tokio::time::sleep(PEER_BLOCK_POLL_INTERVAL).await;
+            waited = waited.saturating_add(PEER_BLOCK_POLL_INTERVAL);
         }
     }
 }
 
+/// The outcome of one pass over a peer's message stream.
+#[derive(Debug, PartialEq, Eq)]
+struct PeerPass {
+    /// Where the next pass resumes from.
+    cursor: Option<Slot>,
+    /// Set when the pass ended early on a message that would not decode.
+    stalled_at: Option<Slot>,
+}
+
 /// Reads a peer zone's finalized blocks from Bedrock into the shared cache.
 #[expect(
     clippy::infinite_loop,
@@ -300,41 +461,124 @@ async fn read_peer(
     );
 
     let mut cursor = None;
+    // The slot the reader is stuck on and how many passes it has spent there.
+    // Keyed by slot: the retry budget is per slot, so a failure at a new slot
+    // must not inherit an older slot's count and be skipped on its first try.
+    let mut stalled: Option<(Slot, u32)> = None;
+    let mut skip_slot = None;
     loop {
-        let stream = match zone_indexer.next_messages(cursor).await {
-            Ok(stream) => stream,
-            Err(err) => {
-                error!(
-                    "Peer reader next_messages failed for {}: {err}",
-                    hex::encode(peer_zone)
-                );
-                tokio::time::sleep(poll_interval).await;
-                continue;
-            }
-        };
-        let mut stream = std::pin::pin!(stream);
-
-        while let Some((msg, slot)) = stream.next().await {
-            if let ZoneMessage::Block(zone_block) = msg {
-                match borsh::from_slice::<Block>(&zone_block.data) {
-                    Ok(block) => peers.insert(peer_zone, block).await,
-                    Err(err) => error!("Peer reader failed to deserialize block: {err}"),
+        match zone_indexer.next_messages(cursor).await {
+            Ok(stream) => {
+                let pass = consume_peer_stream(stream, peer_zone, &peers, cursor, skip_slot).await;
+                cursor = pass.cursor;
+                if let Some(slot) = pass.stalled_at {
+                    let attempts = match stalled {
+                        Some((prev, attempts)) if prev == slot => attempts.saturating_add(1),
+                        _ => 1,
+                    };
+                    if attempts >= DECODE_RETRY_LIMIT {
+                        // Reading on leaves a hole: dispatches referencing the
+                        // skipped block can no longer be verified, but every
+                        // later block stays readable.
+                        error!(
+                            "Peer reader for {} could not decode slot {slot:?} after {attempts} attempts; reading past it.",
+                            hex::encode(peer_zone)
+                        );
+                        skip_slot = Some(slot);
+                        stalled = None;
+                    } else {
+                        stalled = Some((slot, attempts));
+                    }
+                } else {
+                    stalled = None;
+                    skip_slot = None;
                 }
             }
-            cursor = Some(slot);
+            Err(err) => error!(
+                "Peer reader next_messages failed for {}: {err}",
+                hex::encode(peer_zone)
+            ),
         }
 
         tokio::time::sleep(poll_interval).await;
     }
 }
 
+/// Caches the finalized peer blocks carried by `stream`.
+///
+/// A block that fails to deserialize ends the pass and holds the cursor at the
+/// last fully-consumed slot, so the next poll re-reads it and a transient
+/// failure heals itself. `skip_slot` names a slot the caller gave up on after
+/// [`DECODE_RETRY_LIMIT`] attempts, which is read past instead so a permanently
+/// undecodable inscription cannot wedge the reader. Skipping only leaves a hole,
+/// which cannot advance [`PeerChain::verified_prefix`] past itself.
+///
+/// The cursor advances only on a slot boundary, since one slot can carry several
+/// messages and resuming mid-slot would skip the ones after the failure. This
+/// relies on the stream never truncating mid-slot, which holds because the
+/// zone-sdk materializes a whole batch before yielding and batches end on slot
+/// boundaries.
+async fn consume_peer_stream<S>(
+    stream: S,
+    peer_zone: ZoneId,
+    peers: &PeerBlocks,
+    resume_from: Option<Slot>,
+    skip_slot: Option<Slot>,
+) -> PeerPass
+where
+    S: Stream<Item = (ZoneMessage, Slot)>,
+{
+    let mut stream = std::pin::pin!(stream);
+    let mut cursor = resume_from;
+    // The slot being consumed: cached so far, but there may be more to come.
+    let mut in_progress: Option<Slot> = None;
+
+    while let Some((msg, slot)) = stream.next().await {
+        if in_progress != Some(slot) {
+            cursor = in_progress.or(cursor);
+            in_progress = Some(slot);
+        }
+
+        let ZoneMessage::Block(zone_block) = msg else {
+            continue;
+        };
+        match borsh::from_slice::<Block>(&zone_block.data) {
+            Ok(block) => peers.insert(peer_zone, block).await,
+            Err(err) if skip_slot == Some(slot) => {
+                debug!(
+                    "Peer reader skipping undecodable block from {} at slot {slot:?}: {err}",
+                    hex::encode(peer_zone)
+                );
+            }
+            Err(err) => {
+                error!(
+                    "Peer reader failed to deserialize block from {} at slot {slot:?}: {err}. Holding the cursor and retrying.",
+                    hex::encode(peer_zone)
+                );
+                return PeerPass {
+                    cursor,
+                    stalled_at: Some(slot),
+                };
+            }
+        }
+    }
+
+    PeerPass {
+        cursor: in_progress.or(cursor),
+        stalled_at: None,
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use common::test_utils::produce_dummy_block;
+    use futures::stream;
     use lee::{
         PrivateKey, PublicKey, 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 super::*;
@@ -375,6 +619,44 @@ mod tests {
         ))
     }
 
+    /// A peer-stream item inscribing `data` at `slot`.
+    fn peer_msg(data: Vec<u8>, 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 hash-linked chain of `len` blocks from genesis, each carrying a `b"hi"`
+    /// emission. Only a chain built this way advances the verified prefix.
+    fn linked_chain(len: u64) -> Vec<Block> {
+        let mut prev = None;
+        let mut blocks = Vec::new();
+        for offset in 0..len {
+            let block = produce_dummy_block(
+                GENESIS_BLOCK_ID.saturating_add(offset),
+                prev,
+                vec![emission(b"hi")],
+            );
+            prev = Some(block.header.hash);
+            blocks.push(block);
+        }
+        blocks
+    }
+
+    /// A peer-stream item carrying `block`.
+    fn peer_block_msg(block: &Block, slot: u64) -> (ZoneMessage, Slot) {
+        peer_msg(borsh::to_vec(block).expect("block serializes"), slot)
+    }
+
+    /// A peer-stream item whose inscription is not a decodable block.
+    fn undecodable_msg(slot: u64) -> (ZoneMessage, Slot) {
+        peer_msg(b"not a block".to_vec(), slot)
+    }
+
     /// The dispatch a watcher would inject for a `PEER_BLOCK_ID` emission of `payload`.
     fn dispatch(payload: &[u8]) -> LeeTransaction {
         let receiver_id = programs::ping_receiver().id();
@@ -472,28 +754,6 @@ mod tests {
         );
     }
 
-    #[tokio::test]
-    async fn rejects_reference_to_a_block_the_peer_never_finalized() {
-        let verifier = verifier();
-        // The reader has finalized past PEER_BLOCK_ID (it holds a later block) but
-        // never saw PEER_BLOCK_ID itself, so a dispatch referencing it is a forgery
-        // and must be rejected rather than waited on forever.
-        verifier
-            .peers
-            .insert(
-                PEER_ZONE,
-                produce_dummy_block(PEER_BLOCK_ID + 1, None, vec![emission(b"hi")]),
-            )
-            .await;
-
-        let block = produce_dummy_block(9, None, vec![dispatch(b"hi")]);
-        let err = verifier.verify_block(&block).await.unwrap_err();
-        assert!(
-            err.to_string().contains("forged"),
-            "unexpected error: {err}"
-        );
-    }
-
     #[tokio::test]
     async fn accepts_replayed_dispatch_as_noop() {
         let verifier = verifier();
@@ -565,4 +825,208 @@ mod tests {
             "unexpected error: {err}"
         );
     }
+
+    #[tokio::test]
+    async fn peer_reader_advances_over_a_fully_decoded_stream() {
+        let peers = PeerBlocks::default();
+        let chain = linked_chain(2);
+        let stream = stream::iter(vec![
+            peer_block_msg(&chain[0], 0),
+            peer_block_msg(&chain[1], 1),
+        ]);
+
+        let pass = consume_peer_stream(stream, PEER_ZONE, &peers, None, None).await;
+
+        assert_eq!(pass.cursor, Some(Slot::from(1)));
+        assert_eq!(pass.stalled_at, None);
+        assert_eq!(peers.verified_prefix(PEER_ZONE).await, Some(2));
+    }
+
+    #[tokio::test]
+    async fn a_block_that_does_not_link_does_not_extend_the_verified_run() {
+        let peers = PeerBlocks::default();
+        let genesis = linked_chain(1);
+        peers.insert(PEER_ZONE, genesis[0].clone()).await;
+        // Claims the next id, but not the predecessor it would have to follow.
+        peers
+            .insert(
+                PEER_ZONE,
+                produce_dummy_block(GENESIS_BLOCK_ID + 1, None, vec![emission(b"hi")]),
+            )
+            .await;
+
+        assert_eq!(
+            peers.verified_prefix(PEER_ZONE).await,
+            Some(GENESIS_BLOCK_ID)
+        );
+    }
+
+    #[tokio::test]
+    async fn a_block_arriving_between_the_two_halves_of_a_lookup_is_not_forged() {
+        // `resolve` answers "cached?" and "inside the verified run?" under one
+        // lock. Split across two, an insert landing between them reads as
+        // absent-and-inside-the-run, the forgery signal, for a cached block.
+        let peers = PeerBlocks::default();
+        for block in linked_chain(2) {
+            peers.insert(PEER_ZONE, block).await;
+        }
+
+        assert!(matches!(
+            peers.resolve(PEER_ZONE, 2).await,
+            PeerLookup::Cached(_)
+        ));
+        assert!(matches!(
+            peers.resolve(PEER_ZONE, 3).await,
+            PeerLookup::Behind
+        ));
+    }
+
+    #[tokio::test]
+    async fn peer_reader_holds_its_cursor_on_an_undecodable_block() {
+        let peers = PeerBlocks::default();
+        let chain = linked_chain(3);
+        let stream = stream::iter(vec![
+            peer_block_msg(&chain[0], 0),
+            undecodable_msg(1),
+            peer_block_msg(&chain[2], 2),
+        ]);
+
+        let pass = consume_peer_stream(stream, PEER_ZONE, &peers, None, None).await;
+
+        assert_eq!(pass.cursor, Some(Slot::from(0)));
+        assert_eq!(pass.stalled_at, Some(Slot::from(1)));
+        assert_eq!(peers.verified_prefix(PEER_ZONE).await, Some(1));
+        assert!(peers.get(PEER_ZONE, 3).await.is_none());
+    }
+
+    #[tokio::test]
+    async fn peer_reader_does_not_resume_inside_a_partially_failed_slot() {
+        let peers = PeerBlocks::default();
+        let chain = linked_chain(1);
+        // One slot can carry several messages; the second one fails.
+        let stream = stream::iter(vec![peer_block_msg(&chain[0], 7), undecodable_msg(7)]);
+
+        let pass = consume_peer_stream(stream, PEER_ZONE, &peers, Some(Slot::from(6)), None).await;
+
+        // Slot 7 is re-read whole next pass, not resumed past the failure.
+        assert_eq!(pass.cursor, Some(Slot::from(6)));
+    }
+
+    #[tokio::test]
+    async fn peer_reader_reads_past_a_slot_it_has_given_up_on() {
+        // After DECODE_RETRY_LIMIT attempts the caller nominates the slot to
+        // skip, so a permanently undecodable inscription cannot wedge the reader.
+        let peers = PeerBlocks::default();
+        let chain = linked_chain(3);
+        let stream = stream::iter(vec![
+            peer_block_msg(&chain[0], 0),
+            undecodable_msg(1),
+            peer_block_msg(&chain[2], 2),
+        ]);
+
+        let pass = consume_peer_stream(stream, PEER_ZONE, &peers, None, Some(Slot::from(1))).await;
+
+        assert_eq!(pass.cursor, Some(Slot::from(2)), "the pass drains");
+        assert_eq!(pass.stalled_at, None);
+        // Block 3 is cached and servable, so dispatches referencing it verify.
+        assert!(peers.get(PEER_ZONE, 3).await.is_some());
+        // But the hole stops the verified run, so block 2 is never called forged.
+        assert_eq!(peers.verified_prefix(PEER_ZONE).await, Some(1));
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn undecodable_peer_block_does_not_make_a_later_dispatch_look_forged() {
+        let verifier = verifier();
+        let chain = linked_chain(3);
+        let stream = stream::iter(vec![
+            peer_block_msg(&chain[0], 0),
+            undecodable_msg(1),
+            peer_block_msg(&chain[2], 2),
+        ]);
+        consume_peer_stream(
+            stream,
+            PEER_ZONE,
+            &verifier.peers,
+            None,
+            Some(Slot::from(1)),
+        )
+        .await;
+
+        // Regression: the reader cached block 3, so the old `max(cached ids)`
+        // high-water mark reached 3 and a dispatch referencing block 2 was
+        // rejected as forged, halting ingestion permanently.
+        let err = verifier
+            .wait_for_peer_block(PEER_ZONE, 2)
+            .await
+            .expect_err("block 2 was never read, so it cannot be resolved");
+        assert!(
+            matches!(err, CrossZoneVerifyError::PeerUnavailable { .. }),
+            "a block outside the verified run is lag, not forgery: {err}"
+        );
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn a_high_block_id_cannot_poison_the_forgery_test() {
+        // A peer picks its own block ids, so one inscribed block claiming a huge
+        // id would drive a `max(cached ids)` high-water mark past every real id
+        // and make each later dispatch look forged. It cannot extend the
+        // verified run, so it is inert.
+        let verifier = verifier();
+        let chain = linked_chain(2);
+        verifier.peers.insert(PEER_ZONE, chain[0].clone()).await;
+        verifier.peers.insert(PEER_ZONE, chain[1].clone()).await;
+        verifier
+            .peers
+            .insert(
+                PEER_ZONE,
+                produce_dummy_block(u64::MAX, None, vec![emission(b"hi")]),
+            )
+            .await;
+
+        assert_eq!(verifier.peers.verified_prefix(PEER_ZONE).await, Some(2));
+        let err = verifier
+            .wait_for_peer_block(PEER_ZONE, 3)
+            .await
+            .expect_err("block 3 has not been read yet");
+        assert!(
+            matches!(err, CrossZoneVerifyError::PeerUnavailable { .. }),
+            "a block beyond the verified run is lag, not forgery: {err}"
+        );
+    }
+
+    #[tokio::test]
+    async fn a_transient_decode_failure_heals_on_the_next_pass() {
+        let verifier = verifier();
+        let chain = linked_chain(PEER_BLOCK_ID);
+
+        let pass = consume_peer_stream(
+            stream::iter(vec![undecodable_msg(0)]),
+            PEER_ZONE,
+            &verifier.peers,
+            None,
+            None,
+        )
+        .await;
+        assert_eq!(pass.cursor, None, "the failed slot is not skipped");
+        assert_eq!(pass.stalled_at, Some(Slot::from(0)));
+
+        // The next pass re-reads the same slot, which now decodes.
+        let pass = consume_peer_stream(
+            stream::iter(chain.iter().enumerate().map(|(index, block)| {
+                peer_block_msg(block, u64::try_from(index).expect("test index fits in u64"))
+            })),
+            PEER_ZONE,
+            &verifier.peers,
+            pass.cursor,
+            None,
+        )
+        .await;
+        assert_eq!(pass.stalled_at, None);
+
+        let block = produce_dummy_block(9, None, vec![dispatch(b"hi")]);
+        verifier
+            .verify_block(&block)
+            .await
+            .expect("the dispatch verifies once the peer block has been read");
+    }
 }

lez/indexer/core/src/lib.rs

@@ -16,7 +16,7 @@ use retry::ApplyRetryGate;
 use crate::{
     block_store::IndexerStore,
     config::IndexerConfig,
-    cross_zone_verifier::CrossZoneVerifier,
+    cross_zone_verifier::{CrossZoneVerifier, CrossZoneVerifyError},
     status::{IndexerStatus, IndexerSyncStatus},
 };
 
@@ -29,6 +29,16 @@ pub mod status;
 /// Consecutive failed apply attempts of the same block before parking.
 const APPLY_RETRY_LIMIT: u32 = 3;
 
+/// Which slot the ingest loop is currently inside, so the read cursor only ever
+/// moves on a slot boundary.
+///
+/// One L1 slot can carry several L2 blocks, and the channel stream resumes
+/// *after* the stored slot. Advancing the cursor as each block is handled would
+/// therefore put a later block in the same slot beyond the cursor whenever a
+/// pass ends early, and nothing would ever read it again.
+#[derive(Default)]
+struct SlotProgress(Option<Slot>);
+
 #[derive(Clone)]
 pub struct IndexerCore {
     pub zone_indexer: Arc<ZoneIndexer<NodeHttpClient>>,
@@ -42,6 +52,23 @@ pub struct IndexerCore {
     pub verifier: Option<CrossZoneVerifier>,
 }
 
+impl SlotProgress {
+    /// Records that a message from `slot` is being handled, returning the slot
+    /// that just completed, if this message begins a new one.
+    fn enter(&mut self, slot: Slot) -> Option<Slot> {
+        if self.0 == Some(slot) {
+            return None;
+        }
+        self.0.replace(slot)
+    }
+
+    /// The slot in progress when the stream drained cleanly, which is therefore
+    /// complete. Not called when a pass ends early: that slot must be re-read.
+    const fn drained(self) -> Option<Slot> {
+        self.0
+    }
+}
+
 impl IndexerCore {
     /// Builds the core, then verifies the stored chain matches the channel's by
     /// re-reading the channel at the stored tip's position.
@@ -264,8 +291,20 @@ impl IndexerCore {
 
                 let mut announced_syncing = false;
                 let mut had_cycle_error = false;
+                // The slot being consumed: every message of it seen so far is
+                // handled, but another may follow, so the cursor may not move
+                // onto it yet. One L1 slot can carry several L2 blocks, and the
+                // stream resumes *after* the stored slot, so advancing inside a
+                // slot would put a later message in it beyond the cursor
+                // for ever if this pass ends early.
+                let mut in_progress = SlotProgress::default();
 
                 while let Some((msg, slot)) = stream.next().await {
+                    // A message from a later slot means the previous one is complete.
+                    if let Some(done) = in_progress.enter(slot) {
+                        self.advance_cursor(&mut cursor, done);
+                    }
+
                     if !announced_syncing {
                         self.set_status(IndexerSyncStatus::syncing());
                         announced_syncing = true;
@@ -286,30 +325,43 @@ impl IndexerCore {
                                 break;
                             }
                             // L1 proceeds regardless
-                            self.advance_cursor(&mut cursor, slot);
                             continue;
                         }
                     };
 
-                    // Option B: re-derive and verify every cross-zone dispatch
-                    // before applying the block. A forged dispatch halts ingestion
-                    // rather than persisting an invalid state; a replay is accepted
-                    // since the inbox no-ops it on chain. The verified keys are
-                    // marked seen only once the block applies (below), so a block
-                    // that does not apply cannot poison the seen-set.
+                    // Re-derive and verify every cross-zone dispatch the block
+                    // carries before applying it, so the destination never trusts
+                    // a dispatch just because a sequencer signed the block: a
+                    // forged one halts ingestion rather than persisting invalid
+                    // state, while a replay is accepted since the inbox no-ops it
+                    // on chain. The verified keys are marked seen only once the
+                    // block applies (below), so a block that does not apply
+                    // cannot poison the seen-set.
                     let verified_keys = match &self.verifier {
                         Some(verifier) => match verifier.verify_block(&block).await {
                             Ok(keys) => keys,
-                            Err(err) => {
+                            Err(err @ CrossZoneVerifyError::Forged(_)) => {
                                 error!(
-                                    "Cross-zone verification failed for block {}: {err:#}. Halting indexer ingestion.",
+                                    "Cross-zone verification failed for block {}: {err}. Halting indexer ingestion.",
                                     block.header.block_id
                                 );
                                 self.set_status(IndexerSyncStatus::error(format!(
-                                    "cross-zone verification failed: {err:#}"
+                                    "cross-zone verification failed: {err}"
                                 )));
                                 return;
                             }
+                            // Not judged either way yet, so retry rather than halt.
+                            Err(err @ CrossZoneVerifyError::PeerUnavailable { .. }) => {
+                                error!(
+                                    "Cross-zone verification of block {} stalled: {err}. Holding the cursor and retrying.",
+                                    block.header.block_id
+                                );
+                                self.set_status(IndexerSyncStatus::error(format!(
+                                    "cross-zone peer unavailable: {err}"
+                                )));
+                                had_cycle_error = true;
+                                break;
+                            }
                         },
                         None => Vec::new(),
                     };
@@ -322,7 +374,6 @@ impl IndexerCore {
                             retry_gate.reset();
                             info!("Indexed L2 block {}", block.header.block_id);
                             self.set_status(IndexerSyncStatus::syncing());
-                            self.advance_cursor(&mut cursor, slot);
                             yield Ok(block);
                         }
                         Ok(AcceptOutcome::AlreadyApplied) => {
@@ -330,7 +381,6 @@ impl IndexerCore {
                                 "Skipping already-applied block {}",
                                 block.header.block_id
                             );
-                            self.advance_cursor(&mut cursor, slot);
                         }
                         Ok(AcceptOutcome::Parked(ingest_err)) => {
                             error!(
@@ -339,7 +389,6 @@ impl IndexerCore {
                             );
                             self.set_status(IndexerSyncStatus::stalled(ingest_err.to_string()));
                             // L1 proceeds regardless
-                            self.advance_cursor(&mut cursor, slot);
                         }
                         Ok(AcceptOutcome::RetryableFailure(ingest_err)) => {
                             let attempts = retry_gate.register_failure(block.header.block_id);
@@ -365,7 +414,6 @@ impl IndexerCore {
                                     break;
                                 }
                                 self.set_status(IndexerSyncStatus::stalled(ingest_err.to_string()));
-                                self.advance_cursor(&mut cursor, slot);
                                 retry_gate.reset();
                             } else {
                                 error!(
@@ -396,10 +444,17 @@ impl IndexerCore {
                 }
 
                 if had_cycle_error {
+                    // The slot in progress is not finished, so the cursor stays
+                    // below it and the next pass re-reads it whole.
                     tokio::time::sleep(poll_interval).await;
                     continue;
                 }
 
+                // The stream drained cleanly, so the slot in progress completed too.
+                if let Some(done) = in_progress.drained() {
+                    self.advance_cursor(&mut cursor, done);
+                }
+
                 // Stream drained. Stay Stalled if parked; otherwise we are caught up.
                 // A store error here must not be collapsed to "no stall recorded":
                 // that would wrongly flip us to caught-up, so we log and hold state.
@@ -426,6 +481,51 @@ mod tests {
     use super::*;
     use crate::config::{ChannelId, ClientConfig, IndexerConfig};
 
+    /// The cursor must not move while more of the same slot may still arrive.
+    ///
+    /// Two L2 blocks in one L1 slot: the first applies, the second stalls on an
+    /// unavailable peer and the pass retries. If handling the first had advanced
+    /// the cursor onto the slot, the retry would resume past it and the second
+    /// block would never be read again, silently losing whatever it carried.
+    #[test]
+    fn a_slot_is_only_left_behind_once_it_is_finished() {
+        let mut progress = SlotProgress::default();
+        let slot = Slot::from(7);
+
+        assert_eq!(
+            progress.enter(slot),
+            None,
+            "nothing precedes the first slot"
+        );
+        assert_eq!(
+            progress.enter(slot),
+            None,
+            "a second message in the same slot must not release it"
+        );
+
+        // The pass ends early here, so `drained` is never called and the cursor
+        // is still below slot 7: the next pass re-reads it whole.
+    }
+
+    #[test]
+    fn a_completed_slot_is_released_when_the_next_one_starts() {
+        let mut progress = SlotProgress::default();
+
+        assert_eq!(progress.enter(Slot::from(3)), None);
+        assert_eq!(progress.enter(Slot::from(3)), None);
+        assert_eq!(
+            progress.enter(Slot::from(4)),
+            Some(Slot::from(3)),
+            "slot 3 is complete once a message from slot 4 arrives"
+        );
+        assert_eq!(progress.drained(), Some(Slot::from(4)));
+    }
+
+    #[test]
+    fn draining_an_untouched_stream_releases_nothing() {
+        assert_eq!(SlotProgress::default().drained(), None);
+    }
+
     fn unreachable_core(dir: &std::path::Path) -> IndexerCore {
         let config = IndexerConfig {
             consensus_info_polling_interval: Duration::from_secs(1),