logos-blockchain/logos-execution-zone
1
0
Fork 0
mirror of https://github.com/logos-blockchain/logos-execution-zone.git synced 2026-07-09 15:29:34 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(indexer): use better error types, tend to few @Arjentix PR review

Browse Source
This commit is contained in:
erhant 2026-07-03 22:36:42 +03:00
parent 9f2fa83945
commit 378d3ab9c2
7 changed files with 136 additions and 80 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
lez/configs/docker-all-in-one/indexer_config.json
View File

@ -3,5 +3,6 @@
"bedrock_config": {
"addr": "http://logos-blockchain-node-0:18080"
},
"channel_id": "0101010101010101010101010101010101010101010101010101010101010101"
"channel_id": "0101010101010101010101010101010101010101010101010101010101010101",
"allow_chain_reset": true
}

66
lez/indexer/core/src/block_store.rs
View File

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

39
lez/indexer/core/src/ingest_error.rs
View File

@ -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");
}
}

83
lez/indexer/core/src/lib.rs
View File

@ -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<Option<Block>> {
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, &current),
ChainIdentityOutcome::Mismatch { .. }
ChainIdentityOutcome::Mismatch(_)
));
}
}

5
lez/indexer/core/src/status.rs
View File

@ -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,
}),

11
lez/indexer/service/protocol/src/convert.rs
View File

@ -745,7 +745,16 @@ impl From<indexer_core::BlockIngestError> 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 }
}
}
}
}

9
lez/indexer/service/protocol/src/lib.rs
View File

@ -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.