fix(indexer): keep track of L1 slot of last inscription for correct tracking

lez/indexer/core/src/block_store.rs

@@ -136,6 +136,13 @@ impl IndexerStore {
         Ok(())
     }
 
+    /// The L1 inscription slot of the validated tip, written atomically with it
+    /// by [`Self::accept_block`]. `None` on a cold store or one written before
+    /// the slot was recorded.
+    pub fn get_tip_slot(&self) -> Result<Option<Slot>> {
+        Ok(self.dbio.get_meta_tip_slot_in_db()?.map(Slot::from))
+    }
+
     pub fn get_stall_reason(&self) -> Result<Option<StallReason>> {
         let Some(bytes) = self.dbio.get_stall_reason_bytes()? else {
             return Ok(None);
@@ -254,7 +261,7 @@ impl IndexerStore {
         let mut stored = block.clone();
         stored.bedrock_status = BedrockStatus::Finalized;
         self.dbio
-            .put_block(&stored, [0_u8; 32])
+            .put_block(&stored, [0_u8; 32], l1_slot.into_inner())
             .context("Failed to persist accepted block")?;
 
         // Commit in-memory state (infallible) only after the DB write succeeded.
@@ -668,6 +675,46 @@ mod accept_tests {
         );
     }
 
+    #[tokio::test]
+    async fn accept_block_records_tip_inscription_slot() {
+        let dir = tempfile::tempdir().expect("tempdir");
+        let store = IndexerStore::open_db(dir.path()).expect("open store");
+
+        assert_eq!(store.get_tip_slot().expect("get"), None);
+
+        let genesis = produce_dummy_block(1, None, vec![]);
+        store
+            .accept_block(&genesis, Slot::from(1_000))
+            .await
+            .expect("accept");
+        assert_eq!(store.get_tip_slot().expect("get"), Some(Slot::from(1_000)));
+
+        let block2 = produce_dummy_block(2, Some(genesis.header.hash), vec![]);
+        store
+            .accept_block(&block2, Slot::from(1_005))
+            .await
+            .expect("accept");
+        assert_eq!(store.get_tip_slot().expect("get"), Some(Slot::from(1_005)));
+
+        // A parked block freezes the tip, so its slot must not advance either.
+        let bad = produce_dummy_block(4, Some(block2.header.hash), vec![]);
+        assert!(matches!(
+            store.accept_block(&bad, Slot::from(1_010)).await.unwrap(),
+            AcceptOutcome::Parked(_)
+        ));
+        assert_eq!(store.get_tip_slot().expect("get"), Some(Slot::from(1_005)));
+
+        // Neither must a re-delivered old block move it.
+        assert!(matches!(
+            store
+                .accept_block(&genesis, Slot::from(1_015))
+                .await
+                .unwrap(),
+            AcceptOutcome::AlreadyApplied
+        ));
+        assert_eq!(store.get_tip_slot().expect("get"), Some(Slot::from(1_005)));
+    }
+
     #[tokio::test]
     async fn redelivered_tip_block_is_idempotent_not_parked() {
         use testnet_initial_state::initial_pub_accounts_private_keys;

lez/indexer/core/src/chain_consistency.rs

@@ -173,46 +173,59 @@ enum AnchorProbe {
     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.
+    /// Verifies whether the channel still serves the same chain the store was built from.
+    /// This may change frequently during development where we reset the chain from time to
+    /// time in devnet/testnet, but we do not expect [`ChainConsistency::Inconsistent`] in
+    /// production.
     ///
-    /// 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`].
+    /// To compare the chains, we use an [`Anchor`] block that is either the parked L2 block
+    /// while stalled, or the tip L2 block at its own inscription L1 slot.
     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)),
-            }
+        let Some(anchor) = self.get_startup_anchor()? else {
+            // empty or cold store: nothing to compare
+            return Ok(ChainConsistency::Inconclusive);
         };
 
         self.verify_chain_at_anchor(&anchor).await
     }
 
