refactor!(indexer): migr to chain_state

Cargo.lock

@@ -3864,6 +3864,7 @@ dependencies = [
  "arc-swap",
  "async-stream",
  "borsh",
+ "chain_state",
  "common",
  "futures",
  "humantime-serde",
@@ -3877,7 +3878,6 @@ dependencies = [
  "storage",
  "tempfile",
  "testnet_initial_state",
- "thiserror 2.0.18",
  "tokio",
  "url",
 ]

lez/chain_state/src/apply.rs

@@ -27,6 +27,10 @@ pub enum AcceptOutcome {
     AlreadyApplied,
     /// Did not chain or failed to apply; the tip stays frozen.
     Parked(BlockIngestError),
+    /// Chained but failed to apply, possibly transiently
+    /// ([`BlockIngestError::is_retryable`]); nothing recorded, tip and state
+    /// untouched. The caller retries and parks once it gives up.
+    RetryableFailure(BlockIngestError),
 }
 
 /// Validates `block` against `tip`, then applies it to `state`.

lez/chain_state/src/ingest_error.rs

@@ -40,6 +40,19 @@ pub enum BlockIngestError {
     },
 }
 
+impl BlockIngestError {
+    /// Whether the failure may be transient rather than a property of the block.
+    ///
+    /// FIXME: `StateTransition` is too coarse — its `reason` string mixes genuine
+    /// state-transition rejections with infra failures (risc0 executor teardown,
+    /// storage errors). Once it carries a structured cause, narrow this so only
+    /// infra failures retry.
+    #[must_use]
+    pub const fn is_retryable(&self) -> bool {
+        matches!(self, Self::StateTransition { .. })
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

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
 

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,28 +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),
-    /// Chained but failed to apply, possibly transiently
-    /// ([`BlockIngestError::is_retryable`]); nothing recorded, tip and state
-    /// untouched. The caller retries and parks via
-    /// [`IndexerStore::record_stall`] once it gives up.
-    RetryableFailure(BlockIngestError),
-}
-
 #[derive(Clone)]
 pub struct IndexerStore {
     dbio: Arc<RocksDBIO>,
@@ -252,14 +230,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) {
             if err.is_retryable() {
                 return Ok(AcceptOutcome::RetryableFailure(err));
             }
@@ -284,112 +257,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() {
@@ -538,7 +410,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")

lez/indexer/core/src/chain_consistency.rs

@@ -305,8 +305,7 @@ mod tests {
 
     use super::*;
     use crate::{
-        BlockIngestError,
+        AcceptOutcome, BlockIngestError,
-        block_store::AcceptOutcome,
         config::{ChannelId, ClientConfig, IndexerConfig},
     };
 

lez/indexer/core/src/ingest_error.rs

@@ -1,80 +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,
-    },
-}
-
-impl BlockIngestError {
-    /// Whether the failure may be transient rather than a property of the block.
-    ///
-    /// FIXME: `StateTransition` is too coarse — its `reason` string mixes genuine
-    /// state-transition rejections with infra failures (risc0 executor teardown,
-    /// storage errors). Once it carries a structured cause, narrow this so only
-    /// infra failures retry.
-    #[must_use]
-    pub const fn is_retryable(&self) -> bool {
-        matches!(self, Self::StateTransition { .. })
-    }
-}
-
-#[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
-            }
-        ));
-    }
-}

lez/indexer/core/src/lib.rs

@@ -2,19 +2,18 @@ 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,
 };
 use retry::ApplyRetryGate;
-pub use stall_reason::StallReason;
 
 use crate::{
-    block_store::{AcceptOutcome, IndexerStore},
+    block_store::IndexerStore,
     chain_consistency::ChainConsistency,
     config::IndexerConfig,
     status::{IndexerStatus, IndexerSyncStatus},
@@ -22,9 +21,7 @@ use crate::{
 pub mod block_store;
 pub mod chain_consistency;
 pub mod config;
-pub mod ingest_error;
 mod retry;
-pub mod stall_reason;
 pub mod status;
 
 /// Consecutive failed apply attempts of the same block before parking.

lez/indexer/core/src/stall_reason.rs

@@ -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<u64>,
-    pub block_hash: Option<HashType>,
-    pub prev_block_hash: Option<HashType>,
-    pub l1_slot: Slot,
-    pub error: BlockIngestError,
-    pub first_seen: Option<u64>,
-    /// 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,
-}

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