fix(indexer): allow recovery from parked-and-stalled store (w.r.t Copilot review)

refactor(indexer): move chain-consistency code within a dedicated file

lez/indexer/core/src/chain_consistency.rs

@@ -0,0 +1,500 @@
+//! Startup check that the local store still belongs to the chain the
+//! connected channel serves.
+
+use anyhow::Result;
+use common::{HashType, block::Block};
+use futures::StreamExt as _;
+use log::warn;
+use logos_blockchain_zone_sdk::{Slot, ZoneMessage, adapter::Node as _};
+
+use crate::IndexerCore;
+
+/// 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 the indexer'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 the ingest loop tries to apply and parks.
+    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: (u64, HashType),
+        channel: (u64, 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: (u64, 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.
+struct Anchor {
+    slot: Slot,
+    /// The anchor block's `(id, hash)`
+    ///
+    /// `None` when parked on an undeserializable inscription (no header was recorded).
+    block: Option<(u64, HashType)>,
+}
+
+impl Anchor {
+    /// Probes a channel message read at/after the anchor slot.
+    /// See [`IndexerCore::verify_chain_at_anchor`].
+    pub 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,
+}
+
+#[expect(
+    clippy::multiple_inherent_impl,
+    reason = "split for clarity & isolation of relevant code"
+)]
+impl IndexerCore {
+    /// 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.
+    ///
+    /// Anchors on a block the same chain must still serve at a known slot: 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). See [`Self::verify_chain_at_anchor`].
+    pub(crate) async fn verify_chain_consistency(&self) -> Result<ChainConsistency> {
+        let anchor = if let Some(stall) = self.store.get_stall_reason()? {
+            Anchor {
+                slot: stall.l1_slot,
+                block: 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 {
+                slot: cursor,
+                block: Some((tip_id, tip.header.hash)),
+            }
+        };
+
+        self.verify_chain_at_anchor(&anchor).await
+    }
+
+    /// 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 `Inconsistent`; absence of data stays `Inconclusive`.
+    async fn verify_chain_at_anchor(&self, anchor: &Anchor) -> Result<ChainConsistency> {
+        match self.node.channel_state(self.config.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 scan = async {
+            let stream = self
+                .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)
+            }
+        }
+    }
+}
+
+/// 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 std::time::Duration;
+
+    use common::block::HashableBlockData;
+    use logos_blockchain_core::mantle::ops::channel::{MsgId, inscribe::Inscription};
+    use logos_blockchain_zone_sdk::ZoneBlock;
+
+    use super::*;
+    use crate::{
+        BlockIngestError,
+        block_store::AcceptOutcome,
+        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"))
+    }
+
+    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 {
+            slot,
+            block: Some((block.header.block_id, block.header.hash)),
+        }
+    }
+
+    #[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
+        ));
+    }
+
+    #[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 {
+            slot: Slot::from(1_000),
+            block: 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/indexer/core/src/lib.rs

@@ -1,8 +1,8 @@
 use std::{path::Path, sync::Arc};
 
-use anyhow::{Context as _, Result};
+use anyhow::Result;
 use arc_swap::ArcSwap;
-use common::{HashType, block::Block};
+use common::block::Block;
 // TODO: Remove after testnet
 use futures::StreamExt as _;
 pub use ingest_error::BlockIngestError;
@@ -14,55 +14,22 @@ pub use stall_reason::StallReason;
 
 use crate::{
     block_store::{AcceptOutcome, IndexerStore},
+    chain_consistency::ChainConsistency,
     config::IndexerConfig,
     status::{IndexerStatus, IndexerSyncStatus},
 };
 pub mod block_store;
+pub mod chain_consistency;
 pub mod config;
 pub mod ingest_error;
 pub mod stall_reason;
 pub mod status;
 