+    /// Builds the anchor for the startup check.
+    ///
+    /// - If stalled, returns the recorded _parked_ block
+    /// - If not stalled, returns the validated tip at its _own_ inscription slot.
+    /// - If the store is empty, returns `None`.
+    fn get_startup_anchor(&self) -> Result<Option<Anchor>> {
+        if let Some(stall) = self.store.get_stall_reason()? {
+            return Ok(Some(Anchor {
+                slot: stall.l1_slot,
+                block: stall.block_id.zip(stall.block_hash),
+            }));
+        }
+
+        // not stalled, so anchor on the tip at its own inscription slot
+        let Some(slot) = self.store.get_tip_slot()?.or(self.store.get_zone_cursor()?) else {
+            return Ok(None);
+        };
+        let Some(tip_id) = self.store.get_last_block_id()? else {
+            return Ok(None);
+        };
+        let Some(tip) = self.store.get_block_at_id(tip_id)? else {
+            return Ok(None);
+        };
+        Ok(Some(Anchor {
+            slot,
+            block: Some((tip_id, tip.header.hash)),
+        }))
+    }
+
     /// 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`.
+    /// 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) => {
@@ -388,6 +401,41 @@ mod tests {
         ));
     }
 
+    #[tokio::test]
+    async fn startup_anchor_prefers_tip_slot_over_lagging_cursor() {
+        // Cursor persist failures are warn-only, so the read cursor can lag the
+        // tip by several blocks. The anchor must pair the tip with its own
+        // inscription slot; pairing it with the stale cursor would make the scan
+        // misread the chain's intermediate blocks as re-inscriptions.
+        let dir = tempfile::tempdir().expect("tempdir");
+        let core = unreachable_core(dir.path());
+
+        let genesis = common::test_utils::produce_dummy_block(1, None, vec![]);
+        core.store
+            .accept_block(&genesis, Slot::from(1_000))
+            .await
+            .expect("accept");
+        let block2 = common::test_utils::produce_dummy_block(2, Some(genesis.header.hash), vec![]);
+        core.store
+            .accept_block(&block2, Slot::from(1_005))
+            .await
+            .expect("accept");
+        let block3 = common::test_utils::produce_dummy_block(3, Some(block2.header.hash), vec![]);
+        core.store
+            .accept_block(&block3, Slot::from(1_010))
+            .await
+            .expect("accept");
+
+        // Cursor last persisted at the genesis slot: two blocks behind the tip.
+        core.store
+            .set_zone_cursor(&Slot::from(1_000))
+            .expect("set cursor");
+
+        let anchor = core.get_startup_anchor().expect("anchor").expect("present");
+        assert_eq!(anchor.slot, Slot::from(1_010));
+        assert_eq!(anchor.block, Some((3, block3.header.hash)));
+    }
+
     #[test]
     fn probe_finds_anchor_block_at_slot() {
         let tip = test_block(5, 42);

lez/storage/src/indexer/indexer_cells.rs

@@ -9,7 +9,7 @@ use crate::{
         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_STALL_REASON_KEY,
-        DB_META_ZONE_SDK_INDEXER_CURSOR_KEY, TX_HASH_CELL_NAME,
+        DB_META_TIP_SLOT_KEY, DB_META_ZONE_SDK_INDEXER_CURSOR_KEY, TX_HASH_CELL_NAME,
     },
 };
 
@@ -212,6 +212,27 @@ impl SimpleWritableCell for AccNumTxCell {
     }
 }
 
+/// The L1 inscription slot of the tip block, written atomically with the tip.
+#[derive(Debug, BorshSerialize, BorshDeserialize)]
+pub struct TipSlotCell(pub u64);
+
+impl SimpleStorableCell for TipSlotCell {
+    type KeyParams = ();
+
+    const CELL_NAME: &'static str = DB_META_TIP_SLOT_KEY;
+    const CF_NAME: &'static str = CF_META_NAME;
+}
+
+impl SimpleReadableCell for TipSlotCell {}
+
+impl SimpleWritableCell for TipSlotCell {
+    fn value_constructor(&self) -> DbResult<Vec<u8>> {
+        borsh::to_vec(&self).map_err(|err| {
+            DbError::borsh_cast_message(err, Some("Failed to serialize tip slot".to_owned()))
+        })
+    }
+}
+
 /// Opaque bytes for the zone-sdk indexer cursor `Option<(MsgId, Slot)>`.
 /// The caller serializes via `serde_json` (neither type derives borsh).
 #[derive(BorshDeserialize)]

