From 39c19912c5c6206d8852a482698c31e44cc6b20d Mon Sep 17 00:00:00 2001 From: erhant Date: Fri, 3 Jul 2026 22:36:42 +0300 Subject: [PATCH] fix(indexer): use better error types, tend to few @Arjentix PR review --- .../docker-all-in-one/indexer_config.json | 3 +- lez/indexer/core/src/block_store.rs | 66 ++++++++------- lez/indexer/core/src/ingest_error.rs | 39 +++++---- lez/indexer/core/src/lib.rs | 83 +++++++++++-------- lez/indexer/core/src/status.rs | 5 +- lez/indexer/service/protocol/src/convert.rs | 11 ++- lez/indexer/service/protocol/src/lib.rs | 9 +- 7 files changed, 136 insertions(+), 80 deletions(-) diff --git a/lez/configs/docker-all-in-one/indexer_config.json b/lez/configs/docker-all-in-one/indexer_config.json index c1ff65b0..5791f64a 100644 --- a/lez/configs/docker-all-in-one/indexer_config.json +++ b/lez/configs/docker-all-in-one/indexer_config.json @@ -3,5 +3,6 @@ "bedrock_config": { "addr": "http://logos-blockchain-node-0:18080" }, - "channel_id": "0101010101010101010101010101010101010101010101010101010101010101" + "channel_id": "0101010101010101010101010101010101010101010101010101010101010101", + "allow_chain_reset": true } diff --git a/lez/indexer/core/src/block_store.rs b/lez/indexer/core/src/block_store.rs index 3625396c..7269f687 100644 --- a/lez/indexer/core/src/block_store.rs +++ b/lez/indexer/core/src/block_store.rs @@ -186,9 +186,11 @@ impl IndexerStore { })) } - /// Records the stall reason: the first break is stored verbatim; subsequent - /// breaks only bump `orphans_since`, preserving the original cause. - fn record_stall( + /// Record the stall reason. + /// + /// - First stall is stored verbatim + /// - Subsequent stalls only bump `orphans_since`, preserving the original cause. + pub fn record_stall( &self, header: Option<&BlockHeader>, l1_slot: Slot, @@ -200,6 +202,7 @@ impl IndexerStore { existing } None => StallReason { + // need to map out of `header` because they are not ser/de block_id: header.map(|h| h.block_id), block_hash: header.map(|h| h.hash), prev_block_hash: header.map(|h| h.prev_block_hash), @@ -212,11 +215,6 @@ impl IndexerStore { self.set_stall_reason(&Some(stall)) } - /// Records a stall for an inscription that could not even be parsed. - pub fn record_deserialize_stall(&self, l1_slot: Slot, error: String) -> Result<()> { - self.record_stall(None, l1_slot, BlockIngestError::Deserialize(error)) - } - /// Validates `block` against the tip and, if it chains, applies it atomically /// (scratch clone, commit only on full success) and advances the tip. On any /// failure records the stall and returns `Parked` without touching state. @@ -306,25 +304,26 @@ fn validate_against_tip(tip: Option<&Tip>, block: &Block) -> Result<(), BlockIng /// [`BlockIngestError`] so the caller can park rather than crash. Operates on a /// scratch state; the caller commits only on `Ok`. fn apply_block_to_scratch(block: &Block, state: &mut V03State) -> Result<(), BlockIngestError> { - let (clock_tx, user_txs) = - block.body.transactions.split_last().ok_or_else(|| { - BlockIngestError::StateTransition("block has no transactions".to_owned()) - })?; + let (clock_tx, user_txs) = block + .body + .transactions + .split_last() + .ok_or(BlockIngestError::EmptyBlock)?; let expected_clock = LeeTransaction::Public(clock_invocation(block.header.timestamp)); if *clock_tx != expected_clock { - return Err(BlockIngestError::StateTransition( - "last transaction must be the clock invocation for the block timestamp".to_owned(), - )); + return Err(BlockIngestError::InvalidClockTransaction); } let is_genesis = block.header.block_id == GENESIS_BLOCK_ID; - for transaction in user_txs { + for (tx_index, transaction) in user_txs.iter().enumerate() { + let state_transition = |err: anyhow::Error| BlockIngestError::StateTransition { + tx_index: tx_index.try_into().expect("tx index fits in u64"), + reason: format!("{err:#}"), + }; if is_genesis { let LeeTransaction::Public(public_tx) = transaction else { - return Err(BlockIngestError::StateTransition( - "genesis block should contain only public transactions".to_owned(), - )); + return Err(BlockIngestError::NonPublicGenesisTransaction); }; state .transition_from_public_transaction( @@ -332,19 +331,17 @@ fn apply_block_to_scratch(block: &Block, state: &mut V03State) -> Result<(), Blo block.header.block_id, block.header.timestamp, ) - .map_err(|err| BlockIngestError::StateTransition(format!("{err:?}")))?; + .map_err(|err| state_transition(err.into()))?; } else { transaction .clone() .execute_on_state(state, block.header.block_id, block.header.timestamp) - .map_err(|err| BlockIngestError::StateTransition(format!("{err:?}")))?; + .map_err(|err| state_transition(err.into()))?; } } let LeeTransaction::Public(clock_public_tx) = clock_tx else { - return Err(BlockIngestError::StateTransition( - "clock invocation must be a public transaction".to_owned(), - )); + return Err(BlockIngestError::InvalidClockTransaction); }; state .transition_from_public_transaction( @@ -352,7 +349,10 @@ fn apply_block_to_scratch(block: &Block, state: &mut V03State) -> Result<(), Blo block.header.block_id, block.header.timestamp, ) - .map_err(|err| BlockIngestError::StateTransition(format!("{err:?}")))?; + .map_err(|err| BlockIngestError::StateTransition { + tx_index: user_txs.len().try_into().expect("tx index fits in u64"), + reason: format!("{:#}", anyhow::Error::from(err)), + })?; Ok(()) } @@ -376,7 +376,10 @@ mod stall_reason_tests { block_hash: Some(HashType([1_u8; 32])), prev_block_hash: Some(HashType([2_u8; 32])), l1_slot: Slot::from(42), - error: BlockIngestError::StateTransition("boom".to_owned()), + error: BlockIngestError::StateTransition { + tx_index: 0, + reason: "boom".to_owned(), + }, first_seen: Some(99), orphans_since: 3, }; @@ -385,7 +388,10 @@ mod stall_reason_tests { let got = store.get_stall_reason().expect("get").expect("present"); assert_eq!(got.block_id, Some(7)); assert_eq!(got.orphans_since, 3); - assert!(matches!(got.error, BlockIngestError::StateTransition(_))); + assert!(matches!( + got.error, + BlockIngestError::StateTransition { .. } + )); assert_eq!(got.block_hash, Some(HashType([1_u8; 32]))); assert_eq!(got.prev_block_hash, Some(HashType([2_u8; 32]))); assert_eq!(got.l1_slot, Slot::from(42)); @@ -591,7 +597,11 @@ mod accept_tests { let store = IndexerStore::open_db(dir.path()).expect("open store"); store - .record_deserialize_stall(Slot::from(0), "bad bytes".to_owned()) + .record_stall( + None, + Slot::from(0), + BlockIngestError::Deserialize("bad bytes".to_owned()), + ) .expect("record"); let stall = store.get_stall_reason().expect("get").expect("present"); diff --git a/lez/indexer/core/src/ingest_error.rs b/lez/indexer/core/src/ingest_error.rs index 1235120a..bda53497 100644 --- a/lez/indexer/core/src/ingest_error.rs +++ b/lez/indexer/core/src/ingest_error.rs @@ -1,26 +1,43 @@ use common::HashType; use serde::{Deserialize, Serialize}; -/// Why the indexer could not apply an L2 block from the channel. Stored inside a -/// [`crate::stall_reason::StallReason`] and surfaced on the status snapshot. +/// Why the indexer could not apply an L2 block from the channel. +/// +/// Persisted in `RocksDB`, so every variant must have the following +/// traits: `Clone + Serialize + Deserialize`. #[derive(Debug, Clone, Serialize, Deserialize, thiserror::Error)] pub enum BlockIngestError { - #[error("failed to deserialize L2 block: {0}")] + #[error("Failed to deserialize L2 block: {0}")] + /// Here we store the error string that is derived from [`borsh::from_slice`]'s [`Err`]. Deserialize(String), - #[error("unexpected block id: expected {expected}, got {got}")] + #[error("Unexpected block id: expected {expected}, got {got}")] UnexpectedBlockId { expected: u64, got: u64 }, - #[error("broken chain link: expected prev {expected_prev}, got {got_prev}")] + #[error("Broken chain link: expected prev {expected_prev}, got {got_prev}")] BrokenChainLink { expected_prev: HashType, got_prev: HashType, }, - #[error("block hash mismatch: computed {computed}, header {header}")] + #[error("Block hash mismatch: computed {computed}, header {header}")] HashMismatch { computed: HashType, header: HashType, }, - #[error("state transition failed: {0}")] - StateTransition(String), + #[error("Block has no transactions")] + EmptyBlock, + #[error("Last transaction must be the public clock invocation for the block timestamp")] + InvalidClockTransaction, + #[error("Genesis block must contain only public transactions")] + NonPublicGenesisTransaction, + #[error("State transition failed at transaction {tx_index}: {reason}")] + StateTransition { + /// Index of the failing transaction within the block body. + tx_index: u64, + /// Reason string from `lee::Error` to `anyhow::Error` to `{:#}`. + /// + /// This is required because `lee::Error` is not `Clone + Serialize + Deserialize`, so we + /// cannot store it directly. + reason: String, + }, } #[cfg(test)] @@ -47,10 +64,4 @@ mod tests { } )); } - - #[test] - fn display_is_human_readable() { - let err = BlockIngestError::StateTransition("nonce too low".to_owned()); - assert_eq!(err.to_string(), "state transition failed: nonce too low"); - } } diff --git a/lez/indexer/core/src/lib.rs b/lez/indexer/core/src/lib.rs index d24488ee..7f56ced0 100644 --- a/lez/indexer/core/src/lib.rs +++ b/lez/indexer/core/src/lib.rs @@ -2,7 +2,7 @@ use std::{path::Path, sync::Arc}; use anyhow::{Context as _, Result}; use arc_swap::ArcSwap; -use common::block::Block; +use common::{HashType, block::Block}; // TODO: Remove after testnet use futures::StreamExt as _; pub use ingest_error::BlockIngestError; @@ -29,7 +29,25 @@ enum ChainIdentityOutcome { /// or the check was inconclusive — none of which prove a reset. Consistent, /// The channel serves a different block at one of our ids — a chain reset. - Mismatch { detail: String }, + Mismatch(ChainMismatch), +} + +/// The differing pair behind a [`ChainIdentityOutcome::Mismatch`]: our stored +/// block vs. the block the channel serves at the same id, as `(block_id, hash)`. +struct ChainMismatch { + ours: (u64, HashType), + channel: (u64, HashType), +} + +impl std::fmt::Display for ChainMismatch { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + let Self { ours, channel } = self; + write!( + f, + "stored block {} {} != channel block {} {}", + ours.0, ours.1, channel.0, channel.1 + ) + } } #[derive(Clone)] @@ -51,18 +69,19 @@ impl IndexerCore { let core = Self::open(config.clone(), storage_dir)?; match core.chain_identity_outcome().await? { ChainIdentityOutcome::Consistent => Ok(core), - ChainIdentityOutcome::Mismatch { detail } if config.allow_chain_reset => { + ChainIdentityOutcome::Mismatch(mismatch) if config.allow_chain_reset => { warn!( - "Chain reset detected ({detail}). Wiping indexer store at {} and re-indexing.", + "Chain reset detected ({mismatch}). Wiping indexer store at {} and \ + re-indexing.", home.display() ); drop(core); // sole owner before the ingest task is spawned → closes the DB storage::indexer::RocksDBIO::destroy(&home)?; Self::open(config, storage_dir) } - ChainIdentityOutcome::Mismatch { detail } => Err(anyhow::anyhow!( + ChainIdentityOutcome::Mismatch(mismatch) => Err(anyhow::anyhow!( "Indexer store at {} holds a different chain than the channel now serves \ - ({detail}). Delete the indexer storage directory, point at a fresh one, or \ + ({mismatch}). Delete the indexer storage directory, point at a fresh one, or \ set `allow_chain_reset` in the indexer config.", home.display() )), @@ -121,14 +140,16 @@ impl IndexerCore { Ok(compare_block(&ours, &channel_block)) } - /// Reads the first block the channel serves at/after the tip's slot. `next_messages` - /// is exclusive, so `cursor - 1` includes the tip's own slot. `None` = inconclusive - /// (timeout/error, or bedrock's LIB behind our tip → empty stream). + /// Reads the first block the channel serves at/after the tip's slot. + /// + /// `None` = inconclusive (timeout/error, or bedrock's LIB behind our tip → empty stream). async fn fetch_channel_block_from(&self, cursor: Slot) -> Result> { const TIP_FETCH_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30); - // Slot-0 cursor is degenerate (inscriptions live at wall-clock slots); bail - // rather than let `next_messages(None)` do a from-genesis scan. + + // `next_messages` is exclusive, so `cursor - 1` includes the tip's own slot. let Some(from_slot) = cursor.into_inner().checked_sub(1) else { + // Slot-0 cursor is degenerate (inscriptions live at wall-clock slots); + // so we bail rather than let `next_messages(None)` do a from-genesis scan. return Ok(None); }; let fetch = async { @@ -192,13 +213,20 @@ impl IndexerCore { /// Parks on an inscription that could not be parsed as an L2 block: /// records the stall and flips the status. The validated tip stays frozen. - fn park_undeserializable(&self, slot: Slot, error: &impl std::fmt::Display) { - error!("Failed to deserialize L2 block from zone-sdk: {error}"); - if let Err(err) = self.store.record_deserialize_stall(slot, error.to_string()) { + fn park_undeserializable(&self, slot: Slot, error: std::io::Error) { + let error = anyhow::Error::new(error); + + // use `:#` to get the entire error chain + let reason = format!("{error:#}"); + error!("Failed to deserialize L2 block from zone-sdk: {reason}"); + if let Err(err) = + self.store + .record_stall(None, slot, BlockIngestError::Deserialize(reason.clone())) + { warn!("Failed to record stall reason: {err:#}"); } self.set_status(IndexerSyncStatus::stalled(format!( - "failed to deserialize L2 block: {error}" + "failed to deserialize L2 block: {reason}" ))); } @@ -250,15 +278,7 @@ impl IndexerCore { let block: Block = match borsh::from_slice(&zone_block.data) { Ok(b) => b, Err(error) => { - error!("Failed to deserialize L2 block from zone-sdk: {error}"); - if let Err(err) = self.store.record_deserialize_stall(slot, error.to_string()) { - warn!("Failed to record stall reason: {err:#}"); - } - self.set_status(IndexerSyncStatus::stalled(format!( - "failed to deserialize L2 block: {error}" - ))); - - + self.park_undeserializable(slot, error); // L1 proceeds regardless self.advance_cursor(&mut cursor, slot); continue; @@ -325,15 +345,10 @@ fn compare_block(ours: &Block, channel: &Block) -> ChainIdentityOutcome { if ours.header.hash == channel.header.hash { ChainIdentityOutcome::Consistent } else { - ChainIdentityOutcome::Mismatch { - detail: format!( - "stored block {} {} != channel block {} {}", - ours.header.block_id, - ours.header.hash, - channel.header.block_id, - channel.header.hash - ), - } + ChainIdentityOutcome::Mismatch(ChainMismatch { + ours: (ours.header.block_id, ours.header.hash), + channel: (channel.header.block_id, channel.header.hash), + }) } } @@ -362,7 +377,7 @@ mod chain_identity_tests { let current = block_with_prev(2); assert!(matches!( compare_block(&stored, ¤t), - ChainIdentityOutcome::Mismatch { .. } + ChainIdentityOutcome::Mismatch(_) )); } } diff --git a/lez/indexer/core/src/status.rs b/lez/indexer/core/src/status.rs index 939abec5..e229d4d8 100644 --- a/lez/indexer/core/src/status.rs +++ b/lez/indexer/core/src/status.rs @@ -129,7 +129,10 @@ mod tests { block_hash: None, prev_block_hash: None, l1_slot: Slot::from(0), - error: BlockIngestError::StateTransition("boom".to_owned()), + error: BlockIngestError::StateTransition { + tx_index: 0, + reason: Default::default(), + }, first_seen: None, orphans_since: 2, }), diff --git a/lez/indexer/service/protocol/src/convert.rs b/lez/indexer/service/protocol/src/convert.rs index f497574d..55c4dc6c 100644 --- a/lez/indexer/service/protocol/src/convert.rs +++ b/lez/indexer/service/protocol/src/convert.rs @@ -745,7 +745,16 @@ impl From for BlockIngestError { header: header.into(), } } - indexer_core::BlockIngestError::StateTransition(msg) => Self::StateTransition(msg), + indexer_core::BlockIngestError::EmptyBlock => Self::EmptyBlock, + indexer_core::BlockIngestError::InvalidClockTransaction => { + Self::InvalidClockTransaction + } + indexer_core::BlockIngestError::NonPublicGenesisTransaction => { + Self::NonPublicGenesisTransaction + } + indexer_core::BlockIngestError::StateTransition { tx_index, reason } => { + Self::StateTransition { tx_index, reason } + } } } } diff --git a/lez/indexer/service/protocol/src/lib.rs b/lez/indexer/service/protocol/src/lib.rs index 6ea06a81..e17d539b 100644 --- a/lez/indexer/service/protocol/src/lib.rs +++ b/lez/indexer/service/protocol/src/lib.rs @@ -397,7 +397,14 @@ pub enum BlockIngestError { computed: HashType, header: HashType, }, - StateTransition(String), + EmptyBlock, + InvalidClockTransaction, + NonPublicGenesisTransaction, + StateTransition { + /// Index of the failing transaction within the block body. + tx_index: u64, + reason: String, + }, } /// Diagnostic record of the first block that broke the L2 chain.