-/// Result of comparing the indexer's stored tip against the channel.
-enum ChainConsistency {
-    /// Channel serves the same block at our tip's id.
-    Consistent,
-    /// We could not determine the outcome due to one of:
-    ///
-    /// - cold store
-    /// - the channel served no comparable block, we hold no block at that id
-    /// - 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 the ingest loop tries to apply and parks.
-    Inconclusive,
-    /// The channel serves a different block at one of our ids.
-    ///
-    /// Details in [`BlockMismatch`], and impl's Display trait.
-    Inconsistent(BlockMismatch),
-}
-
-/// The differing pair behind a [`ChainConsistency::Inconsistent`]: our stored
-/// block vs. the block the channel serves at the same id, as `(block_id, hash)`.
-struct BlockMismatch {
-    ours: (u64, HashType),
-    channel: (u64, HashType),
-}
-
-impl std::fmt::Display for BlockMismatch {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        let Self { ours, channel } = self;
-        write!(
-            f,
-            "stored block {} {} != channel block {} {}",
-            ours.0, ours.1, channel.0, channel.1
-        )
-    }
-}
-
 #[derive(Clone)]
 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`.
@@ -115,94 +82,17 @@ 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.
-    ///
-    /// Compares the first block the channel serves at/after the read cursor against our
-    /// stored block of the same id. Conclusive only while caught up (cursor at the tip);
-    /// once parked the cursor runs ahead, so the compared id is one we never stored and
-    /// the result is `Inconclusive` — a genuine reset is then caught by the ingest loop.
-    async fn verify_chain_consistency(&self) -> Result<ChainConsistency> {
-        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(channel) = self.fetch_channel_block_from(cursor).await? else {
-            // if the channel is empty, we have nothing to compare
-            return Ok(ChainConsistency::Inconclusive);
-        };
-
-        let Some(ours) = self.store.get_block_at_id(channel.header.block_id)? else {
-            // if we do not have the block the channel serves at that id, we cannot compare
-            return Ok(ChainConsistency::Inconclusive);
-        };
-
-        // compare the block w.r.t hashes
-        let comparison_result = if ours.header.hash == channel.header.hash {
-            ChainConsistency::Consistent
-        } else {
-            ChainConsistency::Inconsistent(BlockMismatch {
-                ours: (ours.header.block_id, ours.header.hash),
-                channel: (channel.header.block_id, channel.header.hash),
-            })
-        };
-
-        Ok(comparison_result)
-    }
-
-    /// Reads the first block the channel serves at/after the tip's slot.
-    ///
-    /// `None` means we could not read a comparable block.
-    async fn fetch_channel_block_from(&self, cursor: Slot) -> Result<Option<Block>> {
-        const TIP_FETCH_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(60);
-
-        // `next_messages` is exclusive, so `cursor - 1` includes the tip's own slot.
-        let Some(from_slot) = cursor.into_inner().checked_sub(1) else {
-            // Slot-0 cursor is degenerate (inscriptions live at wall-clock slots);
-            // so we bail rather than let `next_messages(None)` do a from-genesis scan.
-            return Ok(None);
-        };
-        let fetch = async {
-            let stream = self
-                .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 {
-                if let ZoneMessage::Block(zone_block) = msg {
-                    let block: Block = borsh::from_slice(&zone_block.data)
-                        .context("Failed to deserialize channel block")?;
-                    return Ok::<Option<Block>, anyhow::Error>(Some(block));
-                }
-            }
-            Ok(None)
-        };
-
-        match tokio::time::timeout(TIP_FETCH_TIMEOUT, fetch).await {
-            Ok(Ok(found)) => Ok(found),
-            Ok(Err(err)) => {
-                warn!("Failed to read channel tip for the consistency check; proceeding: {err:#}");
-                Ok(None)
-            }
-            Err(_elapsed) => {
-                warn!("Timed out reading channel tip for the consistency check; proceeding");
-                Ok(None)
-            }
-        }
-    }
-
     /// 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
@@ -382,32 +272,3 @@ impl IndexerCore {
         }
     }
 }
-
-#[cfg(test)]
-mod chain_identity_tests {
-    use std::time::Duration;
-
-    use super::{ChainConsistency, IndexerCore};
-    use crate::config::{ChannelId, ClientConfig, IndexerConfig};
-
-    #[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 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,
-        };
-        let core = IndexerCore::open(config, dir.path()).expect("open core");
-        assert!(matches!(
-            core.verify_chain_consistency().await.expect("verify"),
-            ChainConsistency::Inconclusive
-        ));
-    }
-}