lez/storage/src/indexer/mod.rs

@@ -26,6 +26,8 @@ pub const DB_META_LAST_BREAKPOINT_ID: &str = "last_breakpoint_id";
 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";
+/// Key base for storing the L1 inscription slot of the tip block.
+pub const DB_META_TIP_SLOT_KEY: &str = "tip_slot";
 
 /// Cell name for a breakpoint.
 pub const BREAKPOINT_CELL_NAME: &str = "breakpoint";

lez/storage/src/indexer/read_once.rs

@@ -4,7 +4,7 @@ use crate::{
     cells::shared_cells::{BlockCell, FirstBlockCell, FirstBlockSetCell, LastBlockCell},
     indexer::indexer_cells::{
         AccNumTxCell, BlockHashToBlockIdMapCell, BreakpointCellOwned, LastBreakpointIdCell,
-        LastObservedL1LibHeaderCell, StallReasonCellOwned, TxHashToBlockIdMapCell,
+        LastObservedL1LibHeaderCell, StallReasonCellOwned, TipSlotCell, TxHashToBlockIdMapCell,
         ZoneSdkIndexerCursorCellOwned,
     },
 };
@@ -37,6 +37,11 @@ impl RocksDBIO {
             .map(|opt| opt.map(|cell| cell.0))
     }
 
