diff --git a/Cargo.lock b/Cargo.lock index 954d98fb..64f12ceb 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3866,6 +3866,7 @@ dependencies = [ "async-stream", "authenticated_transfer_core", "borsh", + "chain_state", "common", "futures", "humantime-serde", @@ -3879,7 +3880,6 @@ 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 b3877639..1518b540 100644 --- a/lez/indexer/core/Cargo.toml +++ b/lez/indexer/core/Cargo.toml @@ -12,6 +12,7 @@ default = [] testnet = [] [dependencies] +chain_state.workspace = true common.workspace = true logos-blockchain-zone-sdk.workspace = true lee.workspace = true @@ -29,7 +30,6 @@ futures.workspace = true url.workspace = true logos-blockchain-core.workspace = true serde_json.workspace = true -thiserror.workspace = true async-stream.workspace = true tokio.workspace = true diff --git a/lez/indexer/core/src/block_store.rs b/lez/indexer/core/src/block_store.rs index 2a2bfbce..4bfb2644 100644 --- a/lez/indexer/core/src/block_store.rs +++ b/lez/indexer/core/src/block_store.rs @@ -1,12 +1,12 @@ use std::{path::Path, sync::Arc}; use anyhow::{Context as _, Result}; +use chain_state::{AcceptOutcome, BlockIngestError, StallReason, Tip, apply_block}; use common::{ - HashType, block::{BedrockStatus, Block, BlockHeader}, - transaction::{LeeTransaction, clock_invocation}, + transaction::LeeTransaction, }; -use lee::{Account, AccountId, GENESIS_BLOCK_ID, V03State}; +use lee::{Account, AccountId, V03State}; use lee_core::BlockId; use log::warn; use logos_blockchain_core::header::HeaderId; @@ -14,23 +14,6 @@ use logos_blockchain_zone_sdk::Slot; use storage::indexer::RocksDBIO; use tokio::sync::RwLock; -use crate::{ingest_error::BlockIngestError, stall_reason::StallReason}; - -struct Tip { - block_id: u64, - hash: HashType, -} - -/// Outcome of feeding a parsed L2 block to the validated tip. -pub enum AcceptOutcome { - /// Chained and applied; tip and L1 read cursor both advance. - Applied, - /// A duplicate re-delivery of the current tip. Just L2 advances. - AlreadyApplied, - /// Did not chain or failed to apply; tip stays frozen, stall recorded. - Parked(BlockIngestError), -} - #[derive(Clone)] pub struct IndexerStore { dbio: Arc, @@ -246,14 +229,9 @@ impl IndexerStore { return Ok(AcceptOutcome::AlreadyApplied); } - if let Err(err) = validate_against_tip(tip.as_ref(), block) { - self.record_stall(Some(&block.header), l1_slot, err.clone())?; - return Ok(AcceptOutcome::Parked(err)); - } - // TODO: we use scratch state to be atomic, but need to revisit how expensive a clone is let mut scratch = self.current_state.read().await.clone(); - if let Err(err) = apply_block_to_scratch(block, &mut scratch) { + if let Err(err) = apply_block(tip.as_ref(), block, &mut scratch) { self.record_stall(Some(&block.header), l1_slot, err.clone())?; return Ok(AcceptOutcome::Parked(err)); } @@ -275,112 +253,11 @@ impl IndexerStore { } } -/// Checks that `block` is the valid continuation of `tip`: hash integrity, -/// then block-id continuity, then `prev_block_hash` linkage. A `None` tip -/// (cold store) expects the genesis block. -fn validate_against_tip(tip: Option<&Tip>, block: &Block) -> Result<(), BlockIngestError> { - let computed = block.recompute_hash(); - if computed != block.header.hash { - return Err(BlockIngestError::HashMismatch { - computed, - header: block.header.hash, - }); - } - - match tip { - None => { - if block.header.block_id != GENESIS_BLOCK_ID { - return Err(BlockIngestError::UnexpectedBlockId { - expected: GENESIS_BLOCK_ID, - got: block.header.block_id, - }); - } - } - Some(tip) => { - let expected = tip - .block_id - .checked_add(1) - .expect("block id should not overflow"); - if block.header.block_id != expected { - return Err(BlockIngestError::UnexpectedBlockId { - expected, - got: block.header.block_id, - }); - } - if block.header.prev_block_hash != tip.hash { - return Err(BlockIngestError::BrokenChainLink { - expected_prev: tip.hash, - got_prev: block.header.prev_block_hash, - }); - } - } - } - Ok(()) -} - -/// Applies a block's transactions to `state`, mapping every failure to a -/// [`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(BlockIngestError::EmptyBlock)?; - - let expected_clock = LeeTransaction::Public(clock_invocation(block.header.timestamp)); - if *clock_tx != expected_clock { - return Err(BlockIngestError::InvalidClockTransaction); - } - - let is_genesis = block.header.block_id == GENESIS_BLOCK_ID; - 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::NonPublicGenesisTransaction); - }; - state - .transition_from_public_transaction( - public_tx, - block.header.block_id, - block.header.timestamp, - ) - .map_err(|err| state_transition(err.into()))?; - } else { - transaction - .clone() - .execute_on_state(state, block.header.block_id, block.header.timestamp) - .map_err(|err| state_transition(err.into()))?; - } - } - - let LeeTransaction::Public(clock_public_tx) = clock_tx else { - return Err(BlockIngestError::InvalidClockTransaction); - }; - state - .transition_from_public_transaction( - clock_public_tx, - block.header.block_id, - block.header.timestamp, - ) - .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(()) -} - #[cfg(test)] mod stall_reason_tests { use common::HashType; use super::*; - use crate::{ingest_error::BlockIngestError, stall_reason::StallReason}; #[tokio::test] async fn stall_reason_roundtrips_and_clears() { @@ -529,7 +406,6 @@ mod accept_tests { use common::{HashType, block::HashableBlockData, test_utils::produce_dummy_block}; use super::*; - use crate::ingest_error::BlockIngestError; fn signing_key() -> lee::PrivateKey { lee::PrivateKey::try_new([7_u8; 32]).expect("valid key") diff --git a/lez/indexer/core/src/chain_consistency.rs b/lez/indexer/core/src/chain_consistency.rs index 00f28822..30c3d552 100644 --- a/lez/indexer/core/src/chain_consistency.rs +++ b/lez/indexer/core/src/chain_consistency.rs @@ -305,8 +305,7 @@ mod tests { use super::*; use crate::{ - BlockIngestError, - block_store::AcceptOutcome, + AcceptOutcome, BlockIngestError, config::{ChannelId, ClientConfig, IndexerConfig}, }; diff --git a/lez/indexer/core/src/ingest_error.rs b/lez/indexer/core/src/ingest_error.rs deleted file mode 100644 index bda53497..00000000 --- a/lez/indexer/core/src/ingest_error.rs +++ /dev/null @@ -1,67 +0,0 @@ -use common::HashType; -use serde::{Deserialize, Serialize}; - -/// 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}")] - /// 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}")] - UnexpectedBlockId { expected: u64, got: u64 }, - #[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}")] - HashMismatch { - computed: HashType, - header: HashType, - }, - #[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)] -mod tests { - use super::*; - - #[test] - fn serializes_and_round_trips_externally_tagged() { - let err = BlockIngestError::UnexpectedBlockId { - expected: 5, - got: 7, - }; - let value = serde_json::to_value(&err).expect("serialize"); - assert_eq!( - value, - serde_json::json!({ "UnexpectedBlockId": { "expected": 5, "got": 7 } }) - ); - let back: BlockIngestError = serde_json::from_value(value).expect("deserialize"); - assert!(matches!( - back, - BlockIngestError::UnexpectedBlockId { - expected: 5, - got: 7 - } - )); - } -} diff --git a/lez/indexer/core/src/lib.rs b/lez/indexer/core/src/lib.rs index 77839296..600630c6 100644 --- a/lez/indexer/core/src/lib.rs +++ b/lez/indexer/core/src/lib.rs @@ -2,18 +2,17 @@ use std::{path::Path, sync::Arc}; use anyhow::Result; use arc_swap::ArcSwap; +pub use chain_state::{AcceptOutcome, BlockIngestError, StallReason}; use common::block::Block; // TODO: Remove after testnet use futures::StreamExt as _; -pub use ingest_error::BlockIngestError; use log::{error, info, warn}; use logos_blockchain_zone_sdk::{ CommonHttpClient, Slot, ZoneMessage, adapter::NodeHttpClient, indexer::ZoneIndexer, }; -pub use stall_reason::StallReason; use crate::{ - block_store::{AcceptOutcome, IndexerStore}, + block_store::IndexerStore, chain_consistency::ChainConsistency, config::IndexerConfig, status::{IndexerStatus, IndexerSyncStatus}, @@ -21,8 +20,6 @@ use crate::{ pub mod block_store; pub mod chain_consistency; pub mod config; -pub mod ingest_error; -pub mod stall_reason; pub mod status; #[derive(Clone)] diff --git a/lez/indexer/core/src/stall_reason.rs b/lez/indexer/core/src/stall_reason.rs deleted file mode 100644 index c5118fcc..00000000 --- a/lez/indexer/core/src/stall_reason.rs +++ /dev/null @@ -1,25 +0,0 @@ -use common::HashType; -use logos_blockchain_zone_sdk::Slot; -use serde::{Deserialize, Serialize}; - -use crate::ingest_error::BlockIngestError; - -/// Diagnostic record of the first block that broke the L2 chain. -/// -/// The block-derived fields are `None` for a deserialize break (no header was -/// ever parsed). `l1_slot` is the L1 slot the breaking inscription was read at. -/// `first_seen` is the breaking block's L2 timestamp (`None` for a deserialize break). -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct StallReason { - pub block_id: Option, - pub block_hash: Option, - pub prev_block_hash: Option, - pub l1_slot: Slot, - pub error: BlockIngestError, - pub first_seen: Option, - /// Number of later non-chaining blocks (orphans, since the tip is frozen). - /// - /// TODO: We could store a different "branch" of blocks following this break, but for now we - /// just count them. - pub orphans_since: u64, -} diff --git a/lez/indexer/core/src/status.rs b/lez/indexer/core/src/status.rs index a483fde6..aa182c23 100644 --- a/lez/indexer/core/src/status.rs +++ b/lez/indexer/core/src/status.rs @@ -1,7 +1,6 @@ +use chain_state::StallReason; use serde::Serialize; -use crate::stall_reason::StallReason; - /// Coarse lifecycle state of the indexer's ingestion loop, so a client can tell /// "still catching up" apart from "something went wrong". #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] @@ -117,10 +116,9 @@ mod tests { #[test] fn stalled_status_serializes_with_stall_reason() { + use chain_state::{BlockIngestError, StallReason}; use logos_blockchain_zone_sdk::Slot; - use crate::{ingest_error::BlockIngestError, stall_reason::StallReason}; - let status = IndexerStatus { sync: IndexerSyncStatus::stalled("broken chain link".to_owned()), indexed_block_id: Some(41),