diff --git a/Cargo.lock b/Cargo.lock index b470cdb3..fa99c5d9 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4155,6 +4155,7 @@ dependencies = [ "storage", "tempfile", "testnet_initial_state", + "thiserror 2.0.18", "tokio", "url", ] diff --git a/lez/indexer/core/Cargo.toml b/lez/indexer/core/Cargo.toml index afba6f29..c8c8590f 100644 --- a/lez/indexer/core/Cargo.toml +++ b/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 diff --git a/lez/indexer/core/src/cross_zone_verifier.rs b/lez/indexer/core/src/cross_zone_verifier.rs index 5d95df4e..da568ab8 100644 --- a/lez/indexer/core/src/cross_zone_verifier.rs +++ b/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, + /// 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, +} + +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), + /// 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>>>, + chains: Arc>>, } 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 { 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 { + /// 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 { 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> { + pub async fn verify_block( + &self, + block: &Block, + ) -> Result, 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 { + async fn rederive( + &self, + msg: &CrossZoneMessage, + ) -> Result { 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 { + /// 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 { 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, + /// Set when the pass ended early on a message that would not decode. + stalled_at: Option, +} + /// 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::(&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( + stream: S, + peer_zone: ZoneId, + peers: &PeerBlocks, + resume_from: Option, + skip_slot: Option, +) -> PeerPass +where + S: Stream, +{ + 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 = 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::(&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, 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 { + 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"); + } } diff --git a/lez/indexer/core/src/lib.rs b/lez/indexer/core/src/lib.rs index 4cf7d58f..ccf8dbe7 100644 --- a/lez/indexer/core/src/lib.rs +++ b/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); + #[derive(Clone)] pub struct IndexerCore { pub zone_indexer: Arc>, @@ -42,6 +52,23 @@ pub struct IndexerCore { pub verifier: Option, } +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 { + 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 { + 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),