Merge 0428dae84d2a3280674786f9c94c2e34f84d79ae into 2921438e932becd0ec69acd6eaa92317620106e8

Cargo.lock

@@ -1349,6 +1349,25 @@ dependencies = [
  "rand_core 0.10.1",
 ]
 
+[[package]]
+name = "chain_consistency"
+version = "0.1.0"
+dependencies = [
+ "anyhow",
+ "borsh",
+ "common",
+ "futures",
+ "lee",
+ "lee_core",
+ "log",
+ "logos-blockchain-core",
+ "logos-blockchain-zone-sdk",
+ "serde",
+ "serde_json",
+ "thiserror 2.0.18",
+ "tokio",
+]
+
 [[package]]
 name = "chkstk_stub"
 version = "0.1.0"
@@ -3850,6 +3869,7 @@ dependencies = [
  "arc-swap",
  "async-stream",
  "borsh",
+ "chain_consistency",
  "common",
  "futures",
  "humantime-serde",
@@ -3863,6 +3883,7 @@ dependencies = [
  "storage",
  "tempfile",
  "testnet_initial_state",
+ "thiserror 2.0.18",
  "tokio",
  "url",
 ]
@@ -3910,6 +3931,7 @@ dependencies = [
  "base64",
  "common",
  "hex",
+ "indexer_core",
  "lee",
  "lee_core",
  "schemars 1.2.1",
@@ -8890,6 +8912,7 @@ dependencies = [
  "borsh",
  "bridge_core",
  "bytesize",
+ "chain_consistency",
  "chrono",
  "common",
  "faucet_core",

Cargo.toml

@@ -32,6 +32,7 @@ members = [
   "lez/wallet",
   "lez/wallet-ffi",
   "lez/common",
+  "lez/chain_consistency",
   "lez/programs",
   "lez/programs/amm",
   "lez/programs/associated_token_account",
@@ -64,6 +65,7 @@ members = [
 lee = { path = "lee/state_machine" }
 lee_core = { path = "lee/state_machine/core" }
 common = { path = "lez/common" }
+chain_consistency = { path = "lez/chain_consistency" }
 mempool = { path = "lez/mempool" }
 storage = { path = "lez/storage" }
 key_protocol = { path = "lee/key_protocol" }

integration_tests/tests/indexer_stall.rs

@@ -0,0 +1,54 @@
+#![expect(
+    clippy::tests_outside_test_module,
+    reason = "We don't care about these in tests"
+)]
+
+use std::time::Duration;
+
+use anyhow::{Context as _, Result};
+use indexer_service_protocol::IndexerSyncState;
+use indexer_service_rpc::RpcClient as _;
+use integration_tests::{TestContext, wait_for_indexer_to_catch_up};
+use log::info;
+
+const CAUGHT_UP_STATUS_TIMEOUT: Duration = Duration::from_secs(60);
+
+/// Test that the indexer status RPC reports caught-up with no stall after a clean run.
+///
+/// The sequencer keeps producing blocks while we assert, so the status is polled until a
+/// `CaughtUp` snapshot is observed and the indexed tip is checked as a lower bound.
+///
+/// TODO: Integration-level park testing (publishing a bad block to force a stall) is a follow-up
+/// needing fault injection support in the test harness.
+#[tokio::test]
+async fn indexer_status_rpc_reports_caught_up_with_no_stall() -> Result<()> {
+    let ctx = TestContext::new().await?;
+
+    let indexer_tip = wait_for_indexer_to_catch_up(&ctx).await?;
+
+    let status = tokio::time::timeout(CAUGHT_UP_STATUS_TIMEOUT, async {
+        loop {
+            let status = ctx.indexer_client().get_status().await?;
+            if status.state == IndexerSyncState::CaughtUp {
+                return anyhow::Ok(status);
+            }
+            info!("Waiting for caught-up indexer status, got {status:?}");
+            tokio::time::sleep(Duration::from_millis(500)).await;
+        }
+    })
+    .await
+    .context("Timed out waiting for indexer status to report caught-up")??;
+
+    assert!(
+        status.stall_reason.is_none(),
+        "indexer should have no stall reason after a clean run, got {status:?}"
+    );
+    // test for >= here because the sequencer keeps producing blocks while we assert,
+    // so the indexed tip may be ahead of the tip we observed when we waited for caught-up.
+    assert!(
+        status.indexed_block_id >= Some(indexer_tip),
+        "status indexed_block_id should be at least the caught-up tip {indexer_tip}, got {status:?}"
+    );
+
+    Ok(())
+}

lez/chain_consistency/Cargo.toml

@@ -0,0 +1,26 @@
+[package]
+name = "chain_consistency"
+version = "0.1.0"
+edition = "2024"
+license.workspace = true
+
+[lints]
+workspace = true
+
+[dependencies]
+common.workspace = true
+lee.workspace = true
+lee_core.workspace = true
+
+logos-blockchain-zone-sdk.workspace = true
+logos-blockchain-core.workspace = true
+anyhow.workspace = true
+borsh.workspace = true
+futures.workspace = true
+log.workspace = true
+serde.workspace = true
+thiserror.workspace = true
+tokio.workspace = true
+
+[dev-dependencies]
+serde_json.workspace = true

lez/chain_consistency/src/apply.rs

@@ -0,0 +1,182 @@
+//! Validating and applying channel L2 blocks to a `V03State`.
+//!
+//! These primitives are shared by the consumers that reconstruct L2 state from
+//! the Bedrock channel.
+
+use common::{
+    HashType,
+    block::Block,
+    transaction::{LeeTransaction, clock_invocation},
+};
+use lee::{GENESIS_BLOCK_ID, V03State};
+use lee_core::BlockId;
+use serde::{Deserialize, Serialize};
+
+/// Why a channel L2 block could not be validated or applied.
+///
+/// Persisted in `RocksDB` (via the Indexer's stall reason), so every variant
+/// must be `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: BlockId, got: BlockId },
+    #[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,
+    },
+}
+
+/// The last successfully applied block a candidate must extend.
+pub struct Tip {
+    pub block_id: BlockId,
+    pub hash: HashType,
+}
+
+/// 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.
+pub 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`.
+pub fn apply_block(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 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/chain_consistency/src/consistency.rs

@@ -0,0 +1,429 @@
+//! Startup check that a local store still belongs to the chain the connected
+//! channel serves.
+
+use anyhow::Result;
+use common::{HashType, block::Block};
+use futures::StreamExt as _;
+use lee_core::BlockId;
+use log::warn;
+use logos_blockchain_core::mantle::ops::channel::ChannelId;
+use logos_blockchain_zone_sdk::{Slot, ZoneMessage, adapter, indexer::ZoneIndexer};
+
+/// Upper bound on the channel reads of the startup consistency check.
+const CHANNEL_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(60);
+
+/// Result of comparing a caller's stored chain against the channel.
+pub enum ChainConsistency {
+    /// Channel still serves our anchor block (the stored tip position, or the
+    /// parked block while stalled).
+    Consistent,
+    /// We could not determine the outcome due to one of:
+    ///
+    /// - cold store (no anchor to compare at)
+    /// - the channel served only blocks newer than the anchor
+    /// - or the channel read was inconclusive (timeout / error / empty stream)
+    ///
+    /// NOTE: None of these prove a reset, so the caller proceeds.
+    /// A genuine divergence is still caught later when applying the channel history.
+    Inconclusive,
+    /// Positive evidence that the channel is a different chain than the store.
+    ///
+    /// Details in [`ChainMismatch`], and impl's Display trait.
+    Inconsistent(ChainMismatch),
+}
+
+/// The evidence behind a [`ChainConsistency::Inconsistent`].
+pub enum ChainMismatch {
+    /// The channel serves a different block at the anchor's id.
+    Block {
+        ours: (BlockId, HashType),
+        channel: (BlockId, HashType),
+    },
+    /// The channel serves a block at/below the anchor's id past the anchor
+    /// slot; on the same chain those ids live at earlier slots.
+    ReinscribedBlock {
+        channel: (BlockId, HashType),
+        slot: Slot,
+        anchor_slot: Slot,
+    },
+    /// The channel has content past the anchor slot but no longer the
+    /// inscription we anchored on.
+    AnchorSlotChanged { anchor_slot: Slot },
+    /// The channel does not exist on the connected chain.
+    ChannelMissing,
+    /// The channel's history ends before the anchor slot.
+    ChannelBehindAnchor { tip_slot: Slot, anchor_slot: Slot },
+}
+
+impl std::fmt::Display for ChainMismatch {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            Self::Block { ours, channel } => write!(
+                f,
+                "stored block {} {} != channel block {} {}",
+                ours.0, ours.1, channel.0, channel.1
+            ),
+            Self::ReinscribedBlock {
+                channel,
+                slot,
+                anchor_slot,
+            } => write!(
+                f,
+                "channel re-serves block {} {} at slot {} past our anchor slot {}",
+                channel.0,
+                channel.1,
+                slot.into_inner(),
+                anchor_slot.into_inner()
+            ),
+            Self::AnchorSlotChanged { anchor_slot } => write!(
+                f,
+                "channel content at slot {} no longer includes the inscription we parked on",
+                anchor_slot.into_inner()
+            ),
+            Self::ChannelMissing => write!(f, "channel does not exist on the connected chain"),
+            Self::ChannelBehindAnchor {
+                tip_slot,
+                anchor_slot,
+            } => write!(
+                f,
+                "channel tip slot {} is behind our anchor slot {}",
+                tip_slot.into_inner(),
+                anchor_slot.into_inner()
+            ),
+        }
+    }
+}
+
+/// A block that must still be inscribed at `slot` if the channel is the chain
+/// the store was built from: the tip at the read cursor, or the recorded
+/// parked block while stalled.
+pub struct Anchor {
+    slot: Slot,
+    /// The anchor block's `(id, hash)`.
+    ///
+    /// `None` when anchored on an undeserializable inscription (no header was recorded).
+    block: Option<(BlockId, HashType)>,
+}
+
+impl Anchor {
+    /// Builds an anchor at `slot` on the block `(id, hash)`, or a headerless
+    /// anchor (`None`) when only the slot is known.
+    #[must_use]
+    pub const fn new(slot: Slot, block: Option<(BlockId, HashType)>) -> Self {
+        Self { slot, block }
+    }
+
+    /// Probes a channel message read at/after the anchor slot.
+    /// See [`verify_chain_consistency`].
+    fn probe_anchor_slot(&self, msg: &ZoneMessage, slot: Slot) -> AnchorProbe {
+        if slot < self.slot {
+            return AnchorProbe::KeepLooking;
+        }
+        let Some((anchor_id, anchor_hash)) = self.block else {
+            // Anchored on an undeserializable inscription: any message still
+            // present at that slot means the history is intact.
+            return if slot == self.slot {
+                AnchorProbe::SameChain
+            } else {
+                AnchorProbe::Mismatch(ChainMismatch::AnchorSlotChanged {
+                    anchor_slot: self.slot,
+                })
+            };
+        };
+        let ZoneMessage::Block(zone_block) = msg else {
+            return AnchorProbe::KeepLooking;
+        };
+        let Ok(block) = borsh::from_slice::<Block>(&zone_block.data) else {
+            return AnchorProbe::KeepLooking;
+        };
+        let (id, hash) = (block.header.block_id, block.header.hash);
+        if id == anchor_id {
+            return if hash == anchor_hash {
+                AnchorProbe::SameChain
+            } else {
+                AnchorProbe::Mismatch(ChainMismatch::Block {
+                    ours: (anchor_id, anchor_hash),
+                    channel: (id, hash),
+                })
+            };
+        }
+        if id > anchor_id {
+            return AnchorProbe::Bail;
+        }
+        if slot == self.slot {
+            // Older ids can share the anchor's slot on the same chain.
+            return AnchorProbe::KeepLooking;
+        }
+        // An id below the anchor served past the anchor slot is impossible on the
+        // same chain, even if the content is identical (deterministic genesis).
+        AnchorProbe::Mismatch(ChainMismatch::ReinscribedBlock {
+            channel: (id, hash),
+            slot,
+            anchor_slot: self.slot,
+        })
+    }
+}
+
+/// What a single channel message tells the anchored consistency check.
+enum AnchorProbe {
+    /// The anchor is still in place: same chain.
+    SameChain,
+    Mismatch(ChainMismatch),
+    /// Only newer ids past the anchor: plausible on the same chain, so stop
+    /// scanning without a verdict.
+    Bail,
+    KeepLooking,
+}
+
+/// Detects when a local store belongs to a different chain than the connected
+/// L1 (e.g. a wiped/restarted Bedrock) so startup can react instead of silently
+/// diverging.
+///
+/// Verifies the channel still carries the anchor block at its slot. The anchor
+/// was finalized at `anchor.slot`, so the same chain must still serve it there,
+/// while a reset chain re-inscribes its content only at later wall-clock slots.
+/// Only positive evidence of a different chain yields
+/// [`ChainConsistency::Inconsistent`]; absence of data stays
+/// [`ChainConsistency::Inconclusive`].
+///
+/// `node` need only implement the zone-sdk [`adapter::Node`] trait; a throwaway
+/// [`ZoneIndexer`] is built internally for the channel read.
+pub async fn verify_chain_consistency<N>(
+    node: &N,
+    channel_id: ChannelId,
+    anchor: &Anchor,
+) -> Result<ChainConsistency>
+where
+    N: adapter::Node + Clone + Sync,
+{
+    match node.channel_state(channel_id).await {
+        Ok(state) => {
+            if let Some(mismatch) = frontier_verdict(anchor.slot, state.map(|s| s.tip_slot)) {
+                return Ok(ChainConsistency::Inconsistent(mismatch));
+            }
+        }
+        Err(err) => {
+            warn!("Failed to read channel state for the consistency check: {err:#}");
+        }
+    }
+
+    // `next_messages` is exclusive, so `slot - 1` includes the anchor slot.
+    let Some(from_slot) = anchor.slot.into_inner().checked_sub(1) else {
+        return Ok(ChainConsistency::Inconclusive);
+    };
+
+    let zone_indexer = ZoneIndexer::new(channel_id, node.clone());
+    let scan = async {
+        let stream = zone_indexer
+            .next_messages(Some(Slot::from(from_slot)))
+            .await?;
+        let mut stream = std::pin::pin!(stream);
+
+        while let Some((msg, slot)) = stream.next().await {
+            match anchor.probe_anchor_slot(&msg, slot) {
+                AnchorProbe::SameChain => return Ok(Some(ChainConsistency::Consistent)),
+                AnchorProbe::Mismatch(mismatch) => {
+                    return Ok(Some(ChainConsistency::Inconsistent(mismatch)));
+                }
+                AnchorProbe::Bail => return Ok(Some(ChainConsistency::Inconclusive)),
+                AnchorProbe::KeepLooking => { /* dont do anything */ }
+            }
+        }
+        Ok::<_, anyhow::Error>(None)
+    };
+
+    match tokio::time::timeout(CHANNEL_READ_TIMEOUT, scan).await {
+        Ok(Ok(Some(outcome))) => Ok(outcome),
+        Ok(Ok(None)) => Ok(ChainConsistency::Inconclusive),
+        Ok(Err(err)) => {
+            warn!("Failed to read the anchor slot for the consistency check; proceeding: {err:#}");
+            Ok(ChainConsistency::Inconclusive)
+        }
+        Err(_elapsed) => {
+            warn!("Timed out reading the anchor slot for the consistency check; proceeding");
+            Ok(ChainConsistency::Inconclusive)
+        }
+    }
+}
+
+/// Classifies an already-fetched slice of channel messages against `anchor`.
+///
+/// The message-slice counterpart of [`verify_chain_consistency`], for callers
+/// that read the channel history themselves (e.g. a sequencer that reads once
+/// and both verifies and reconstructs from the same messages). `messages` must
+/// be the finalized channel history from the anchor slot onward, in slot order.
+///
+/// Reuses the same [`Anchor::probe_anchor_slot`] and [`frontier_verdict`] logic
+/// as the streaming path, so both consumers share one definition of a mismatch.
+#[must_use]
+pub fn classify_channel(
+    anchor: &Anchor,
+    channel_tip_slot: Option<Slot>,
+    messages: &[(ZoneMessage, Slot)],
+) -> ChainConsistency {
+    if let Some(mismatch) = frontier_verdict(anchor.slot, channel_tip_slot) {
+        return ChainConsistency::Inconsistent(mismatch);
+    }
+    for (msg, slot) in messages {
+        match anchor.probe_anchor_slot(msg, *slot) {
+            AnchorProbe::SameChain => return ChainConsistency::Consistent,
+            AnchorProbe::Mismatch(mismatch) => return ChainConsistency::Inconsistent(mismatch),
+            AnchorProbe::Bail => return ChainConsistency::Inconclusive,
+            AnchorProbe::KeepLooking => {}
+        }
+    }
+    ChainConsistency::Inconclusive
+}
+
+/// Checks the channel frontier against the anchor slot.
+///
+/// The anchor block was finalized at `anchor_slot`, so on the same chain the
+/// channel tip can never be behind it, and the channel must exist.
+fn frontier_verdict(anchor_slot: Slot, channel_tip_slot: Option<Slot>) -> Option<ChainMismatch> {
+    match channel_tip_slot {
+        None => Some(ChainMismatch::ChannelMissing),
+        Some(tip_slot) if tip_slot < anchor_slot => Some(ChainMismatch::ChannelBehindAnchor {
+            tip_slot,
+            anchor_slot,
+        }),
+        Some(_) => None,
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use common::block::HashableBlockData;
+    use logos_blockchain_core::mantle::ops::channel::{MsgId, inscribe::Inscription};
+    use logos_blockchain_zone_sdk::ZoneBlock;
+
+    use super::*;
+
+    fn test_block(block_id: BlockId, timestamp: u64) -> Block {
+        HashableBlockData {
+            block_id,
+            prev_block_hash: HashType([0; 32]),
+            timestamp,
+            transactions: vec![],
+        }
+        .into_pending_block(&lee::PrivateKey::try_new([7; 32]).expect("valid key"))
+    }
+
+    fn block_msg(block: &Block) -> ZoneMessage {
+        let bytes = borsh::to_vec(block).expect("serialize");
+        ZoneMessage::Block(ZoneBlock {
+            id: MsgId::from([0_u8; 32]),
+            data: Inscription::try_from(bytes.as_slice()).expect("inscription"),
+        })
+    }
+
+    fn anchor_for(block: &Block, slot: Slot) -> Anchor {
+        Anchor::new(slot, Some((block.header.block_id, block.header.hash)))
+    }
+
+    #[test]
+    fn probe_finds_anchor_block_at_slot() {
+        let tip = test_block(5, 42);
+        let anchor = anchor_for(&tip, Slot::from(1_000));
+        assert!(matches!(
+            anchor.probe_anchor_slot(&block_msg(&tip), Slot::from(1_000)),
+            AnchorProbe::SameChain
+        ));
+    }
+
+    #[test]
+    fn probe_flags_different_block_at_anchor_id() {
+        let tip = test_block(5, 42);
+        let anchor = anchor_for(&tip, Slot::from(1_000));
+        // Same id, different content (timestamp) => different hash.
+        let other = test_block(5, 43);
+        assert!(matches!(
+            anchor.probe_anchor_slot(&block_msg(&other), Slot::from(1_000)),
+            AnchorProbe::Mismatch(ChainMismatch::Block { .. })
+        ));
+    }
+
+    #[test]
+    fn probe_flags_old_id_reinscribed_past_the_anchor_slot() {
+        // A reset chain re-inscribes from genesis at later slots. Even if the
+        // content is byte-identical (deterministic genesis), an id at/below
+        // the anchor past the anchor slot is impossible on the same chain.
+        let tip = test_block(5, 42);
+        let anchor = anchor_for(&tip, Slot::from(1_000));
+        let genesis = test_block(1, 0);
+        assert!(matches!(
+            anchor.probe_anchor_slot(&block_msg(&genesis), Slot::from(1_001)),
+            AnchorProbe::Mismatch(ChainMismatch::ReinscribedBlock { .. })
+        ));
+    }
+
+    #[test]
+    fn probe_skips_older_blocks_sharing_the_anchor_slot() {
+        let tip = test_block(5, 42);
+        let anchor = anchor_for(&tip, Slot::from(1_000));
+        let earlier = test_block(4, 41);
+        assert!(matches!(
+            anchor.probe_anchor_slot(&block_msg(&earlier), Slot::from(1_000)),
+            AnchorProbe::KeepLooking
+        ));
+    }
+
+    #[test]
+    fn probe_bails_on_newer_ids_past_the_anchor() {
+        // Blocks newer than the anchor are plausible on the same chain (e.g.
+        // published while we were down), so they must never count as evidence.
+        let tip = test_block(5, 42);
+        let anchor = anchor_for(&tip, Slot::from(1_000));
+        let newer = test_block(6, 43);
+        assert!(matches!(
+            anchor.probe_anchor_slot(&block_msg(&newer), Slot::from(1_001)),
+            AnchorProbe::Bail
+        ));
+    }
+
+    #[test]
+    fn probe_skips_undeserializable_inscriptions() {
+        let tip = test_block(5, 42);
+        let anchor = anchor_for(&tip, Slot::from(1_000));
+        let garbage = ZoneMessage::Block(ZoneBlock {
+            id: MsgId::from([0_u8; 32]),
+            data: Inscription::try_from(&[1_u8, 2, 3][..]).expect("inscription"),
+        });
+        assert!(matches!(
+            anchor.probe_anchor_slot(&garbage, Slot::from(1_000)),
+            AnchorProbe::KeepLooking
+        ));
+    }
+
+    #[test]
+    fn probe_accepts_any_message_for_a_headerless_anchor() {
+        // A deserialize park records no header: any message still present at
+        // the anchor slot means the history is intact.
+        let anchor = Anchor::new(Slot::from(1_000), None);
+        let garbage = ZoneMessage::Block(ZoneBlock {
+            id: MsgId::from([0_u8; 32]),
+            data: Inscription::try_from(&[1_u8, 2, 3][..]).expect("inscription"),
+        });
+        assert!(matches!(
+            anchor.probe_anchor_slot(&garbage, Slot::from(1_000)),
+            AnchorProbe::SameChain
+        ));
+        assert!(matches!(
+            anchor.probe_anchor_slot(&garbage, Slot::from(1_001)),
+            AnchorProbe::Mismatch(ChainMismatch::AnchorSlotChanged { .. })
+        ));
+    }
+
+    #[test]
+    fn frontier_flags_missing_channel_and_short_history() {
+        assert!(matches!(
+            frontier_verdict(Slot::from(1_000), None),
+            Some(ChainMismatch::ChannelMissing)
+        ));
+        assert!(matches!(
+            frontier_verdict(Slot::from(1_000), Some(Slot::from(999))),
+            Some(ChainMismatch::ChannelBehindAnchor { .. })
+        ));
+        assert!(frontier_verdict(Slot::from(1_000), Some(Slot::from(1_000))).is_none());
+        assert!(frontier_verdict(Slot::from(1_000), Some(Slot::from(2_000))).is_none());
+    }
+}

lez/chain_consistency/src/lib.rs

@@ -0,0 +1,9 @@
+//! Reconstructing and verifying L2 chain state from a Bedrock (L1) channel.
+
+pub use apply::{BlockIngestError, Tip, apply_block, validate_against_tip};
+pub use consistency::{
+    Anchor, ChainConsistency, ChainMismatch, classify_channel, verify_chain_consistency,
+};
+
+pub mod apply;
+pub mod consistency;

lez/common/src/block.rs

@@ -55,6 +55,21 @@ pub struct Block {
     pub bedrock_status: BedrockStatus,
 }
 
+impl Block {
+    /// Recomputes the hash from this block's contents, for integrity verification
+    /// against the value stored in `header.hash`.
+    #[must_use]
+    pub fn recompute_hash(&self) -> BlockHash {
+        HashableBlockData {
+            block_id: self.header.block_id,
+            prev_block_hash: self.header.prev_block_hash,
+            timestamp: self.header.timestamp,
+            transactions: self.body.transactions.clone(),
+        }
+        .compute_hash()
+    }
+}
+
 impl Serialize for Block {
     fn serialize<S: serde::Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
         crate::borsh_base64::serialize(self, serializer)
@@ -76,11 +91,13 @@ pub struct HashableBlockData {
 }
 
 impl HashableBlockData {
+    /// Domain-separated hash of the block contents: `SHA256(PREFIX || borsh(self))`.
+    /// The single source of truth for both producing and verifying a block hash.
     #[must_use]
-    pub fn into_pending_block(self, signing_key: &lee::PrivateKey) -> Block {
+    pub fn compute_hash(&self) -> BlockHash {
         const PREFIX: &[u8; 32] = b"/LEE/v0.3/Message/Block/\x00\x00\x00\x00\x00\x00\x00\x00";
 
-        let data_bytes = borsh::to_vec(&self).unwrap();
+        let data_bytes = borsh::to_vec(self).unwrap();
         let mut bytes = Vec::with_capacity(
             PREFIX
                 .len()
@@ -89,8 +106,12 @@ impl HashableBlockData {
         );
         bytes.extend_from_slice(PREFIX);
         bytes.extend_from_slice(&data_bytes);
+        OwnHasher::hash(&bytes)
+    }
 
-        let hash = OwnHasher::hash(&bytes);
+    #[must_use]
+    pub fn into_pending_block(self, signing_key: &lee::PrivateKey) -> Block {
+        let hash = self.compute_hash();
         let signature = lee::Signature::new(signing_key, &hash.0);
         Block {
             header: BlockHeader {
@@ -132,4 +153,33 @@ mod tests {
         let block_from_bytes = borsh::from_slice::<HashableBlockData>(&bytes).unwrap();
         assert_eq!(hashable, block_from_bytes);
     }
+
+    #[test]
+    fn recompute_hash_matches_header_for_well_formed_block() {
+        let key = lee::PrivateKey::try_new([7_u8; 32]).expect("valid key");
+        let block = HashableBlockData {
+            block_id: 5,
+            prev_block_hash: HashType([9_u8; 32]),
+            timestamp: 42,
+            transactions: vec![test_utils::produce_dummy_empty_transaction()],
+        }
+        .into_pending_block(&key);
+        assert_eq!(block.recompute_hash(), block.header.hash);
+    }
+
+    #[test]
+    fn recompute_hash_detects_tampering() {
+        let key = lee::PrivateKey::try_new([7_u8; 32]).expect("valid key");
+        let block = HashableBlockData {
+            block_id: 5,
+            prev_block_hash: HashType([9_u8; 32]),
+            timestamp: 42,
+            transactions: vec![test_utils::produce_dummy_empty_transaction()],
+        }
+        .into_pending_block(&key);
+
+        let mut tampered = block;
+        tampered.header.timestamp = 99; // header changed; stale hash no longer matches
+        assert_ne!(tampered.recompute_hash(), tampered.header.hash);
+    }
 }

lez/common/src/transaction.rs

@@ -92,9 +92,8 @@ impl LeeTransaction {
         Ok(diff)
     }
 
-    /// Computes the validated state diff without enforcing the system-account
-    /// restriction. Shared by [`Self::validate_on_state`] and
-    /// [`Self::execute_without_system_accounts_check_on_state`].
+    /// Computes the validated state diff. Shared by [`Self::validate_on_state`]
+    /// (which adds the system-account guards) and [`Self::execute_on_state`].
     fn compute_state_diff(
         &self,
         state: &V03State,
@@ -129,16 +128,12 @@ impl LeeTransaction {
         Ok(self)
     }
 
-    /// Similar to [`Self::execute_check_on_state`], but skips the system-account guard.
+    /// Executes the transaction against the current state and applies the resulting diff,
+    /// without the system-account guards enforced by [`Self::execute_check_on_state`].
     ///
-    /// FIXME: HOT FIX (testnet v0.2): the indexer replays blocks the sequencer already
-    /// accepted, including sequencer-generated deposit transactions that
-    /// legitimately modify the bridge account. The `TransactionOrigin::Sequencer`
-    /// tag that lets the sequencer bypass the guard is not carried in the block,
-    /// so the indexer cannot yet distinguish deposit txs from user txs.
-    ///
-    /// REMOVE ME when the indexer can authenticate deposit transactions.
-    pub fn execute_without_system_accounts_check_on_state(
+    /// The indexer replays blocks the sequencer already validated and inscribed on Bedrock,
+    /// so it trusts those inscriptions and re-derives state without re-validating them.
+    pub fn execute_on_state(
         self,
         state: &mut V03State,
         block_id: BlockId,

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
 }

lez/indexer/core/Cargo.toml

@@ -13,6 +13,7 @@ testnet = []
 
 [dependencies]
 common.workspace = true
+chain_consistency.workspace = true
 logos-blockchain-zone-sdk.workspace = true
 lee.workspace = true
 lee_core.workspace = true
@@ -29,6 +30,7 @@ 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,18 +1,31 @@
 use std::{path::Path, sync::Arc};
 
 use anyhow::{Context as _, Result};
+use chain_consistency::{BlockIngestError, Tip};
 use common::{
-    block::{BedrockStatus, Block},
-    transaction::{LeeTransaction, clock_invocation},
+    block::{BedrockStatus, Block, BlockHeader},
+    transaction::LeeTransaction,
 };
 use lee::{Account, AccountId, V03State};
 use lee_core::BlockId;
-use log::info;
+use log::warn;
 use logos_blockchain_core::header::HeaderId;
 use logos_blockchain_zone_sdk::Slot;
 use storage::indexer::RocksDBIO;
 use tokio::sync::RwLock;
 
+use crate::stall_reason::StallReason;
+
+/// 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<RocksDBIO>,
@@ -118,6 +131,29 @@ impl IndexerStore {
         Ok(())
     }
 
+    pub fn get_stall_reason(&self) -> Result<Option<StallReason>> {
+        let Some(bytes) = self.dbio.get_stall_reason_bytes()? else {
+            return Ok(None);
+        };
+        let stall: Option<StallReason> =
+            serde_json::from_slice(&bytes).context("Failed to deserialize stored stall reason")?;
+        Ok(stall)
+    }
+
+    pub fn set_stall_reason(&self, stall: &Option<StallReason>) -> Result<()> {
+        let bytes = serde_json::to_vec(stall).context("Failed to serialize stall reason")?;
+        self.dbio.put_stall_reason_bytes(&bytes)?;
+        Ok(())
+    }
+
+    /// Clears a recorded stall marker if one is present, skipping the write otherwise.
+    fn clear_stall_if_present(&self) -> Result<()> {
+        if self.get_stall_reason()?.is_some() {
+            self.set_stall_reason(&None)?;
+        }
+        Ok(())
+    }
+
     /// Recalculation of final state directly from DB.
     ///
     /// Used for indexer healthcheck.
@@ -139,137 +175,147 @@ impl IndexerStore {
             .get_account_by_id(*account_id))
     }
 
-    pub async fn put_block(&self, mut block: Block, l1_header: HeaderId) -> Result<()> {
-        info!("Applying block {}", block.header.block_id);
-        {
-            let mut state_guard = self.current_state.write().await;
+    /// The last successfully applied block, or `None` on a cold store.
+    /// Read fresh from the store each call.
+    fn validated_tip(&self) -> Result<Option<Tip>> {
+        let Some(block_id) = self.dbio.get_meta_last_block_id_in_db()? else {
+            return Ok(None);
+        };
+        let Some(block) = self.dbio.get_block(block_id)? else {
+            return Ok(None);
+        };
+        Ok(Some(Tip {
+            block_id,
+            hash: block.header.hash,
+        }))
+    }
 
-            let (clock_tx, user_txs) = block
-                .body
-                .transactions
-                .split_last()
-                .ok_or_else(|| anyhow::anyhow!("Block has no transactions"))?;
-
-            anyhow::ensure!(
-                *clock_tx == LeeTransaction::Public(clock_invocation(block.header.timestamp)),
-                "Last transaction in block must be the clock invocation for the block timestamp"
-            );
-
-            let is_genesis = block.header.block_id == 1;
-            for transaction in user_txs {
-                if is_genesis {
-                    let genesis_tx = match transaction {
-                        LeeTransaction::Public(public_tx) => public_tx,
-                        LeeTransaction::PrivacyPreserving(_)
-                        | LeeTransaction::ProgramDeployment(_) => {
-                            anyhow::bail!("Genesis block should contain only public transactions")
-                        }
-                    };
-                    state_guard
-                        .transition_from_public_transaction(
-                            genesis_tx,
-                            block.header.block_id,
-                            block.header.timestamp,
-                        )
-                        .context("Failed to execute genesis public transaction")?;
-                } else {
-                    transaction
-                        .clone()
-                        .transaction_stateless_check()?
-                        // FIXME: HOT FIX (testnet v0.2): does not check for system account updates due to
-                        // sequencer-generated deposit tx'es;
-                        // CHANGE ME back to `execute_check_on_state` when the indexer can authenticate deposit transactions
-                        .execute_without_system_accounts_check_on_state(
-                            &mut state_guard,
-                            block.header.block_id,
-                            block.header.timestamp,
-                        )?;
-                }
+    /// 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,
+        error: BlockIngestError,
+    ) -> Result<()> {
+        let stall = match self.get_stall_reason()? {
+            Some(mut existing) => {
+                existing.orphans_since = existing.orphans_since.saturating_add(1);
+                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),
+                first_seen: header.map(|h| h.timestamp),
+                l1_slot,
+                error,
+                orphans_since: 0,
+            },
+        };
+        self.set_stall_reason(&Some(stall))
+    }
 
-            // Apply the clock invocation directly (it is expected to modify clock accounts).
-            let LeeTransaction::Public(clock_public_tx) = clock_tx else {
-                anyhow::bail!("Clock invocation must be a public transaction");
-            };
-            state_guard.transition_from_public_transaction(
-                clock_public_tx,
-                block.header.block_id,
-                block.header.timestamp,
-            )?;
+    /// 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.
+    pub async fn accept_block(&self, block: &Block, l1_slot: Slot) -> Result<AcceptOutcome> {
+        let tip = self.validated_tip()?;
+
+        // Re-delivery of an already-applied block is idempotent, not a divergence
+        if let Some(tip) = &tip
+            && block.header.block_id <= tip.block_id
+            && let Some(stored) = self.get_block_at_id(block.header.block_id)?
+            && stored.header.hash == block.header.hash
+        {
+            return Ok(AcceptOutcome::AlreadyApplied);
         }
 
-        // ToDo: Currently we are fetching only finalized blocks
-        // if it changes, the following lines need to be updated
-        // to represent correct block finality
-        block.bedrock_status = BedrockStatus::Finalized;
+        if let Err(err) = chain_consistency::validate_against_tip(tip.as_ref(), block) {
+            self.record_stall(Some(&block.header), l1_slot, err.clone())?;
+            return Ok(AcceptOutcome::Parked(err));
+        }
 
-        info!("Putting block {} into DB", block.header.block_id);
-        Ok(self.dbio.put_block(&block, l1_header.into())?)
+        // 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) = chain_consistency::apply_block(block, &mut scratch) {
+            self.record_stall(Some(&block.header), l1_slot, err.clone())?;
+            return Ok(AcceptOutcome::Parked(err));
+        }
+
+        let mut stored = block.clone();
+        stored.bedrock_status = BedrockStatus::Finalized;
+        self.dbio
+            .put_block(&stored, [0_u8; 32])
+            .context("Failed to persist accepted block")?;
+
+        // Commit in-memory state (infallible) only after the DB write succeeded.
+        *self.current_state.write().await = scratch;
+        // Best-effort: the block is durably applied, so a failed stall clear must not
+        // fail the apply. It self-heals on the next clear.
+        if let Err(err) = self.clear_stall_if_present() {
+            warn!("Failed to clear stall marker after applying block: {err:#}");
+        }
+        Ok(AcceptOutcome::Applied)
+    }
+}
+
+#[cfg(test)]
+mod stall_reason_tests {
+    use common::HashType;
+
+    use super::*;
+    use crate::stall_reason::StallReason;
+
+    #[tokio::test]
+    async fn stall_reason_roundtrips_and_clears() {
+        let dir = tempfile::tempdir().expect("tempdir");
+        let store = IndexerStore::open_db(dir.path()).expect("open store");
+
+        assert!(store.get_stall_reason().expect("get").is_none());
+
+        let stall = StallReason {
+            block_id: Some(7),
+            block_hash: Some(HashType([1_u8; 32])),
+            prev_block_hash: Some(HashType([2_u8; 32])),
+            l1_slot: Slot::from(42),
+            error: BlockIngestError::StateTransition {
+                tx_index: 0,
+                reason: "boom".to_owned(),
+            },
+            first_seen: Some(99),
+            orphans_since: 3,
+        };
+        store.set_stall_reason(&Some(stall)).expect("set stall");
+
+        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_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));
+        assert_eq!(got.first_seen, Some(99));
+
+        store.set_stall_reason(&None).expect("clear");
+        assert!(store.get_stall_reason().expect("get").is_none());
     }
 }
 
 #[cfg(test)]
 mod tests {
-    use common::{HashType, block::HashableBlockData};
+    use common::test_utils::{create_transaction_native_token_transfer, produce_dummy_block};
     use tempfile::tempdir;
     use testnet_initial_state::initial_pub_accounts_private_keys;
 
     use super::*;
 
-    struct TestFixture {
-        storage: IndexerStore,
-        from: AccountId,
-        to: AccountId,
-        _home: tempfile::TempDir,
-    }
-
-    #[expect(
-        clippy::arithmetic_side_effects,
-        reason = "test helper with bounded inputs"
-    )]
-    async fn store_with_transfer_blocks(
-        block_count: u64,
-        prev_hash: Option<common::HashType>,
-    ) -> TestFixture {
-        let home = tempdir().unwrap();
-        let storage = IndexerStore::open_db(home.path()).unwrap();
-
-        let initial_accounts = initial_pub_accounts_private_keys();
-        let from = initial_accounts[0].account_id;
-        let to = initial_accounts[1].account_id;
-        let sign_key = initial_accounts[0].pub_sign_key.clone();
-
-        let mut prev_hash = prev_hash;
-        for i in 0..block_count {
-            let tx = common::test_utils::create_transaction_native_token_transfer(
-                from,
-                u128::from(i),
-                to,
-                10,
-                &sign_key,
-            );
-            let block_id = i + 1;
-
-            let next_block = common::test_utils::produce_dummy_block(block_id, prev_hash, vec![tx]);
-            prev_hash = Some(next_block.header.hash);
-
-            storage
-                .put_block(
-                    next_block,
-                    HeaderId::from([u8::try_from(i + 1).unwrap(); 32]),
-                )
-                .await
-                .unwrap();
-        }
-
-        TestFixture {
-            storage,
-            from,
-            to,
-            _home: home,
-        }
-    }
-
     #[test]
     fn correct_startup() {
         let home = tempdir().unwrap();
@@ -282,75 +328,348 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn state_transition() {
+    async fn accept_block_applies_transfers_and_advances_tip() {
         let home = tempdir().unwrap();
-        let storage = IndexerStore::open_db(home.as_ref()).unwrap();
+        let store = IndexerStore::open_db(home.as_ref()).unwrap();
 
         let initial_accounts = initial_pub_accounts_private_keys();
         let from = initial_accounts[0].account_id;
         let to = initial_accounts[1].account_id;
         let sign_key = initial_accounts[0].pub_sign_key.clone();
 
-        let clock_tx = LeeTransaction::Public(clock_invocation(0));
-        let genesis_block_data = HashableBlockData {
-            block_id: 1,
-            prev_block_hash: HashType::default(),
-            timestamp: 0,
-            transactions: vec![clock_tx],
-        };
-        let genesis_block = genesis_block_data
-            .into_pending_block(&common::test_utils::sequencer_sign_key_for_testing());
-        let mut prev_hash = Some(genesis_block.header.hash);
-        storage
-            .put_block(genesis_block, HeaderId::from([0_u8; 32]))
-            .await
-            .unwrap();
+        // Genesis (block 1): clock-only.
+        let genesis = produce_dummy_block(1, None, vec![]);
+        let mut prev_hash = genesis.header.hash;
+        assert!(matches!(
+            store.accept_block(&genesis, Slot::from(0)).await.unwrap(),
+            AcceptOutcome::Applied
+        ));
 
-        for i in 0..10_u128 {
-            let tx = common::test_utils::create_transaction_native_token_transfer(
-                from, i, to, 10, &sign_key,
-            );
-            let block_id = u64::try_from(i + 1).unwrap();
-            let next_block = common::test_utils::produce_dummy_block(block_id, prev_hash, vec![tx]);
-            prev_hash = Some(next_block.header.hash);
-            storage
-                .put_block(
-                    next_block,
-                    HeaderId::from([u8::try_from(i + 1).unwrap(); 32]),
-                )
-                .await
-                .unwrap();
+        // Blocks 2..=11: one native transfer of 10 each (nonces 0..=9).
+        for i in 0..10_u64 {
+            let tx = create_transaction_native_token_transfer(from, i.into(), to, 10, &sign_key);
+            let block = produce_dummy_block(i + 2, Some(prev_hash), vec![tx]);
+            prev_hash = block.header.hash;
+            assert!(matches!(
+                store.accept_block(&block, Slot::from(0)).await.unwrap(),
+                AcceptOutcome::Applied
+            ));
         }
 
-        let acc1_val = storage.account_current_state(&from).await.unwrap();
-        let acc2_val = storage.account_current_state(&to).await.unwrap();
-
-        assert_eq!(acc1_val.balance, 9900);
-        assert_eq!(acc2_val.balance, 20100);
+        assert_eq!(
+            store.account_current_state(&from).await.unwrap().balance,
+            9900
+        );
+        assert_eq!(
+            store.account_current_state(&to).await.unwrap().balance,
+            20100
+        );
+        // Tip advanced to the last applied block; a clean run leaves no stall.
+        assert_eq!(store.get_last_block_id().unwrap(), Some(11));
+        assert!(store.get_stall_reason().unwrap().is_none());
     }
 
     #[tokio::test]
-    async fn account_state_at_block() {
-        let TestFixture {
-            storage,
-            from,
-            to,
-            _home,
-        } = store_with_transfer_blocks(10, None).await;
+    async fn account_state_at_block_reflects_history() {
+        let home = tempdir().unwrap();
+        let store = IndexerStore::open_db(home.as_ref()).unwrap();
 
-        let acc1_at_1 = storage.account_state_at_block(&from, 1).unwrap();
-        let acc2_at_1 = storage.account_state_at_block(&to, 1).unwrap();
-        assert_eq!(acc1_at_1.balance, 9990);
-        assert_eq!(acc2_at_1.balance, 20010);
+        let initial_accounts = initial_pub_accounts_private_keys();
+        let from = initial_accounts[0].account_id;
+        let to = initial_accounts[1].account_id;
+        let sign_key = initial_accounts[0].pub_sign_key.clone();
 
-        let acc1_at_5 = storage.account_state_at_block(&from, 5).unwrap();
-        let acc2_at_5 = storage.account_state_at_block(&to, 5).unwrap();
-        assert_eq!(acc1_at_5.balance, 9950);
-        assert_eq!(acc2_at_5.balance, 20050);
+        let genesis = produce_dummy_block(1, None, vec![]);
+        let mut prev_hash = genesis.header.hash;
+        store.accept_block(&genesis, Slot::from(0)).await.unwrap();
 
-        let acc1_at_9 = storage.account_state_at_block(&from, 9).unwrap();
-        let acc2_at_9 = storage.account_state_at_block(&to, 9).unwrap();
-        assert_eq!(acc1_at_9.balance, 9910);
-        assert_eq!(acc2_at_9.balance, 20090);
+        for i in 0..10_u64 {
+            let tx = create_transaction_native_token_transfer(from, i.into(), to, 10, &sign_key);
+            let block = produce_dummy_block(i + 2, Some(prev_hash), vec![tx]);
+            prev_hash = block.header.hash;
+            store.accept_block(&block, Slot::from(0)).await.unwrap();
+        }
+
+        // State at block N is inclusive of block N.
+        // Block 1 (genesis, clock-only): no transfers yet.
+        assert_eq!(
+            store.account_state_at_block(&from, 1).unwrap().balance,
+            10000
+        );
+        assert_eq!(store.account_state_at_block(&to, 1).unwrap().balance, 20000);
+        // Through block 5: 4 transfers applied (blocks 2..=5).
+        assert_eq!(
+            store.account_state_at_block(&from, 5).unwrap().balance,
+            9960
+        );
+        assert_eq!(store.account_state_at_block(&to, 5).unwrap().balance, 20040);
+        // Through block 9: 8 transfers applied (blocks 2..=9).
+        assert_eq!(
+            store.account_state_at_block(&from, 9).unwrap().balance,
+            9920
+        );
+        assert_eq!(store.account_state_at_block(&to, 9).unwrap().balance, 20080);
+    }
+}
+
+#[cfg(test)]
+mod accept_tests {
+    use chain_consistency::BlockIngestError;
+    use common::{HashType, block::HashableBlockData, test_utils::produce_dummy_block};
+
+    use super::*;
+
+    fn signing_key() -> lee::PrivateKey {
+        lee::PrivateKey::try_new([7_u8; 32]).expect("valid key")
+    }
+
+    // A block with a correct hash but empty body — enough to exercise the
+    // acceptance checks (id/link/hash), which run before any state application.
+    fn valid_hash_block(block_id: u64, prev: HashType) -> common::block::Block {
+        HashableBlockData {
+            block_id,
+            prev_block_hash: prev,
+            timestamp: 0,
+            transactions: vec![],
+        }
+        .into_pending_block(&signing_key())
+    }
+
+    #[tokio::test]
+    async fn non_genesis_first_block_parks_with_unexpected_id() {
+        let dir = tempfile::tempdir().expect("tempdir");
+        let store = IndexerStore::open_db(dir.path()).expect("open store");
+
+        let block = valid_hash_block(2, HashType([0_u8; 32]));
+        let outcome = store
+            .accept_block(&block, Slot::from(0))
+            .await
+            .expect("accept");
+
+        assert!(matches!(
+            outcome,
+            AcceptOutcome::Parked(BlockIngestError::UnexpectedBlockId {
+                expected: 1,
+                got: 2
+            })
+        ));
+        let stall = store.get_stall_reason().expect("get").expect("present");
+        assert_eq!(stall.block_id, Some(2));
+        assert_eq!(stall.orphans_since, 0);
+    }
+
+    #[tokio::test]
+    async fn hash_mismatch_parks() {
+        let dir = tempfile::tempdir().expect("tempdir");
+        let store = IndexerStore::open_db(dir.path()).expect("open store");
+
+        let mut block = valid_hash_block(1, HashType([0_u8; 32]));
+        block.header.timestamp = 999; // invalidates the stored hash
+
+        let outcome = store
+            .accept_block(&block, Slot::from(0))
+            .await
+            .expect("accept");
+        assert!(matches!(
+            outcome,
+            AcceptOutcome::Parked(BlockIngestError::HashMismatch { .. })
+        ));
+    }
+
+    #[tokio::test]
+    async fn second_break_bumps_orphan_count_and_keeps_first() {
+        let dir = tempfile::tempdir().expect("tempdir");
+        let store = IndexerStore::open_db(dir.path()).expect("open store");
+
+        let first = valid_hash_block(2, HashType([0_u8; 32]));
+        store
+            .accept_block(&first, Slot::from(0))
+            .await
+            .expect("accept");
+        let second = valid_hash_block(3, HashType([0_u8; 32]));
+        store
+            .accept_block(&second, Slot::from(0))
+            .await
+            .expect("accept");
+
+        let stall = store.get_stall_reason().expect("get").expect("present");
+        assert_eq!(stall.block_id, Some(2), "first stall preserved");
+        assert_eq!(stall.orphans_since, 1, "second break counted as orphan");
+    }
+
+    #[tokio::test]
+    async fn deserialize_break_records_stall_without_header() {
+        let dir = tempfile::tempdir().expect("tempdir");
+        let store = IndexerStore::open_db(dir.path()).expect("open store");
+
+        store
+            .record_stall(
+                None,
+                Slot::from(0),
+                BlockIngestError::Deserialize("bad bytes".to_owned()),
+            )
+            .expect("record");
+
+        let stall = store.get_stall_reason().expect("get").expect("present");
+        assert_eq!(stall.block_id, None);
+        assert!(matches!(stall.error, BlockIngestError::Deserialize(_)));
+    }
+
+    #[tokio::test]
+    async fn parks_then_recovers_on_valid_continuation() {
+        let dir = tempfile::tempdir().expect("tempdir");
+        let store = IndexerStore::open_db(dir.path()).expect("open store");
+
+        // Genesis (block 1, clock-only) applies and advances the tip.
+        let genesis = produce_dummy_block(1, None, vec![]);
+        assert!(matches!(
+            store.accept_block(&genesis, Slot::from(0)).await.unwrap(),
+            AcceptOutcome::Applied
+        ));
+
+        // A block that skips ahead (id 3 while the tip is 1) parks the indexer.
+        let bad = produce_dummy_block(3, Some(genesis.header.hash), vec![]);
+        assert!(matches!(
+            store.accept_block(&bad, Slot::from(0)).await.unwrap(),
+            AcceptOutcome::Parked(BlockIngestError::UnexpectedBlockId {
+                expected: 2,
+                got: 3
+            })
+        ));
+        assert!(
+            store.get_stall_reason().unwrap().is_some(),
+            "indexer should be parked after the bad block"
+        );
+        assert_eq!(
+            store.get_last_block_id().unwrap(),
+            Some(1),
+            "validated tip must stay frozen at genesis while parked"
+        );
+
+        // The valid continuation (block 2 chaining on genesis) recovers the chain.
+        let next = produce_dummy_block(2, Some(genesis.header.hash), vec![]);
+        assert!(matches!(
+            store.accept_block(&next, Slot::from(0)).await.unwrap(),
+            AcceptOutcome::Applied
+        ));
+        assert!(
+            store.get_stall_reason().unwrap().is_none(),
+            "stall reason must clear on recovery"
+        );
+        assert_eq!(
+            store.get_last_block_id().unwrap(),
+            Some(2),
+            "tip must advance to the recovered block"
+        );
+    }
+
+    #[tokio::test]
+    async fn redelivered_tip_block_is_idempotent_not_parked() {
+        use testnet_initial_state::initial_pub_accounts_private_keys;
+
+        let dir = tempfile::tempdir().expect("tempdir");
+        let store = IndexerStore::open_db(dir.path()).expect("open store");
+
+        let accounts = initial_pub_accounts_private_keys();
+        let from = accounts[0].account_id;
+        let to = accounts[1].account_id;
+        let sign_key = accounts[0].pub_sign_key.clone();
+
+        let genesis = produce_dummy_block(1, None, vec![]);
+        store
+            .accept_block(&genesis, Slot::from(0))
+            .await
+            .expect("accept genesis");
+
+        // Block 2: a single transfer of 10.
+        let tx = common::test_utils::create_transaction_native_token_transfer(
+            from, 0, to, 10, &sign_key,
+        );
+        let block = produce_dummy_block(2, Some(genesis.header.hash), vec![tx]);
+        assert!(matches!(
+            store.accept_block(&block, Slot::from(0)).await.unwrap(),
+            AcceptOutcome::Applied
+        ));
+        let balance_after = store.account_current_state(&from).await.unwrap().balance;
+
+        // Re-deliver the exact same block: idempotent skip, no state change, no park.
+        assert!(matches!(
+            store.accept_block(&block, Slot::from(0)).await.unwrap(),
+            AcceptOutcome::AlreadyApplied
+        ));
+        assert_eq!(
+            store.account_current_state(&from).await.unwrap().balance,
+            balance_after,
+            "re-delivered block must not be applied twice"
+        );
+        assert_eq!(
+            store.get_last_block_id().unwrap(),
+            Some(2),
+            "tip must stay at the already-applied block"
+        );
+        assert!(
+            store.get_stall_reason().unwrap().is_none(),
+            "a benign duplicate must not park the indexer"
+        );
+    }
+
+    #[tokio::test]
+    async fn redelivered_block_below_tip_is_idempotent_not_parked() {
+        use testnet_initial_state::initial_pub_accounts_private_keys;
+
+        let dir = tempfile::tempdir().expect("tempdir");
+        let store = IndexerStore::open_db(dir.path()).expect("open store");
+
+        let accounts = initial_pub_accounts_private_keys();
+        let from = accounts[0].account_id;
+        let to = accounts[1].account_id;
+        let sign_key = accounts[0].pub_sign_key.clone();
+
+        // Build a short chain: genesis (1) -> block 2 -> block 3, so the tip is 3.
+        let genesis = produce_dummy_block(1, None, vec![]);
+        store
+            .accept_block(&genesis, Slot::from(0))
+            .await
+            .expect("accept genesis");
+
+        let tx2 = common::test_utils::create_transaction_native_token_transfer(
+            from, 0, to, 10, &sign_key,
+        );
+        let block2 = produce_dummy_block(2, Some(genesis.header.hash), vec![tx2]);
+        assert!(matches!(
+            store.accept_block(&block2, Slot::from(0)).await.unwrap(),
+            AcceptOutcome::Applied
+        ));
+
+        let tx3 = common::test_utils::create_transaction_native_token_transfer(
+            from, 1, to, 10, &sign_key,
+        );
+        let block3 = produce_dummy_block(3, Some(block2.header.hash), vec![tx3]);
+        assert!(matches!(
+            store.accept_block(&block3, Slot::from(0)).await.unwrap(),
+            AcceptOutcome::Applied
+        ));
+
+        let balance_after = store.account_current_state(&from).await.unwrap().balance;
+
+        // Re-deliver block 2 (id below the tip): a re-delivery, not a divergence.
+        assert!(matches!(
+            store.accept_block(&block2, Slot::from(0)).await.unwrap(),
+            AcceptOutcome::AlreadyApplied
+        ));
+        assert_eq!(
+            store.account_current_state(&from).await.unwrap().balance,
+            balance_after,
+            "re-delivered block below the tip must not be applied again"
+        );
+        assert_eq!(
+            store.get_last_block_id().unwrap(),
+            Some(3),
+            "tip must stay at the current head"
+        );
+        assert!(
+            store.get_stall_reason().unwrap().is_none(),
+            "a benign re-delivery must not park the indexer"
+        );
     }
 }

lez/indexer/core/src/config.rs

@@ -20,6 +20,13 @@ pub struct IndexerConfig {
     pub consensus_info_polling_interval: Duration,
     pub bedrock_config: ClientConfig,
     pub channel_id: ChannelId,
+    /// Whether to wipe the indexer store and re-index from scratch when the startup
+    /// chain-identity check finds the channel serving a different block than the one
+    /// stored at the same id.
+    ///
+    /// Defaults to `false`: on mismatch the indexer refuses to start.
+    #[serde(default)]
+    pub allow_chain_reset: bool,
 }
 
 impl IndexerConfig {

lez/indexer/core/src/lib.rs

@@ -2,28 +2,31 @@ use std::{path::Path, sync::Arc};
 
 use anyhow::Result;
 use arc_swap::ArcSwap;
+use chain_consistency::{Anchor, BlockIngestError, ChainConsistency};
 use common::block::Block;
-// ToDo: Remove after testnet
+// TODO: Remove after testnet
 use futures::StreamExt as _;
 use log::{error, info, warn};
-use logos_blockchain_core::header::HeaderId;
 use logos_blockchain_zone_sdk::{
-    CommonHttpClient, ZoneMessage, adapter::NodeHttpClient, indexer::ZoneIndexer,
+    CommonHttpClient, Slot, ZoneMessage, adapter::NodeHttpClient, indexer::ZoneIndexer,
 };
+pub use stall_reason::StallReason;
 
 use crate::{
-    block_store::IndexerStore,
+    block_store::{AcceptOutcome, IndexerStore},
     config::IndexerConfig,
     status::{IndexerStatus, IndexerSyncStatus},
 };
-
 pub mod block_store;
 pub mod config;
+pub mod stall_reason;
 pub mod status;
 
 #[derive(Clone)]
 pub struct IndexerCore {
     pub zone_indexer: Arc<ZoneIndexer<NodeHttpClient>>,
+    /// Direct node handle for queries outside `ZoneIndexer`'s streaming API.
+    pub node: NodeHttpClient,
     pub config: IndexerConfig,
     pub store: IndexerStore,
     /// Live ingestion status; updated by the ingest stream, read by `status`.
@@ -31,7 +34,42 @@ pub struct IndexerCore {
 }
 
 impl IndexerCore {
-    pub fn new(config: IndexerConfig, storage_dir: &Path) -> Result<Self> {
+    /// Builds the core, then verifies the stored chain matches the channel's by
+    /// re-reading the channel at the stored tip's position.
+    ///
+    /// On mismatch: refuse (error) unless `config.allow_chain_reset` is set, in which case wipe the
+    /// store and re-index from scratch.
+    pub async fn new(config: IndexerConfig, storage_dir: &Path) -> Result<Self> {
+        let home = storage_dir.join(format!("rocksdb-{}", config.channel_id));
+        let core = Self::open(config.clone(), storage_dir)?;
+        match core.verify_chain_consistency().await? {
+            // `Inconclusive` is deliberately treated the same as `Consistent`.
+            //
+            // We could not prove a reset, so proceed from the cursor without wiping
+            // a possibly-valid store. A genuinely divergent chain is still caught
+            // later when the ingest loop tries to apply and parks.
+            ChainConsistency::Consistent | ChainConsistency::Inconclusive => Ok(core),
+            ChainConsistency::Inconsistent(mismatch) if config.allow_chain_reset => {
+                warn!(
+                    "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)
+            }
+            ChainConsistency::Inconsistent(mismatch) => Err(anyhow::anyhow!(
+                "Indexer store at {} holds a different chain than the channel now serves \
+                 ({mismatch}). Delete the indexer storage directory, point at a fresh one, or \
+                 set `allow_chain_reset` in the indexer config.",
+                home.display()
+            )),
+        }
+    }
+
+    /// Opens the store and builds the core without the chain-identity check.
+    fn open(config: IndexerConfig, storage_dir: &Path) -> Result<Self> {
         // Namespace the DB by channel so indexers on different channels can
         // share a storage dir without their RocksDB state colliding.
         let home = storage_dir.join(format!("rocksdb-{}", config.channel_id));
@@ -41,16 +79,46 @@ impl IndexerCore {
             CommonHttpClient::new(basic_auth),
             config.bedrock_config.addr.clone(),
         );
-        let zone_indexer = ZoneIndexer::new(config.channel_id, node);
+        let zone_indexer = ZoneIndexer::new(config.channel_id, node.clone());
 
         Ok(Self {
             zone_indexer: Arc::new(zone_indexer),
+            node,
             config,
             store: IndexerStore::open_db(&home)?,
             status: Arc::new(ArcSwap::from_pointee(IndexerSyncStatus::starting())),
         })
     }
 
+    /// Detects when the local store belongs to a different chain than the connected
+    /// L1 (e.g. a wiped/restarted bedrock) so startup can reset instead of silently
+    /// diverging. Best-effort dev convenience; won't trigger during normal live indexing.
+    ///
+    /// The anchoring/probe logic lives in the shared [`chain_consistency`] crate;
+    /// here we build the [`Anchor`] from the store (the recorded parked block while
+    /// stalled, otherwise the tip at the read cursor — only blocks advance the
+    /// cursor, so when not parked it is the tip's inscription slot) and delegate.
+    async fn verify_chain_consistency(&self) -> Result<ChainConsistency> {
+        let anchor = if let Some(stall) = self.store.get_stall_reason()? {
+            Anchor::new(stall.l1_slot, stall.block_id.zip(stall.block_hash))
+        } else {
+            let Some(cursor) = self.store.get_zone_cursor()? else {
+                // if we have no cursor, the store is empty or cold: nothing to compare
+                return Ok(ChainConsistency::Inconclusive);
+            };
+            let Some(tip_id) = self.store.get_last_block_id()? else {
+                return Ok(ChainConsistency::Inconclusive);
+            };
+            let Some(tip) = self.store.get_block_at_id(tip_id)? else {
+                return Ok(ChainConsistency::Inconclusive);
+            };
+            Anchor::new(cursor, Some((tip_id, tip.header.hash)))
+        };
+
+        chain_consistency::verify_chain_consistency(&self.node, self.config.channel_id, &anchor)
+            .await
+    }
+
     /// Snapshot of the current ingestion status (sync state + indexed tip).
     ///
     /// Combines the ingest loop's live status with the L2 tip read fresh from the
@@ -58,10 +126,27 @@ impl IndexerCore {
     #[must_use]
     pub fn status(&self) -> IndexerStatus {
         let sync = IndexerSyncStatus::clone(&self.status.load());
-        let indexed_block_id = self.store.get_last_block_id().ok().flatten();
+        // Log-and-fall-back rather than collapsing a store error into the same
+        // `None` as "legitimately absent": a DB read failure must not silently
+        // masquerade as "no tip yet" / "no stall recorded" in the snapshot.
+        let indexed_block_id = match self.store.get_last_block_id() {
+            Ok(id) => id,
+            Err(err) => {
+                warn!("Failed to read last indexed block id for status: {err:#}");
+                None
+            }
+        };
+        let stall_reason = match self.store.get_stall_reason() {
+            Ok(reason) => reason,
+            Err(err) => {
+                warn!("Failed to read stall reason for status: {err:#}");
+                None
+            }
+        };
         IndexerStatus {
             sync,
             indexed_block_id,
+            stall_reason,
         }
     }
 
@@ -70,6 +155,35 @@ impl IndexerCore {
         self.status.store(Arc::new(status));
     }
 
+    /// Advances the in-memory L1 read cursor past `slot` and persists it.
+    /// A persist failure is only logged: the worst case is re-reading a batch
+    /// after a restart, which ingestion handles idempotently.
+    fn advance_cursor(&self, cursor: &mut Option<Slot>, slot: Slot) {
+        *cursor = Some(slot);
+        if let Err(err) = self.store.set_zone_cursor(&slot) {
+            warn!("Failed to persist indexer cursor: {err:#}");
+        }
+    }
+
+    /// 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: 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: {reason}"
+        )));
+    }
+
     pub fn subscribe_parse_block_stream(&self) -> impl futures::Stream<Item = Result<Block>> + '_ {
         let poll_interval = self.config.consensus_info_polling_interval;
         let initial_cursor = self
@@ -80,8 +194,8 @@ impl IndexerCore {
         async_stream::stream! {
             let mut cursor = initial_cursor;
 
-            if cursor.is_some() {
-                info!("Resuming indexer from cursor {cursor:?}");
+            if let Some(slot) = &cursor {
+                info!("Resuming indexer from cursor {slot:?}");
             } else {
                 info!("Starting indexer from beginning of channel");
             }
@@ -90,8 +204,6 @@ impl IndexerCore {
                 let stream = match self.zone_indexer.next_messages(cursor).await {
                     Ok(s) => s,
                     Err(err) => {
-                        // `next_messages` reads L1 consensus info internally, so
-                        // this also covers an unreachable/misconfigured L1 node.
                         error!("Failed to start zone-sdk next_messages stream: {err}");
                         self.set_status(IndexerSyncStatus::error(format!(
                             "cannot reach L1 / read channel: {err}"
@@ -102,11 +214,8 @@ impl IndexerCore {
                 };
                 let mut stream = std::pin::pin!(stream);
 
-                // Flip to Syncing on the first message of this cycle (not merely on
-                // a successful poll) so the steady-state CaughtUp status doesn't
-                // flicker. Until then the state stays Starting (cold-start scan of
-                // empty L1 history) or CaughtUp (idle).
                 let mut announced_syncing = false;
+                let mut had_cycle_error = false;
 
                 while let Some((msg, slot)) = stream.next().await {
                     if !announced_syncing {
@@ -116,48 +225,163 @@ impl IndexerCore {
 
                     let zone_block = match msg {
                         ZoneMessage::Block(b) => b,
-                        // Non-block messages don't carry a cursor position; the
-                        // next ZoneBlock advances past them implicitly.
+                        // FIXME: will be handled in prep of decentralized sequencers
                         ZoneMessage::Deposit(_) | ZoneMessage::Withdraw(_) => continue,
                     };
 
                     let block: Block = match borsh::from_slice(&zone_block.data) {
                         Ok(b) => b,
-                        Err(e) => {
-                            error!("Failed to deserialize L2 block from zone-sdk: {e}");
-                            // Advance past the broken inscription so we don't
-                            // re-process it on restart.
-                            cursor = Some(slot);
-                            if let Err(err) = self.store.set_zone_cursor(&slot) {
-                                warn!("Failed to persist indexer cursor: {err:#}");
-                            }
+                        Err(error) => {
+                            self.park_undeserializable(slot, error);
+                            // L1 proceeds regardless
+                            self.advance_cursor(&mut cursor, slot);
                             continue;
                         }
                     };
 
-                    info!("Indexed L2 block {}", block.header.block_id);
-
-                    // TODO: Remove l1_header placeholder once storage layer
-                    // no longer requires it. Zone-sdk handles L1 tracking internally.
-                    let placeholder_l1_header = HeaderId::from([0_u8; 32]);
-                    if let Err(err) = self.store.put_block(block.clone(), placeholder_l1_header).await {
-                        error!("Failed to store block {}: {err:#}", block.header.block_id);
+                    match self.store.accept_block(&block, slot).await {
+                        Ok(AcceptOutcome::Applied) => {
+                            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) => {
+                            info!(
+                                "Skipping already-applied block {}",
+                                block.header.block_id
+                            );
+                            self.advance_cursor(&mut cursor, slot);
+                        }
+                        Ok(AcceptOutcome::Parked(ingest_err)) => {
+                            error!(
+                                "Parked at block {}: {ingest_err}",
+                                block.header.block_id
+                            );
+                            self.set_status(IndexerSyncStatus::stalled(ingest_err.to_string()));
+                            // L1 proceeds regardless
+                            self.advance_cursor(&mut cursor, slot);
+                        }
+                        Err(err) => {
+                            // Infrastructure error (DB read/write), not a bad block.
+                            // will re-poll from the same cursor next cycle.
+                            error!(
+                                "Store error applying block {}: {err:#}",
+                                block.header.block_id
+                            );
+                            self.set_status(IndexerSyncStatus::error(format!(
+                                "store error: {err:#}"
+                            )));
+                            had_cycle_error = true;
+                            break;
+                        }
                     }
-
-                    cursor = Some(slot);
-                    if let Err(err) = self.store.set_zone_cursor(&slot) {
-                        warn!("Failed to persist indexer cursor: {err:#}");
-                    }
-                    yield Ok(block);
                 }
 
-                // Stream drained: caught up to LIB as of this cycle. Clears any
-                // prior error (e.g. a transient L1 disconnect that left no
-                // backlog, so the `Syncing` branch above never ran). Sleep then
-                // poll again.
-                self.set_status(IndexerSyncStatus::caught_up());
+                if had_cycle_error {
+                    tokio::time::sleep(poll_interval).await;
+                    continue;
+                }
+
+                // 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.
+                match self.store.get_stall_reason() {
+                    Ok(None) => self.set_status(IndexerSyncStatus::caught_up()),
+                    Ok(Some(_)) => {}
+                    Err(err) => {
+                        warn!("Failed to read stall reason after draining stream; not marking caught up: {err:#}");
+                    }
+                }
                 tokio::time::sleep(poll_interval).await;
             }
         }
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use std::time::Duration;
+
+    use common::{HashType, block::HashableBlockData};
+    use logos_blockchain_zone_sdk::Slot;
+
+    use super::*;
+    use crate::config::{ChannelId, ClientConfig, IndexerConfig};
+
+    fn unreachable_core(dir: &std::path::Path) -> IndexerCore {
+        let config = IndexerConfig {
+            consensus_info_polling_interval: Duration::from_secs(1),
+            bedrock_config: ClientConfig {
+                addr: "http://localhost:1".parse().expect("url"),
+                auth: None,
+            },
+            channel_id: ChannelId::from([1; 32]),
+            allow_chain_reset: false,
+        };
+        IndexerCore::open(config, dir).expect("open core")
+    }
+
+    fn test_block(block_id: u64, timestamp: u64) -> Block {
+        HashableBlockData {
+            block_id,
+            prev_block_hash: HashType([0; 32]),
+            timestamp,
+            transactions: vec![],
+        }
+        .into_pending_block(&lee::PrivateKey::try_new([7; 32]).expect("valid key"))
+    }
+
+    #[tokio::test]
+    async fn cold_store_is_inconclusive() {
+        // An empty store has no cursor, so there is nothing to compare: the check
+        // must be Inconclusive (not Consistent), and it returns before any L1 read.
+        let dir = tempfile::tempdir().expect("tempdir");
+        let core = unreachable_core(dir.path());
+        assert!(matches!(
+            core.verify_chain_consistency().await.expect("verify"),
+            ChainConsistency::Inconclusive
+        ));
+    }
+
+    #[tokio::test]
+    async fn parked_store_with_unreachable_node_is_inconclusive() {
+        // Network failure is not evidence of a reset: a parked store must stay
+        // parked (Inconclusive), not error out or trip the wipe path.
+        let dir = tempfile::tempdir().expect("tempdir");
+        let core = unreachable_core(dir.path());
+        let parked = test_block(5, 42);
+        core.store
+            .record_stall(
+                Some(&parked.header),
+                Slot::from(1_000),
+                BlockIngestError::EmptyBlock,
+            )
+            .expect("record stall");
+        assert!(matches!(
+            core.verify_chain_consistency().await.expect("verify"),
+            ChainConsistency::Inconclusive
+        ));
+    }
+
+    #[tokio::test]
+    async fn caught_up_store_with_unreachable_node_is_inconclusive() {
+        let dir = tempfile::tempdir().expect("tempdir");
+        let core = unreachable_core(dir.path());
+        let genesis = common::test_utils::produce_dummy_block(1, None, vec![]);
+        assert!(matches!(
+            core.store
+                .accept_block(&genesis, Slot::from(1_000))
+                .await
+                .expect("accept"),
+            AcceptOutcome::Applied
+        ));
+        core.store
+            .set_zone_cursor(&Slot::from(1_000))
+            .expect("set cursor");
+        assert!(matches!(
+            core.verify_chain_consistency().await.expect("verify"),
+            ChainConsistency::Inconclusive
+        ));
+    }
+}

lez/indexer/core/src/stall_reason.rs

@@ -0,0 +1,25 @@
+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,9 +1,10 @@
 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)]
-#[serde(rename_all = "snake_case")]
 pub enum IndexerSyncState {
     /// Booted; no ingestion cycle has run yet.
     Starting,
@@ -13,12 +14,14 @@ pub enum IndexerSyncState {
     CaughtUp,
     /// The last cycle failed (e.g. the L1 node is unreachable). See `last_error`.
     Error,
+    /// Parked on a stall reason: the validated tip is frozen awaiting a valid
+    /// continuation. See `last_error` and the snapshot's `stall_reason`.
+    Stalled,
 }
 
 /// Live ingestion status owned by the ingest loop: the coarse `state` plus the
 /// reason when it is `Error`.
 #[derive(Debug, Clone, Serialize)]
-#[serde(rename_all = "camelCase")]
 pub struct IndexerSyncStatus {
     pub state: IndexerSyncState,
     pub last_error: Option<String>,
@@ -56,6 +59,15 @@ impl IndexerSyncStatus {
             last_error: Some(reason),
         }
     }
+
+    /// Parked on a stall reason; `reason` mirrors the stall's error message.
+    /// The full stall is attached to the [`IndexerStatus`] snapshot.
+    pub(crate) const fn stalled(reason: String) -> Self {
+        Self {
+            state: IndexerSyncState::Stalled,
+            last_error: Some(reason),
+        }
+    }
 }
 
 /// Full status snapshot returned to callers (FFI/RPC): the live [`IndexerSyncStatus`]
@@ -64,11 +76,11 @@ impl IndexerSyncStatus {
 /// The tip is tracked by the store, not the ingest loop, so it lives here on the
 /// returned snapshot rather than inside the shared [`IndexerSyncStatus`].
 #[derive(Debug, Clone, Serialize)]
-#[serde(rename_all = "camelCase")]
 pub struct IndexerStatus {
     #[serde(flatten)]
     pub sync: IndexerSyncStatus,
     pub indexed_block_id: Option<u64>,
+    pub stall_reason: Option<StallReason>,
 }
 
 #[cfg(test)]
@@ -80,14 +92,16 @@ mod tests {
         let status = IndexerStatus {
             sync: IndexerSyncStatus::error("boom".to_owned()),
             indexed_block_id: Some(7),
+            stall_reason: None,
         };
         let value = serde_json::to_value(&status).expect("serialize");
         assert_eq!(
             value,
             serde_json::json!({
-                "state": "error",
-                "lastError": "boom",
-                "indexedBlockId": 7,
+                "state": "Error",
+                "last_error": "boom",
+                "indexed_block_id": 7,
+                "stall_reason": null,
             })
         );
     }
@@ -97,7 +111,36 @@ mod tests {
         let value = serde_json::to_value(IndexerSyncStatus::caught_up()).expect("serialize");
         assert_eq!(
             value,
-            serde_json::json!({ "state": "caught_up", "lastError": null })
+            serde_json::json!({ "state": "CaughtUp", "last_error": null })
         );
     }
+
+    #[test]
+    fn stalled_status_serializes_with_stall_reason() {
+        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),
+            stall_reason: Some(StallReason {
+                block_id: Some(42),
+                block_hash: None,
+                prev_block_hash: None,
+                l1_slot: Slot::from(0),
+                error: BlockIngestError::StateTransition {
+                    tx_index: 0,
+                    reason: String::default(),
+                },
+                first_seen: None,
+                orphans_since: 2,
+            }),
+        };
+        let value = serde_json::to_value(&status).expect("serialize");
+        assert_eq!(value["state"], serde_json::json!("Stalled"));
+        assert_eq!(value["last_error"], serde_json::json!("broken chain link"));
+        assert_eq!(value["indexed_block_id"], serde_json::json!(41));
+        assert_eq!(value["stall_reason"]["orphans_since"], serde_json::json!(2));
+    }
 }

lez/indexer/ffi/indexer_ffi.h

@@ -503,9 +503,9 @@ struct LastBlockIdResult query_last_block(const struct IndexerServiceFFI *indexe
  * Query the indexer's current sync status as a JSON C-string.
  *
  * The JSON schema is owned by `indexer_core` (`IndexerStatus`): an object with
- * `state` (`starting`/`syncing`/`caught_up`/`error`), `indexedBlockId`, and
- * `lastError`. Lets a client distinguish "still catching up" from "something
- * went wrong".
+ * `state` (`Starting`/`Syncing`/`CaughtUp`/`Error`/`Stalled`),
+ * `indexed_block_id`, `last_error`, and `stall_reason`. Lets a client
+ * distinguish "still catching up" from "something went wrong".
  *
  * # Arguments
  *

lez/indexer/ffi/src/api/lifecycle.rs

@@ -112,10 +112,12 @@ unsafe fn setup_indexer(
         unsafe { Runtime::from_borrowed(caller.as_ref()) }
     };
 
-    let core = IndexerCore::new(config, &storage_dir).map_err(|e| {
-        log::error!("Could not initialize indexer core: {e}");
-        OperationStatus::InitializationError
-    })?;
+    let core = runtime
+        .block_on(IndexerCore::new(config, &storage_dir))
+        .map_err(|e| {
+            log::error!("Could not initialize indexer core: {e}");
+            OperationStatus::InitializationError
+        })?;
 
     // The block stream writes each parsed block into the store as a side effect
     // of being polled, so we spawn a task that simply drains it. There are no

lez/indexer/ffi/src/api/query.rs

@@ -91,9 +91,9 @@ pub unsafe extern "C" fn query_last_block(indexer: *const IndexerServiceFFI) ->
 /// Query the indexer's current sync status as a JSON C-string.
 ///
 /// The JSON schema is owned by `indexer_core` (`IndexerStatus`): an object with
-/// `state` (`starting`/`syncing`/`caught_up`/`error`), `indexedBlockId`, and
-/// `lastError`. Lets a client distinguish "still catching up" from "something
-/// went wrong".
+/// `state` (`Starting`/`Syncing`/`CaughtUp`/`Error`/`Stalled`),
+/// `indexed_block_id`, `last_error`, and `stall_reason`. Lets a client
+/// distinguish "still catching up" from "something went wrong".
 ///
 /// # Arguments
 ///

lez/indexer/service/configs/debug/indexer_config.json

@@ -1,7 +1,8 @@
 {
-    "consensus_info_polling_interval": "1s",
-    "bedrock_config": {
-        "addr": "http://localhost:18080"
-    },
-    "channel_id": "0101010101010101010101010101010101010101010101010101010101010101"
+  "consensus_info_polling_interval": "1s",
+  "bedrock_config": {
+    "addr": "http://localhost:18080"
+  },
+  "channel_id": "0101010101010101010101010101010101010101010101010101010101010101",
+  "allow_chain_reset": true
 }

lez/indexer/service/configs/docker/indexer_config.json

@@ -3,5 +3,6 @@
   "bedrock_config": {
     "addr": "http://host.docker.internal:18080"
   },
-  "channel_id": "0101010101010101010101010101010101010101010101010101010101010101"
+  "channel_id": "0101010101010101010101010101010101010101010101010101010101010101",
+  "allow_chain_reset": true
 }

lez/indexer/service/protocol/Cargo.toml

@@ -11,6 +11,7 @@ workspace = true
 lee_core = { workspace = true, optional = true, features = ["host"] }
 lee = { workspace = true, optional = true }
 common = { workspace = true, optional = true }
+indexer_core = { workspace = true, optional = true }
 
 serde = { workspace = true, features = ["derive"] }
 serde_with.workspace = true
@@ -22,4 +23,4 @@ anyhow.workspace = true
 
 [features]
 # Enable conversion to/from LEE core types
-convert = ["dep:lee_core", "dep:lee", "dep:common"]
+convert = ["dep:lee_core", "dep:lee", "dep:common", "dep:indexer_core"]

lez/indexer/service/protocol/src/convert.rs

@@ -3,11 +3,12 @@
 use lee_core::account::Nonce;
 
 use crate::{
-    Account, AccountId, BedrockStatus, Block, BlockBody, BlockHeader, Ciphertext, Commitment,
-    CommitmentSetDigest, Data, EncryptedAccountData, EphemeralPublicKey, HashType, Nullifier,
-    PrivacyPreservingMessage, PrivacyPreservingTransaction, ProgramDeploymentMessage,
-    ProgramDeploymentTransaction, ProgramId, Proof, PublicKey, PublicMessage, PublicTransaction,
-    Signature, Transaction, ValidityWindow, WitnessSet,
+    Account, AccountId, BedrockStatus, Block, BlockBody, BlockHeader, BlockIngestError, Ciphertext,
+    Commitment, CommitmentSetDigest, Data, EncryptedAccountData, EphemeralPublicKey, HashType,
+    IndexerStatus, IndexerSyncState, Nullifier, PrivacyPreservingMessage,
+    PrivacyPreservingTransaction, ProgramDeploymentMessage, ProgramDeploymentTransaction,
+    ProgramId, Proof, PublicKey, PublicMessage, PublicTransaction, Signature, StallReason,
+    Transaction, ValidityWindow, WitnessSet,
 };
 
 // ============================================================================
@@ -707,3 +708,94 @@ impl TryFrom<ValidityWindow> for lee_core::program::ValidityWindow<u64> {
         value.0.try_into()
     }
 }
+
+// ============================================================================
+// Indexer status conversions
+// ============================================================================
+
+impl From<indexer_core::status::IndexerSyncState> for IndexerSyncState {
+    fn from(value: indexer_core::status::IndexerSyncState) -> Self {
+        match value {
+            indexer_core::status::IndexerSyncState::Starting => Self::Starting,
+            indexer_core::status::IndexerSyncState::Syncing => Self::Syncing,
+            indexer_core::status::IndexerSyncState::CaughtUp => Self::CaughtUp,
+            indexer_core::status::IndexerSyncState::Error => Self::Error,
+            indexer_core::status::IndexerSyncState::Stalled => Self::Stalled,
+        }
+    }
+}
+
+impl From<indexer_core::BlockIngestError> for BlockIngestError {
+    fn from(value: indexer_core::BlockIngestError) -> Self {
+        match value {
+            indexer_core::BlockIngestError::Deserialize(msg) => Self::Deserialize(msg),
+            indexer_core::BlockIngestError::UnexpectedBlockId { expected, got } => {
+                Self::UnexpectedBlockId { expected, got }
+            }
+            indexer_core::BlockIngestError::BrokenChainLink {
+                expected_prev,
+                got_prev,
+            } => Self::BrokenChainLink {
+                expected_prev: expected_prev.into(),
+                got_prev: got_prev.into(),
+            },
+            indexer_core::BlockIngestError::HashMismatch { computed, header } => {
+                Self::HashMismatch {
+                    computed: computed.into(),
+                    header: header.into(),
+                }
+            }
+            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 }
+            }
+        }
+    }
+}
+
+impl From<indexer_core::StallReason> for StallReason {
+    fn from(value: indexer_core::StallReason) -> Self {
+        let indexer_core::StallReason {
+            block_id,
+            block_hash,
+            prev_block_hash,
+            l1_slot,
+            error,
+            first_seen,
+            orphans_since,
+        } = value;
+
+        Self {
+            block_id,
+            block_hash: block_hash.map(Into::into),
+            prev_block_hash: prev_block_hash.map(Into::into),
+            l1_slot: l1_slot.into_inner(),
+            error: error.into(),
+            first_seen,
+            orphans_since,
+        }
+    }
+}
+
+impl From<indexer_core::status::IndexerStatus> for IndexerStatus {
+    fn from(value: indexer_core::status::IndexerStatus) -> Self {
+        let indexer_core::status::IndexerStatus {
+            sync,
+            indexed_block_id,
+            stall_reason,
+        } = value;
+
+        Self {
+            state: sync.state.into(),
+            last_error: sync.last_error,
+            indexed_block_id,
+            stall_reason: stall_reason.map(Into::into),
+        }
+    }
+}

lez/indexer/service/protocol/src/lib.rs

@@ -363,3 +363,73 @@ pub enum BedrockStatus {
     Safe,
     Finalized,
 }
+
+/// Coarse lifecycle state of the indexer's ingestion loop, so a client can tell
+/// "still catching up" apart from "something went wrong".
+#[derive(Debug, Copy, Clone, PartialEq, Eq, Hash, Serialize, Deserialize, JsonSchema)]
+pub enum IndexerSyncState {
+    /// Booted; no ingestion cycle has run yet.
+    Starting,
+    /// Streaming finalized messages toward the L1 frontier.
+    Syncing,
+    /// Drained the stream up to the last finalized block; idle until new blocks finalize.
+    CaughtUp,
+    /// The last cycle failed (e.g. the L1 node is unreachable). See `last_error`.
+    Error,
+    /// Parked on a stall reason: the validated tip is frozen awaiting a valid
+    /// continuation. See `last_error` and `stall_reason`.
+    Stalled,
+}
+
+/// Why the indexer could not apply an L2 block from the channel.
+#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize, JsonSchema)]
+pub enum BlockIngestError {
+    Deserialize(String),
+    UnexpectedBlockId {
+        expected: u64,
+        got: u64,
+    },
+    BrokenChainLink {
+        expected_prev: HashType,
+        got_prev: HashType,
+    },
+    HashMismatch {
+        computed: HashType,
+        header: HashType,
+    },
+    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.
+///
+/// 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.
+#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize, JsonSchema)]
+pub struct StallReason {
+    pub block_id: Option<u64>,
+    pub block_hash: Option<HashType>,
+    pub prev_block_hash: Option<HashType>,
+    pub l1_slot: u64,
+    pub error: BlockIngestError,
+    /// The breaking block's L2 timestamp (`None` for a deserialize break).
+    pub first_seen: Option<Timestamp>,
+    /// Number of later non-chaining blocks seen while the tip is frozen.
+    pub orphans_since: u64,
+}
+
+/// Status snapshot returned by `getStatus`: the ingestion state plus the
+/// indexed L2 tip and, when stalled, the stall diagnostics.
+#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize, JsonSchema)]
+pub struct IndexerStatus {
+    pub state: IndexerSyncState,
+    pub last_error: Option<String>,
+    pub indexed_block_id: Option<BlockId>,
+    pub stall_reason: Option<StallReason>,
+}

lez/indexer/service/rpc/src/lib.rs

@@ -1,4 +1,6 @@
-use indexer_service_protocol::{Account, AccountId, Block, BlockId, HashType, Transaction};
+use indexer_service_protocol::{
+    Account, AccountId, Block, BlockId, HashType, IndexerStatus, Transaction,
+};
 use jsonrpsee::proc_macros::rpc;
 #[cfg(feature = "server")]
 use jsonrpsee::{core::SubscriptionResult, types::ErrorObjectOwned};
@@ -69,6 +71,9 @@ pub trait Rpc {
         limit: u64,
     ) -> Result<Vec<Transaction>, ErrorObjectOwned>;
 
+    #[method(name = "getStatus")]
+    async fn get_status(&self) -> Result<IndexerStatus, ErrorObjectOwned>;
+
     // ToDo: expand healthcheck response into some kind of report
     #[method(name = "checkHealth")]
     async fn healthcheck(&self) -> Result<(), ErrorObjectOwned>;

lez/indexer/service/src/lib.rs

@@ -87,6 +87,7 @@ pub async fn run_server(
     #[cfg(not(feature = "mock-responses"))]
     let handle = {
         let service = service::IndexerService::new(config, storage_dir)
+            .await
             .context("Failed to initialize indexer service")?;
         server.start(service.into_rpc())
     };

lez/indexer/service/src/mock_service.rs

@@ -10,10 +10,10 @@ use std::{collections::HashMap, sync::Arc, time::Duration};
 
 use indexer_service_protocol::{
     Account, AccountId, BedrockStatus, Block, BlockBody, BlockHeader, BlockId, Commitment,
-    CommitmentSetDigest, Data, EncryptedAccountData, HashType, PrivacyPreservingMessage,
-    PrivacyPreservingTransaction, ProgramDeploymentMessage, ProgramDeploymentTransaction,
-    ProgramId, PublicMessage, PublicTransaction, Signature, Transaction, ValidityWindow,
-    WitnessSet,
+    CommitmentSetDigest, Data, EncryptedAccountData, HashType, IndexerStatus, IndexerSyncState,
+    PrivacyPreservingMessage, PrivacyPreservingTransaction, ProgramDeploymentMessage,
+    ProgramDeploymentTransaction, ProgramId, PublicMessage, PublicTransaction, Signature,
+    Transaction, ValidityWindow, WitnessSet,
 };
 use jsonrpsee::{
     core::{SubscriptionResult, async_trait},
@@ -325,6 +325,24 @@ impl indexer_service_rpc::RpcServer for MockIndexerService {
             .collect())
     }
 
+    async fn get_status(&self) -> Result<IndexerStatus, ErrorObjectOwned> {
+        let indexed_block_id = self
+            .state
+            .read()
+            .await
+            .blocks
+            .iter()
+            .rev()
+            .find(|block| block.bedrock_status == BedrockStatus::Finalized)
+            .map(|block| block.header.block_id);
+        Ok(IndexerStatus {
+            state: IndexerSyncState::CaughtUp,
+            last_error: None,
+            indexed_block_id,
+            stall_reason: None,
+        })
+    }
+
     async fn healthcheck(&self) -> Result<(), ErrorObjectOwned> {
         Ok(())
     }

lez/indexer/service/src/service.rs

@@ -4,7 +4,9 @@ use anyhow::{Context as _, Result, bail};
 use arc_swap::ArcSwap;
 use futures::{StreamExt as _, never::Never};
 use indexer_core::{IndexerCore, config::IndexerConfig};
-use indexer_service_protocol::{Account, AccountId, Block, BlockId, HashType, Transaction};
+use indexer_service_protocol::{
+    Account, AccountId, Block, BlockId, HashType, IndexerStatus, Transaction,
+};
 use jsonrpsee::{
     SubscriptionSink,
     core::{Serialize, SubscriptionResult, async_trait},
@@ -19,8 +21,8 @@ pub struct IndexerService {
 }
 
 impl IndexerService {
-    pub fn new(config: IndexerConfig, storage_dir: &Path) -> Result<Self> {
-        let indexer = IndexerCore::new(config, storage_dir)?;
+    pub async fn new(config: IndexerConfig, storage_dir: &Path) -> Result<Self> {
+        let indexer = IndexerCore::new(config, storage_dir).await?;
         let subscription_service = SubscriptionService::spawn_new(indexer.clone());
 
         Ok(Self {
@@ -149,6 +151,10 @@ impl indexer_service_rpc::RpcServer for IndexerService {
         Ok(tx_res)
     }
 
+    async fn get_status(&self) -> Result<IndexerStatus, ErrorObjectOwned> {
+        Ok(self.indexer.status().into())
+    }
+
     async fn healthcheck(&self) -> Result<(), ErrorObjectOwned> {
         // Checking, that indexer can calculate last state
         let _ = self

lez/sequencer/core/Cargo.toml

@@ -11,6 +11,7 @@ workspace = true
 lee.workspace = true
 lee_core.workspace = true
 common.workspace = true
+chain_consistency.workspace = true
 storage.workspace = true
 mempool.workspace = true
 logos-blockchain-zone-sdk.workspace = true

lez/sequencer/core/src/block_publisher.rs

@@ -2,14 +2,16 @@ use std::{pin::Pin, sync::Arc, time::Duration};
 
 use anyhow::{Context as _, Result, anyhow};
 use common::block::Block;
+use futures::StreamExt as _;
 use log::{info, warn};
 pub use logos_blockchain_core::mantle::ops::channel::MsgId;
 use logos_blockchain_core::mantle::ops::channel::{ChannelId, inscribe::Inscription};
 pub use logos_blockchain_key_management_system_service::keys::{Ed25519Key, ZkKey};
 pub use logos_blockchain_zone_sdk::sequencer::SequencerCheckpoint;
 use logos_blockchain_zone_sdk::{
-    CommonHttpClient,
-    adapter::NodeHttpClient,
+    CommonHttpClient, Slot, ZoneMessage,
+    adapter::{Node as _, NodeHttpClient},
+    indexer::ZoneIndexer,
     sequencer::{
         DepositInfo, Event, FinalizedOp, InscriptionInfo,
         SequencerConfig as ZoneSdkSequencerConfig, WithdrawArg, WithdrawInfo, ZoneSequencer,
@@ -63,12 +65,27 @@ pub trait BlockPublisherTrait: Clone {
     async fn publish_block(&self, block: &Block, withdrawals: Vec<WithdrawArg>) -> Result<()>;
 
     fn channel_id(&self) -> ChannelId;
+
+    /// Current channel frontier slot on the connected chain, or `None` if the
+    /// channel does not exist there. Drives the startup frontier check.
+    async fn channel_tip_slot(&self) -> Result<Option<Slot>>;
+
+    /// Finalized channel messages from `after_slot` (exclusive) up to LIB, used
+    /// for the startup consistency check and reconstruction. Pass `None` to read
+    /// from the channel's genesis.
+    async fn read_channel_after(
+        &self,
+        after_slot: Option<Slot>,
+    ) -> Result<Vec<(ZoneMessage, Slot)>>;
 }
 
 /// Real block publisher backed by zone-sdk's `ZoneSequencer`.
 #[derive(Clone)]
 pub struct ZoneSdkPublisher {
     channel_id: ChannelId,
+    /// Direct node handle retained for channel reads (startup consistency check
+    /// and reconstruction); the sequencer itself lives in the drive task.
+    node: NodeHttpClient,
     publish_tx: mpsc::Sender<(Inscription, Vec<WithdrawArg>)>,
     // Aborts the drive task when the last clone is dropped.
     _drive_task: Arc<DriveTaskGuard>,
@@ -104,7 +121,7 @@ impl BlockPublisherTrait for ZoneSdkPublisher {
         let mut sequencer = ZoneSequencer::init_with_config(
             config.channel_id,
             bedrock_signing_key,
-            node,
+            node.clone(),
             zone_sdk_config,
             initial_checkpoint,
         );
@@ -196,6 +213,7 @@ impl BlockPublisherTrait for ZoneSdkPublisher {
 
         Ok(Self {
             channel_id: config.channel_id,
+            node,
             publish_tx,
             _drive_task: Arc::new(DriveTaskGuard(drive_task)),
         })
@@ -218,6 +236,27 @@ impl BlockPublisherTrait for ZoneSdkPublisher {
     fn channel_id(&self) -> ChannelId {
         self.channel_id
     }
+
+    async fn channel_tip_slot(&self) -> Result<Option<Slot>> {
+        Ok(self
+            .node
+            .channel_state(self.channel_id)
+            .await
+            .context("Failed to read channel state")?
+            .map(|state| state.tip_slot))
+    }
+
+    async fn read_channel_after(
+        &self,
+        after_slot: Option<Slot>,
+    ) -> Result<Vec<(ZoneMessage, Slot)>> {
+        let indexer = ZoneIndexer::new(self.channel_id, self.node.clone());
+        let stream = indexer
+            .next_messages(after_slot)
+            .await
+            .context("Failed to start channel read stream")?;
+        Ok(stream.collect().await)
+    }
 }
 
 /// Deserialize inscription payload as a `Block` and return it's`block_id`.

lez/sequencer/core/src/block_store.rs

@@ -13,7 +13,7 @@ use logos_blockchain_zone_sdk::sequencer::SequencerCheckpoint;
 pub use storage::DbResult;
 use storage::sequencer::{
     RocksDBIO,
-    sequencer_cells::{PendingDepositEventRecord, WithdrawalReconciliationKey},
+    sequencer_cells::{PendingDepositEventRecord, WithdrawalReconciliationKey, ZoneAnchorRecord},
 };
 
 pub struct SequencerStore {
@@ -166,6 +166,16 @@ impl SequencerStore {
         Ok(())
     }
 
+    /// The last channel block read back and verified from Bedrock (L1 slot +
+    /// `id`/`hash`), or `None` before any block has been read from the channel.
+    pub fn get_zone_anchor(&self) -> DbResult<Option<ZoneAnchorRecord>> {
+        self.dbio.get_zone_cursor()
+    }
+
+    pub fn set_zone_anchor(&self, anchor: &ZoneAnchorRecord) -> DbResult<()> {
+        self.dbio.put_zone_cursor(anchor)
+    }
+
     pub fn get_unfulfilled_deposit_events(&self) -> DbResult<Vec<PendingDepositEventRecord>> {
         self.dbio.get_pending_deposit_events()
     }

lez/sequencer/core/src/lib.rs

@@ -2,17 +2,21 @@ use std::{path::Path, sync::Arc, time::Instant};
 
 use anyhow::{Context as _, Result, anyhow};
 use borsh::BorshDeserialize;
+use chain_consistency::{Anchor, ChainConsistency, Tip};
 use common::{
     HashType,
     block::{BedrockStatus, Block, HashableBlockData},
     transaction::{LeeTransaction, clock_invocation},
 };
 use config::{GenesisAction, SequencerConfig};
-use lee::{AccountId, PublicTransaction, public_transaction::Message};
+use lee::{AccountId, PublicTransaction, V03State, public_transaction::Message};
 use lee_core::GENESIS_BLOCK_ID;
 use log::{error, info, warn};
 use logos_blockchain_key_management_system_service::keys::{ED25519_SECRET_KEY_SIZE, Ed25519Key};
-use logos_blockchain_zone_sdk::sequencer::{DepositInfo, WithdrawArg};
+use logos_blockchain_zone_sdk::{
+    Slot, ZoneMessage,
+    sequencer::{DepositInfo, WithdrawArg},
+};
 use mempool::{MemPool, MemPoolHandle};
 #[cfg(feature = "mock")]
 pub use mock::SequencerCoreWithMockClients;
@@ -20,7 +24,7 @@ use num_bigint::BigUint;
 pub use storage::error::DbError;
 use storage::sequencer::{
     RocksDBIO,
-    sequencer_cells::{PendingDepositEventRecord, WithdrawalReconciliationKey},
+    sequencer_cells::{PendingDepositEventRecord, WithdrawalReconciliationKey, ZoneAnchorRecord},
 };
 
 use crate::{
@@ -64,6 +68,13 @@ pub struct SequencerCore<BP: BlockPublisherTrait = ZoneSdkPublisher> {
     block_publisher: BP,
 }
 
+/// Outcome of the startup verify-and-reconstruct pass.
+struct ReconstructionSummary {
+    /// Whether the channel served any blocks. Drives the fresh-start decision
+    /// between publishing our own genesis and adopting the channel's history.
+    channel_had_blocks: bool,
+}
+
 impl<BP: BlockPublisherTrait> SequencerCore<BP> {
     /// Starts the sequencer using the provided configuration.
     /// If an existing database is found, the sequencer state is loaded from it and
@@ -124,11 +135,7 @@ impl<BP: BlockPublisherTrait> SequencerCore<BP> {
             load_or_create_signing_key(&config.home.join("bedrock_signing_key"))
                 .expect("Failed to load or create bedrock signing key");
 
-        let (store, state, genesis_block) = Self::open_or_create_store(&config);
-
-        let latest_block_meta = store
-            .latest_block_meta()
-            .expect("Failed to read latest block meta from store");
+        let (mut store, mut state, genesis_block) = Self::open_or_create_store(&config);
 
         let initial_checkpoint = store
             .get_zone_checkpoint()
@@ -151,17 +158,27 @@ impl<BP: BlockPublisherTrait> SequencerCore<BP> {
         .await
         .expect("Failed to initialize Block Publisher");
 
-        // On a truly fresh start (no checkpoint persisted yet), publish the
-        // genesis block so the indexer can find the channel start. After the
-        // first publish, zone-sdk's checkpoint persistence covers further
-        // restarts.
-        if is_fresh_start {
+        // Before producing, verify our local state still belongs to the chain
+        // the channel serves and replay any channel blocks we are missing
+        // (e.g. from other sequencers).
+        let reconstruction = Self::verify_and_reconstruct(&block_publisher, &mut store, &mut state)
+            .await
+            .expect("Failed to verify/reconstruct sequencer state from Bedrock");
+
+        // Publish our genesis only when bootstrapping an empty channel. If the
+        // channel already has blocks (another sequencer bootstrapped it), we
+        // adopted them during reconstruction instead.
+        if is_fresh_start && !reconstruction.channel_had_blocks {
             block_publisher
                 .publish_block(&genesis_block, vec![])
                 .await
                 .expect("Failed to publish genesis block");
         }
 
+        let latest_block_meta = store
+            .latest_block_meta()
+            .expect("Failed to read latest block meta from store");
+
         let sequencer_core = Self {
             state,
             store,
@@ -174,6 +191,155 @@ impl<BP: BlockPublisherTrait> SequencerCore<BP> {
         (sequencer_core, mempool_handle)
     }
 
+    /// Verifies the local store still belongs to the chain the connected channel
+    /// serves and replays any finalized channel blocks missing locally into
+    /// `state`/`store`, recording each block's L1 inscription slot as the new
+    /// anchor. Fails (never parks) on any divergence.
+    async fn verify_and_reconstruct(
+        publisher: &BP,
+        store: &mut SequencerStore,
+        state: &mut V03State,
+    ) -> Result<ReconstructionSummary> {
+        let anchor_record = store
+            .get_zone_anchor()
+            .context("Failed to read zone anchor")?;
+
+        let after_slot = anchor_record
+            .and_then(|record| record.slot.checked_sub(1))
+            .map(Slot::from);
+        let messages = publisher
+            .read_channel_after(after_slot)
+            .await
+            .context("Failed to read channel history for reconstruction")?;
+
+        // Consistency check against the recorded anchor: fail fast on positive
+        // evidence the channel is a different chain (missing/behind/reinscribed).
+        if let Some(record) = anchor_record {
+            let anchor = Anchor::new(
+                Slot::from(record.slot),
+                Some((record.block_id, record.hash)),
+            );
+            let channel_tip_slot = publisher
+                .channel_tip_slot()
+                .await
+                .context("Failed to read channel tip slot")?;
+            if let ChainConsistency::Inconsistent(mismatch) =
+                chain_consistency::classify_channel(&anchor, channel_tip_slot, &messages)
+            {
+                return Err(anyhow!(
+                    "Sequencer store diverges from the Bedrock channel ({mismatch}). \
+                     Delete the sequencer storage directory or point at the correct channel."
+                ));
+            }
+        }
+
+        // Replay: apply channel blocks we are missing, verify the ones we have.
+        let mut channel_had_blocks = false;
+        for (message, slot) in &messages {
+            let ZoneMessage::Block(zone_block) = message else {
+                continue;
+            };
+            let block: Block = borsh::from_slice(&zone_block.data).map_err(|err| {
+                anyhow!(
+                    "Failed to deserialize channel block at slot {}: {err}",
+                    slot.into_inner()
+                )
+            })?;
+            channel_had_blocks = true;
+            Self::apply_reconstructed_block(store, state, &block, *slot)?;
+        }
+
+        Ok(ReconstructionSummary { channel_had_blocks })
+    }
+
+    /// Applies a single channel block during reconstruction: idempotent for
+    /// blocks we already hold (verifying their hash), a validated continuation
+    /// for new ones. Advances the persisted anchor to the block's slot.
+    fn apply_reconstructed_block(
+        store: &mut SequencerStore,
+        state: &mut V03State,
+        block: &Block,
+        slot: Slot,
+    ) -> Result<()> {
+        let tip = store
+            .latest_block_meta()
+            .context("Failed to read latest block meta")?;
+        let block_id = block.header.block_id;
+        let block_hash = block.header.hash;
+
+        let record = ZoneAnchorRecord {
+            slot: slot.into_inner(),
+            block_id,
+            hash: block_hash,
+        };
+
+        // A block at/below the tip must match what we already stored, otherwise
+        // the channel is a different chain.
+        if block_id <= tip.id {
+            match store
+                .get_block_at_id(block_id)
+                .context("Failed to read stored block")?
+            {
+                Some(stored) if stored.header.hash == block_hash => {
+                    store
+                        .set_zone_anchor(&record)
+                        .context("Failed to persist zone anchor")?;
+                    return Ok(());
+                }
+                Some(stored) => {
+                    return Err(anyhow!(
+                        "Channel block {block_id} hash {block_hash} does not match stored hash {}",
+                        stored.header.hash
+                    ));
+                }
+                None => {
+                    return Err(anyhow!(
+                        "Channel block {block_id} is at/below local tip {} but is missing locally",
+                        tip.id
+                    ));
+                }
+            }
+        }
+
+        // New continuation: validate it chains onto the tip, then apply.
+        let cc_tip = Tip {
+            block_id: tip.id,
+            hash: tip.hash,
+        };
+        chain_consistency::validate_against_tip(Some(&cc_tip), block).map_err(|err| {
+            anyhow!(
+                "Channel block {block_id} does not extend local tip {}: {err}",
+                tip.id
+            )
+        })?;
+
+        let mut scratch = state.clone();
+        chain_consistency::apply_block(block, &mut scratch)
+            .map_err(|err| anyhow!("Failed to apply channel block {block_id}: {err}"))?;
+
+        // Derive bridge bookkeeping from the block's transactions, matching the
+        // production path so the reconciliation counters stay consistent.
+        let mut deposit_event_ids = Vec::new();
+        let mut withdrawals = Vec::new();
+        for tx in &block.body.transactions {
+            if let Some(deposit_id) = extract_bridge_deposit_id(tx) {
+                deposit_event_ids.push(deposit_id);
+            }
+            if let Some(withdraw) = extract_bridge_withdraw_data(tx) {
+                withdrawals.push(withdraw_event_reconciliation_key(&withdraw.outputs)?);
+            }
+        }
+
+        store
+            .update(block, &deposit_event_ids, withdrawals, &scratch)
+            .context("Failed to persist reconstructed block")?;
+        *state = scratch;
+        store
+            .set_zone_anchor(&record)
+            .context("Failed to persist zone anchor")?;
+        Ok(())
+    }
+
     fn on_checkpoint(dbio: Arc<RocksDBIO>) -> block_publisher::CheckpointSink {
         Box::new(move |cp| {
             let bytes = match serde_json::to_vec(&cp) {

lez/sequencer/core/src/mock.rs

@@ -4,7 +4,7 @@ use anyhow::Result;
 use common::block::Block;
 use logos_blockchain_core::mantle::ops::channel::ChannelId;
 use logos_blockchain_key_management_system_service::keys::Ed25519Key;
-use logos_blockchain_zone_sdk::sequencer::WithdrawArg;
+use logos_blockchain_zone_sdk::{Slot, ZoneMessage, sequencer::WithdrawArg};
 
 use crate::{
     block_publisher::{
@@ -19,6 +19,28 @@ pub type SequencerCoreWithMockClients = crate::SequencerCore<MockBlockPublisher>
 #[derive(Clone)]
 pub struct MockBlockPublisher {
     channel_id: ChannelId,
+    /// Canned channel frontier returned by [`Self::channel_tip_slot`].
+    tip_slot: Option<Slot>,
+    /// Canned finalized channel history returned by [`Self::read_channel_after`].
+    messages: Vec<(ZoneMessage, Slot)>,
+}
+
+impl MockBlockPublisher {
+    /// Builds a mock publisher backed by a canned channel, for reconstruction
+    /// and consistency tests. The default (via [`BlockPublisherTrait::new`])
+    /// serves an empty channel.
+    #[must_use]
+    pub const fn with_canned_channel(
+        channel_id: ChannelId,
+        tip_slot: Option<Slot>,
+        messages: Vec<(ZoneMessage, Slot)>,
+    ) -> Self {
+        Self {
+            channel_id,
+            tip_slot,
+            messages,
+        }
+    }
 }
 
 impl BlockPublisherTrait for MockBlockPublisher {
@@ -34,6 +56,8 @@ impl BlockPublisherTrait for MockBlockPublisher {
     ) -> Result<Self> {
         Ok(Self {
             channel_id: config.channel_id,
+            tip_slot: None,
+            messages: Vec::new(),
         })
     }
 
@@ -48,4 +72,21 @@ impl BlockPublisherTrait for MockBlockPublisher {
     fn channel_id(&self) -> ChannelId {
         self.channel_id
     }
+
+    async fn channel_tip_slot(&self) -> Result<Option<Slot>> {
+        Ok(self.tip_slot)
+    }
+
+    async fn read_channel_after(
+        &self,
+        after_slot: Option<Slot>,
+    ) -> Result<Vec<(ZoneMessage, Slot)>> {
+        // Mirror `next_messages`: `after_slot` is exclusive.
+        Ok(self
+            .messages
+            .iter()
+            .filter(|(_, slot)| after_slot.is_none_or(|after| *slot > after))
+            .cloned()
+            .collect())
+    }
 }

lez/sequencer/core/src/tests.rs

@@ -1232,3 +1232,153 @@ fn pda_mechanism_with_pinata_token_program() {
         expected_winner_token_holding_post
     );
 }
+
+mod reconstruction {
+    use common::block::Block;
+    use logos_blockchain_core::mantle::ops::channel::{MsgId, inscribe::Inscription};
+    use logos_blockchain_zone_sdk::{Slot, ZoneBlock, ZoneMessage};
+    use storage::sequencer::sequencer_cells::ZoneAnchorRecord;
+
+    use super::*;
+    use crate::{SequencerCore, block_store::SequencerStore, mock::MockBlockPublisher};
+
+    fn block_to_channel_message(block: &Block, slot: u64) -> (ZoneMessage, Slot) {
+        let bytes = borsh::to_vec(block).expect("serialize block");
+        let message = ZoneMessage::Block(ZoneBlock {
+            id: MsgId::from([0_u8; 32]),
+            data: Inscription::try_from(bytes.as_slice()).expect("inscription"),
+        });
+        (message, Slot::from(slot))
+    }
+
+    /// Collects a sequencer's whole chain (genesis..=tip) into a canned channel,
+    /// one block per slot at `slot_step` spacing.
+    fn channel_from_store(store: &SequencerStore, slot_step: u64) -> Vec<(ZoneMessage, Slot)> {
+        let genesis_id = store.genesis_id();
+        let tip_id = store.latest_block_meta().expect("tip").id;
+        (genesis_id..=tip_id)
+            .enumerate()
+            .map(|(index, id)| {
+                let block = store.get_block_at_id(id).expect("read").expect("present");
+                block_to_channel_message(&block, (index as u64 + 1) * slot_step)
+            })
+            .collect()
+    }
+
+    #[tokio::test]
+    async fn reconstructs_missing_channel_blocks_into_fresh_store() {
+        // Sequencer A produces a few blocks; treat its chain as the channel.
+        let config_a = setup_sequencer_config();
+        let (mut seq_a, _handle_a) =
+            SequencerCoreWithMockClients::start_from_config(config_a.clone()).await;
+        seq_a.produce_new_block().await.unwrap();
+        seq_a.produce_new_block().await.unwrap();
+        let tip_a = seq_a.block_store().latest_block_meta().unwrap();
+
+        let messages = channel_from_store(seq_a.block_store(), 10);
+        let tip_slot = messages.last().unwrap().1;
+        let channel_id = config_a.bedrock_config.channel_id;
+
+        // Sequencer B starts from a fresh store and reconstructs A's chain.
+        let config_b = setup_sequencer_config();
+        let (mut store_b, mut state_b, _genesis_b) =
+            SequencerCore::<MockBlockPublisher>::open_or_create_store(&config_b);
+        let mock_b = MockBlockPublisher::with_canned_channel(channel_id, Some(tip_slot), messages);
+
+        let summary = SequencerCore::<MockBlockPublisher>::verify_and_reconstruct(
+            &mock_b,
+            &mut store_b,
+            &mut state_b,
+        )
+        .await
+        .expect("reconstruct");
+        assert!(summary.channel_had_blocks);
+
+        let tip_b = store_b.latest_block_meta().unwrap();
+        assert_eq!(tip_b.id, tip_a.id);
+        assert_eq!(tip_b.hash, tip_a.hash);
+
+        // State matches: initial account balances agree with sequencer A.
+        for account in initial_public_user_accounts() {
+            assert_eq!(
+                state_b.get_account_by_id(account.account_id).balance,
+                seq_a.state().get_account_by_id(account.account_id).balance,
+            );
+        }
+
+        let anchor = store_b.get_zone_anchor().unwrap().expect("anchor recorded");
+        assert_eq!(anchor.block_id, tip_a.id);
+        assert_eq!(anchor.slot, tip_slot.into_inner());
+
+        // Re-running is idempotent: everything is already applied, no error.
+        let summary = SequencerCore::<MockBlockPublisher>::verify_and_reconstruct(
+            &mock_b,
+            &mut store_b,
+            &mut state_b,
+        )
+        .await
+        .expect("reconstruct idempotent");
+        assert!(summary.channel_had_blocks);
+        assert_eq!(store_b.latest_block_meta().unwrap().id, tip_a.id);
+    }
+
+    #[tokio::test]
+    async fn fails_when_channel_serves_a_divergent_block() {
+        let config = setup_sequencer_config();
+        let (mut store, mut state, _genesis) =
+            SequencerCore::<MockBlockPublisher>::open_or_create_store(&config);
+
+        // Anchor on the local genesis at some slot.
+        let genesis_id = store.genesis_id();
+        let genesis = store.get_block_at_id(genesis_id).unwrap().unwrap();
+        let anchor_slot = 100_u64;
+        store
+            .set_zone_anchor(&ZoneAnchorRecord {
+                slot: anchor_slot,
+                block_id: genesis_id,
+                hash: genesis.header.hash,
+            })
+            .unwrap();
+
+        // The channel serves a different block at the anchor id/slot.
+        let mut tampered = genesis.clone();
+        tampered.header.hash = HashType([9_u8; 32]);
+        let messages = vec![block_to_channel_message(&tampered, anchor_slot)];
+        let mock = MockBlockPublisher::with_canned_channel(
+            config.bedrock_config.channel_id,
+            Some(Slot::from(anchor_slot)),
+            messages,
+        );
+
+        let result = SequencerCore::<MockBlockPublisher>::verify_and_reconstruct(
+            &mock, &mut store, &mut state,
+        )
+        .await;
+        assert!(result.is_err(), "divergent channel must abort startup");
+    }
+
+    #[tokio::test]
+    async fn fails_when_channel_is_missing() {
+        let config = setup_sequencer_config();
+        let (mut store, mut state, _genesis) =
+            SequencerCore::<MockBlockPublisher>::open_or_create_store(&config);
+        let genesis_id = store.genesis_id();
+        let genesis = store.get_block_at_id(genesis_id).unwrap().unwrap();
+        store
+            .set_zone_anchor(&ZoneAnchorRecord {
+                slot: 100,
+                block_id: genesis_id,
+                hash: genesis.header.hash,
+            })
+            .unwrap();
+
+        // Anchor present, but the channel does not exist on the connected chain.
+        let mock =
+            MockBlockPublisher::with_canned_channel(config.bedrock_config.channel_id, None, vec![]);
+        let result = SequencerCore::<MockBlockPublisher>::verify_and_reconstruct(
+            &mock, &mut store, &mut state,
+        )
+        .await;
+        assert!(result.is_err(), "missing channel must abort startup");
+    }
+}

lez/storage/src/indexer/indexer_cells.rs

@@ -8,8 +8,8 @@ use crate::{
     indexer::{
         ACC_NUM_CELL_NAME, BLOCK_HASH_CELL_NAME, BREAKPOINT_CELL_NAME, CF_ACC_META,
         CF_BREAKPOINT_NAME, CF_HASH_TO_ID, CF_TX_TO_ID, DB_META_LAST_BREAKPOINT_ID,
-        DB_META_LAST_OBSERVED_L1_LIB_HEADER_ID_IN_DB_KEY, DB_META_ZONE_SDK_INDEXER_CURSOR_KEY,
-        TX_HASH_CELL_NAME,
+        DB_META_LAST_OBSERVED_L1_LIB_HEADER_ID_IN_DB_KEY, DB_META_STALL_REASON_KEY,
+        DB_META_ZONE_SDK_INDEXER_CURSOR_KEY, TX_HASH_CELL_NAME,
     },
 };
 
@@ -247,6 +247,40 @@ impl SimpleWritableCell for ZoneSdkIndexerCursorCellRef<'_> {
     }
 }
 
+/// Opaque JSON bytes for the indexer's persisted `Option<StallReason>`.
+#[derive(BorshDeserialize)]
+pub struct StallReasonCellOwned(pub Vec<u8>);
+
+impl SimpleStorableCell for StallReasonCellOwned {
+    type KeyParams = ();
+
+    const CELL_NAME: &'static str = DB_META_STALL_REASON_KEY;
+    const CF_NAME: &'static str = CF_META_NAME;
+}
+
+impl SimpleReadableCell for StallReasonCellOwned {}
+
+#[derive(BorshSerialize)]
+pub struct StallReasonCellRef<'bytes>(pub &'bytes [u8]);
+
+impl SimpleStorableCell for StallReasonCellRef<'_> {
+    type KeyParams = ();
+
+    const CELL_NAME: &'static str = DB_META_STALL_REASON_KEY;
+    const CF_NAME: &'static str = CF_META_NAME;
+}
+
+impl SimpleWritableCell for StallReasonCellRef<'_> {
+    fn value_constructor(&self) -> DbResult<Vec<u8>> {
+        borsh::to_vec(&self).map_err(|err| {
+            DbError::borsh_cast_message(
+                err,
+                Some("Failed to serialize stall reason cell".to_owned()),
+            )
+        })
+    }
+}
+
 #[cfg(test)]
 mod uniform_tests {
     use crate::{

lez/storage/src/indexer/mod.rs

@@ -24,6 +24,8 @@ pub const DB_META_LAST_OBSERVED_L1_LIB_HEADER_ID_IN_DB_KEY: &str =
 pub const DB_META_LAST_BREAKPOINT_ID: &str = "last_breakpoint_id";
 /// Key base for storing the zone-sdk indexer cursor (opaque bytes).
 pub const DB_META_ZONE_SDK_INDEXER_CURSOR_KEY: &str = "zone_sdk_indexer_cursor";
+/// Key base for storing the persisted `Option<StallReason>` diagnostic record (opaque JSON bytes).
+pub const DB_META_STALL_REASON_KEY: &str = "stall_reason";
 
 /// Cell name for a breakpoint.
 pub const BREAKPOINT_CELL_NAME: &str = "breakpoint";
@@ -211,20 +213,7 @@ fn apply_block_transactions(mut block: Block, state: &mut V03State) -> DbResult<
                 })?;
         } else {
             transaction
-                .transaction_stateless_check()
-                .map_err(|err| {
-                    DbError::db_interaction_error(format!(
-                        "transaction pre check failed with err {err:?}"
-                    ))
-                })?
-                // FIXME: HOT FIX (testnet v0.2): does not check for system account updates due to
-                // sequencer-generated deposit tx'es;
-                // CHANGE ME back to `execute_check_on_state` when the indexer can authenticate deposit transactions
-                .execute_without_system_accounts_check_on_state(
-                    state,
-                    block.header.block_id,
-                    block.header.timestamp,
-                )
+                .execute_on_state(state, block.header.block_id, block.header.timestamp)
                 .map_err(|err| {
                     DbError::db_interaction_error(format!(
                         "transaction execution failed with err {err:?}"

lez/storage/src/indexer/read_once.rs

@@ -4,7 +4,8 @@ use crate::{
     cells::shared_cells::{BlockCell, FirstBlockCell, FirstBlockSetCell, LastBlockCell},
     indexer::indexer_cells::{
         AccNumTxCell, BlockHashToBlockIdMapCell, BreakpointCellOwned, LastBreakpointIdCell,
-        LastObservedL1LibHeaderCell, TxHashToBlockIdMapCell, ZoneSdkIndexerCursorCellOwned,
+        LastObservedL1LibHeaderCell, StallReasonCellOwned, TxHashToBlockIdMapCell,
+        ZoneSdkIndexerCursorCellOwned,
     },
 };
 
@@ -73,4 +74,8 @@ impl RocksDBIO {
             .get_opt::<ZoneSdkIndexerCursorCellOwned>(())?
             .map(|cell| cell.0))
     }
+
+    pub fn get_stall_reason_bytes(&self) -> DbResult<Option<Vec<u8>>> {
+        Ok(self.get_opt::<StallReasonCellOwned>(())?.map(|cell| cell.0))
+    }
 }

lez/storage/src/indexer/write_non_atomic.rs

@@ -3,7 +3,7 @@ use crate::{
     DBIO as _,
     cells::shared_cells::{FirstBlockSetCell, LastBlockCell},
     indexer::indexer_cells::{
-        BreakpointCellRef, LastBreakpointIdCell, LastObservedL1LibHeaderCell,
+        BreakpointCellRef, LastBreakpointIdCell, LastObservedL1LibHeaderCell, StallReasonCellRef,
         ZoneSdkIndexerCursorCellRef,
     },
 };
@@ -35,6 +35,10 @@ impl RocksDBIO {
         self.put(&ZoneSdkIndexerCursorCellRef(bytes), ())
     }
 
+    pub fn put_stall_reason_bytes(&self, bytes: &[u8]) -> DbResult<()> {
+        self.put(&StallReasonCellRef(bytes), ())
+    }
+
     // State
 
     pub fn put_breakpoint(&self, br_id: u64, breakpoint: &V03State) -> DbResult<()> {

lez/storage/src/sequencer/mod.rs

@@ -20,7 +20,7 @@ use crate::{
         LEEStateCellOwned, LEEStateCellRef, LastFinalizedBlockIdCell, LatestBlockMetaCellOwned,
         LatestBlockMetaCellRef, PendingDepositEventRecord, PendingDepositEventsCellOwned,
         PendingDepositEventsCellRef, UnseenWithdrawCountCell, WithdrawalReconciliationKey,
-        ZoneSdkCheckpointCellOwned, ZoneSdkCheckpointCellRef,
+        ZoneAnchorRecord, ZoneCursorCell, ZoneSdkCheckpointCellOwned, ZoneSdkCheckpointCellRef,
     },
 };
 
@@ -32,6 +32,10 @@ pub const DB_META_LAST_FINALIZED_BLOCK_ID: &str = "last_finalized_block_id";
 pub const DB_META_LATEST_BLOCK_META_KEY: &str = "latest_block_meta";
 /// Key base for storing the zone-sdk sequencer checkpoint (opaque bytes).
 pub const DB_META_ZONE_SDK_CHECKPOINT_KEY: &str = "zone_sdk_checkpoint";
+/// Key base for storing the last channel block read back and verified from
+/// Bedrock (its L1 slot + `id`/`hash`) — the anchor for the startup
+/// consistency check and the resume point for reconstruction.
+pub const DB_META_ZONE_CURSOR_KEY: &str = "zone_cursor";
 /// Key base for storing queued deposit events that were not yet
 /// fulfilled on L2.
 pub const DB_META_PENDING_DEPOSIT_EVENTS_KEY: &str = "pending_deposit_events";
@@ -246,6 +250,14 @@ impl RocksDBIO {
         self.put(&ZoneSdkCheckpointCellRef(bytes), ())
     }
 
+    pub fn get_zone_cursor(&self) -> DbResult<Option<ZoneAnchorRecord>> {
+        Ok(self.get_opt::<ZoneCursorCell>(())?.map(|cell| cell.0))
+    }
+
+    pub fn put_zone_cursor(&self, anchor: &ZoneAnchorRecord) -> DbResult<()> {
+        self.put(&ZoneCursorCell(*anchor), ())
+    }
+
     pub fn get_pending_deposit_events(&self) -> DbResult<Vec<PendingDepositEventRecord>> {
         Ok(self
             .get_opt::<PendingDepositEventsCellOwned>(())?

lez/storage/src/sequencer/sequencer_cells.rs

@@ -9,7 +9,8 @@ use crate::{
     sequencer::{
         CF_LEE_STATE_NAME, DB_LEE_STATE_KEY, DB_META_LAST_FINALIZED_BLOCK_ID,
         DB_META_LATEST_BLOCK_META_KEY, DB_META_PENDING_DEPOSIT_EVENTS_KEY,
-        DB_META_UNSEEN_WITHDRAW_COUNT_KEY, DB_META_ZONE_SDK_CHECKPOINT_KEY,
+        DB_META_UNSEEN_WITHDRAW_COUNT_KEY, DB_META_ZONE_CURSOR_KEY,
+        DB_META_ZONE_SDK_CHECKPOINT_KEY,
     },
 };
 
@@ -132,6 +133,39 @@ impl SimpleWritableCell for ZoneSdkCheckpointCellRef<'_> {
     }
 }
 
+/// The last channel block read back and verified from Bedrock.
+///
+/// Holds its L1 inscription `slot` plus the block's `id`/`hash`, and serves as
+/// both the anchor for the startup consistency check and the resume point for
+/// reconstruction. `slot` is stored as a raw `u64` because the zone-sdk `Slot`
+/// does not derive borsh; the caller converts to/from `Slot`.
+#[derive(Debug, Clone, Copy, BorshSerialize, BorshDeserialize)]
+pub struct ZoneAnchorRecord {
+    pub slot: u64,
+    pub block_id: u64,
+    pub hash: HashType,
+}
+
+#[derive(Debug, BorshSerialize, BorshDeserialize)]
+pub struct ZoneCursorCell(pub ZoneAnchorRecord);
+
+impl SimpleStorableCell for ZoneCursorCell {
+    type KeyParams = ();
+
+    const CELL_NAME: &'static str = DB_META_ZONE_CURSOR_KEY;
+    const CF_NAME: &'static str = CF_META_NAME;
+}
+
+impl SimpleReadableCell for ZoneCursorCell {}
+
+impl SimpleWritableCell for ZoneCursorCell {
+    fn value_constructor(&self) -> DbResult<Vec<u8>> {
+        borsh::to_vec(&self).map_err(|err| {
+            DbError::borsh_cast_message(err, Some("Failed to serialize zone cursor".to_owned()))
+        })
+    }
+}
+
 #[derive(Debug, Clone, PartialEq, Eq, BorshSerialize, BorshDeserialize)]
 pub struct PendingDepositEventRecord {
     pub deposit_op_id: HashType,

test_fixtures/src/config.rs

@@ -176,6 +176,7 @@ pub fn indexer_config(bedrock_addr: SocketAddr) -> Result<IndexerConfig> {
             auth: None,
         },
         channel_id: bedrock_channel_id(),
+        allow_chain_reset: false,
     })
 }