diff --git a/lez/indexer/core/src/block_store.rs b/lez/indexer/core/src/block_store.rs index 0fd34c36..2a2bfbce 100644 --- a/lez/indexer/core/src/block_store.rs +++ b/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> { + Ok(self.dbio.get_meta_tip_slot_in_db()?.map(Slot::from)) + } + pub fn get_stall_reason(&self) -> Result> { 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; diff --git a/lez/indexer/core/src/chain_consistency.rs b/lez/indexer/core/src/chain_consistency.rs index 4ffa244c..00f28822 100644 --- a/lez/indexer/core/src/chain_consistency.rs +++ b/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 { - 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> { + 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 { 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); diff --git a/lez/storage/src/indexer/indexer_cells.rs b/lez/storage/src/indexer/indexer_cells.rs index d71a2ed6..e00e6faf 100644 --- a/lez/storage/src/indexer/indexer_cells.rs +++ b/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> { + 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)] diff --git a/lez/storage/src/indexer/mod.rs b/lez/storage/src/indexer/mod.rs index 10c80e63..0226bd48 100644 --- a/lez/storage/src/indexer/mod.rs +++ b/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` 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"; diff --git a/lez/storage/src/indexer/read_once.rs b/lez/storage/src/indexer/read_once.rs index 591f8405..83f58d58 100644 --- a/lez/storage/src/indexer/read_once.rs +++ b/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> { + self.get_opt::(()) + .map(|opt| opt.map(|cell| cell.0)) + } + // Block pub fn get_block(&self, block_id: u64) -> DbResult> { diff --git a/lez/storage/src/indexer/tests.rs b/lez/storage/src/indexer/tests.rs index 44df23d5..5cd0607b 100644 --- a/lez/storage/src/indexer/tests.rs +++ b/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(); diff --git a/lez/storage/src/indexer/write_atomic.rs b/lez/storage/src/indexer/write_atomic.rs index ec28f8ec..bed1c53b 100644 --- a/lez/storage/src/indexer/write_atomic.rs +++ b/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)?;