+    pub fn get_meta_tip_slot_in_db(&self) -> DbResult<Option<u64>> {
+        self.get_opt::<TipSlotCell>(())
+            .map(|opt| opt.map(|cell| cell.0))
+    }
+
     // Block
 
     pub fn get_block(&self, block_id: u64) -> DbResult<Option<Block>> {

lez/storage/src/indexer/tests.rs

@@ -87,7 +87,7 @@ fn one_block_insertion() {
     let dbio = RocksDBIO::open_or_create(temdir_path, &initial_state()).unwrap();
 
     let genesis_block = genesis_block();
-    dbio.put_block(&genesis_block, [0; 32]).unwrap();
+    dbio.put_block(&genesis_block, [0; 32], 0).unwrap();
 
     let prev_hash = genesis_block.header.hash;
     let from = acc1();
@@ -98,7 +98,7 @@ fn one_block_insertion() {
         common::test_utils::create_transaction_native_token_transfer(from, 0, to, 1, &sign_key);
     let block = produce_dummy_block(2, Some(prev_hash), vec![transfer_tx]);
 
-    dbio.put_block(&block, [1; 32]).unwrap();
+    dbio.put_block(&block, [1; 32], 0).unwrap();
 
     let last_id = dbio.get_meta_last_block_id_in_db().unwrap().unwrap();
     let first_id = dbio.get_meta_first_block_id_in_db().unwrap();
@@ -130,6 +130,26 @@ fn one_block_insertion() {
     );
 }
 
+#[test]
+fn put_block_records_tip_inscription_slot() {
+    let temp_dir = tempdir().unwrap();
+    let dbio = RocksDBIO::open_or_create(temp_dir.path(), &initial_state()).unwrap();
+
+    assert_eq!(dbio.get_meta_tip_slot_in_db().unwrap(), None);
+
+    let genesis_block = genesis_block();
+    dbio.put_block(&genesis_block, [0; 32], 1_000).unwrap();
+    assert_eq!(dbio.get_meta_tip_slot_in_db().unwrap(), Some(1_000));
+
+    let block = produce_dummy_block(2, Some(genesis_block.header.hash), vec![]);
+    dbio.put_block(&block, [1; 32], 1_005).unwrap();
+    assert_eq!(dbio.get_meta_tip_slot_in_db().unwrap(), Some(1_005));
+
+    // Re-inserting a block at/below the tip must not move the tip slot.
+    dbio.put_block(&genesis_block, [0; 32], 1_010).unwrap();
+    assert_eq!(dbio.get_meta_tip_slot_in_db().unwrap(), Some(1_005));
+}
+
 #[test]
 fn new_breakpoint() {
     let temp_dir = tempdir().unwrap();
@@ -155,7 +175,7 @@ fn new_breakpoint() {
             &sign_key,
         );
         let block = produce_dummy_block(i.into(), prev_hash, vec![transfer_tx]);
-        dbio.put_block(&block, [i; 32]).unwrap();
+        dbio.put_block(&block, [i; 32], 0).unwrap();
     }
 
     let last_id = dbio.get_meta_last_block_id_in_db().unwrap().unwrap();
@@ -211,7 +231,7 @@ fn simple_maps() {
 
     let control_hash1 = block.header.hash;
 
-    dbio.put_block(&block, [1; 32]).unwrap();
+    dbio.put_block(&block, [1; 32], 0).unwrap();
 
     let last_id = dbio.get_meta_last_block_id_in_db().unwrap().unwrap();
     let last_block = dbio.get_block(last_id).unwrap().unwrap();
@@ -223,7 +243,7 @@ fn simple_maps() {
 
     let control_hash2 = block.header.hash;
 
-    dbio.put_block(&block, [2; 32]).unwrap();
+    dbio.put_block(&block, [2; 32], 0).unwrap();
 
     let last_id = dbio.get_meta_last_block_id_in_db().unwrap().unwrap();
     let last_block = dbio.get_block(last_id).unwrap().unwrap();
@@ -235,7 +255,7 @@ fn simple_maps() {
     let control_tx_hash1 = transfer_tx.hash();
 
     let block = produce_dummy_block(3, Some(prev_hash), vec![transfer_tx]);
-    dbio.put_block(&block, [3; 32]).unwrap();
+    dbio.put_block(&block, [3; 32], 0).unwrap();
 
     let last_id = dbio.get_meta_last_block_id_in_db().unwrap().unwrap();
     let last_block = dbio.get_block(last_id).unwrap().unwrap();
@@ -247,7 +267,7 @@ fn simple_maps() {
     let control_tx_hash2 = transfer_tx.hash();
 
     let block = produce_dummy_block(4, Some(prev_hash), vec![transfer_tx]);
-    dbio.put_block(&block, [4; 32]).unwrap();
+    dbio.put_block(&block, [4; 32], 0).unwrap();
 
     let control_block_id1 = dbio.get_block_id_by_hash(control_hash1.0).unwrap().unwrap();
     let control_block_id2 = dbio.get_block_id_by_hash(control_hash2.0).unwrap().unwrap();
@@ -284,7 +304,7 @@ fn block_batch() {
     let block = produce_dummy_block(1, None, vec![transfer_tx]);
 
     block_res.push(block.clone());
-    dbio.put_block(&block, [1; 32]).unwrap();
+    dbio.put_block(&block, [1; 32], 0).unwrap();
 
     let last_id = dbio.get_meta_last_block_id_in_db().unwrap().unwrap();
     let last_block = dbio.get_block(last_id).unwrap().unwrap();
@@ -295,7 +315,7 @@ fn block_batch() {
     let block = produce_dummy_block(2, Some(prev_hash), vec![transfer_tx]);
 
     block_res.push(block.clone());
-    dbio.put_block(&block, [2; 32]).unwrap();
+    dbio.put_block(&block, [2; 32], 0).unwrap();
 
     let last_id = dbio.get_meta_last_block_id_in_db().unwrap().unwrap();
     let last_block = dbio.get_block(last_id).unwrap().unwrap();
@@ -306,7 +326,7 @@ fn block_batch() {
 
     let block = produce_dummy_block(3, Some(prev_hash), vec![transfer_tx]);
     block_res.push(block.clone());
-    dbio.put_block(&block, [3; 32]).unwrap();
+    dbio.put_block(&block, [3; 32], 0).unwrap();
 
     let last_id = dbio.get_meta_last_block_id_in_db().unwrap().unwrap();
     let last_block = dbio.get_block(last_id).unwrap().unwrap();
@@ -317,7 +337,7 @@ fn block_batch() {
 
     let block = produce_dummy_block(4, Some(prev_hash), vec![transfer_tx]);
     block_res.push(block.clone());
-    dbio.put_block(&block, [4; 32]).unwrap();
+    dbio.put_block(&block, [4; 32], 0).unwrap();
 
     let block_hashes_mem: Vec<[u8; 32]> =
         block_res.into_iter().map(|bl| bl.header.hash.0).collect();
@@ -376,7 +396,7 @@ fn account_map() {
 
     let block = produce_dummy_block(1, None, vec![transfer_tx1, transfer_tx2]);
 
-    dbio.put_block(&block, [1; 32]).unwrap();
+    dbio.put_block(&block, [1; 32], 0).unwrap();
 
     let last_id = dbio.get_meta_last_block_id_in_db().unwrap().unwrap();
     let last_block = dbio.get_block(last_id).unwrap().unwrap();
@@ -391,7 +411,7 @@ fn account_map() {
 
     let block = produce_dummy_block(2, Some(prev_hash), vec![transfer_tx1, transfer_tx2]);
 
-    dbio.put_block(&block, [2; 32]).unwrap();
+    dbio.put_block(&block, [2; 32], 0).unwrap();
 
     let last_id = dbio.get_meta_last_block_id_in_db().unwrap().unwrap();
     let last_block = dbio.get_block(last_id).unwrap().unwrap();
@@ -406,7 +426,7 @@ fn account_map() {
 
     let block = produce_dummy_block(3, Some(prev_hash), vec![transfer_tx1, transfer_tx2]);
 
-    dbio.put_block(&block, [3; 32]).unwrap();
+    dbio.put_block(&block, [3; 32], 0).unwrap();
 
     let last_id = dbio.get_meta_last_block_id_in_db().unwrap().unwrap();
     let last_block = dbio.get_block(last_id).unwrap().unwrap();
@@ -418,7 +438,7 @@ fn account_map() {
 
     let block = produce_dummy_block(4, Some(prev_hash), vec![transfer_tx]);
 
-    dbio.put_block(&block, [4; 32]).unwrap();
+    dbio.put_block(&block, [4; 32], 0).unwrap();
 
     let acc1_tx = dbio.get_acc_transactions(*acc1().value(), 0, 7).unwrap();
     let acc1_tx_hashes: Vec<[u8; 32]> = acc1_tx.into_iter().map(|tx| tx.hash().0).collect();

lez/storage/src/indexer/write_atomic.rs

@@ -8,7 +8,7 @@ use crate::{
     cells::shared_cells::{FirstBlockCell, FirstBlockSetCell, LastBlockCell},
     indexer::indexer_cells::{
         AccNumTxCell, BlockHashToBlockIdMapCell, LastBreakpointIdCell, LastObservedL1LibHeaderCell,
-        TxHashToBlockIdMapCell,
+        TipSlotCell, TxHashToBlockIdMapCell,
     },
 };
 
@@ -139,13 +139,20 @@ impl RocksDBIO {
         self.put_batch(&LastBreakpointIdCell(br_id), (), write_batch)
     }
 
+    pub fn put_meta_tip_slot_in_db_batch(
+        &self,
+        l1_slot: u64,
+        write_batch: &mut WriteBatch,
+    ) -> DbResult<()> {
+        self.put_batch(&TipSlotCell(l1_slot), (), write_batch)
+    }
+
     pub fn put_meta_is_first_block_set_batch(&self, write_batch: &mut WriteBatch) -> DbResult<()> {
         self.put_batch(&FirstBlockSetCell(true), (), write_batch)
     }
 
-    // Block
-
-    pub fn put_block(&self, block: &Block, l1_lib_header: [u8; 32]) -> DbResult<()> {
+    /// Put a block atomically (via [`WriteBatch`]) along with its L1 header and `Slot`.
+    pub fn put_block(&self, block: &Block, l1_lib_header: [u8; 32], l1_slot: u64) -> DbResult<()> {
         let cf_block = self.block_column();
         let last_curr_block = self.get_meta_last_block_id_in_db()?.unwrap_or(0);
         let mut write_batch = WriteBatch::default();
@@ -163,6 +170,7 @@ impl RocksDBIO {
         if block.header.block_id > last_curr_block {
             self.put_meta_last_block_in_db_batch(block.header.block_id, &mut write_batch)?;
             self.put_meta_last_observed_l1_lib_header_in_db_batch(l1_lib_header, &mut write_batch)?;
+            self.put_meta_tip_slot_in_db_batch(l1_slot, &mut write_batch)?;
         }
         if last_curr_block == 0 {
             self.put_meta_first_block_in_db_batch(block, &mut write_batch)?;