diff --git a/Cargo.lock b/Cargo.lock index 3ccb055c..b3373975 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3863,6 +3863,7 @@ dependencies = [ "storage", "tempfile", "testnet_initial_state", + "thiserror 2.0.18", "tokio", "url", ] @@ -3910,6 +3911,7 @@ dependencies = [ "base64", "common", "hex", + "indexer_core", "lee", "lee_core", "schemars 1.2.1", @@ -9458,6 +9460,7 @@ dependencies = [ "borsh", "common", "lee", + "log", "programs", "rocksdb", "system_accounts", @@ -9757,6 +9760,7 @@ dependencies = [ "tempfile", "testcontainers", "tokio", + "tokio-util", "url", "wallet", ] diff --git a/integration_tests/tests/indexer_stall.rs b/integration_tests/tests/indexer_stall.rs new file mode 100644 index 00000000..ae1b8b6a --- /dev/null +++ b/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(()) +} diff --git a/lez/common/src/block.rs b/lez/common/src/block.rs index 6e956f9f..a8149e7c 100644 --- a/lez/common/src/block.rs +++ b/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(&self, serializer: S) -> Result { 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::(&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); + } } diff --git a/lez/common/src/transaction.rs b/lez/common/src/transaction.rs index b5aee648..b5f0bc87 100644 --- a/lez/common/src/transaction.rs +++ b/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, diff --git a/lez/configs/docker-all-in-one/indexer_config.json b/lez/configs/docker-all-in-one/indexer_config.json index c1ff65b0..5791f64a 100644 --- a/lez/configs/docker-all-in-one/indexer_config.json +++ b/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 } diff --git a/lez/indexer/core/Cargo.toml b/lez/indexer/core/Cargo.toml index b49f6fee..772b3536 100644 --- a/lez/indexer/core/Cargo.toml +++ b/lez/indexer/core/Cargo.toml @@ -29,6 +29,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 diff --git a/lez/indexer/core/src/block_store.rs b/lez/indexer/core/src/block_store.rs index 4d583686..4b26dbe4 100644 --- a/lez/indexer/core/src/block_store.rs +++ b/lez/indexer/core/src/block_store.rs @@ -2,17 +2,40 @@ use std::{path::Path, sync::Arc}; use anyhow::{Context as _, Result}; use common::{ - block::{BedrockStatus, Block}, + HashType, + block::{BedrockStatus, Block, BlockHeader}, transaction::{LeeTransaction, clock_invocation}, }; -use lee::{Account, AccountId, V03State}; +use lee::{Account, AccountId, GENESIS_BLOCK_ID, 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::{ingest_error::BlockIngestError, stall_reason::StallReason}; + +struct Tip { + block_id: u64, + hash: HashType, +} + +/// Outcome of feeding a parsed L2 block to the validated tip. +pub enum AcceptOutcome { + /// Chained and applied; tip and L1 read cursor both advance. + Applied, + /// A duplicate re-delivery of the current tip. Just L2 advances. + AlreadyApplied, + /// Did not chain or failed to apply; tip stays frozen, stall recorded. + Parked(BlockIngestError), + /// Chained but failed to apply, possibly transiently + /// ([`BlockIngestError::is_retryable`]); nothing recorded, tip and state + /// untouched. The caller retries and parks via + /// [`IndexerStore::record_stall`] once it gives up. + RetryableFailure(BlockIngestError), +} + #[derive(Clone)] pub struct IndexerStore { dbio: Arc, @@ -118,6 +141,36 @@ 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); + }; + let stall: Option = + serde_json::from_slice(&bytes).context("Failed to deserialize stored stall reason")?; + Ok(stall) + } + + pub fn set_stall_reason(&self, stall: &Option) -> 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 +192,251 @@ 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> { + 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. + /// Retryable apply failures return `RetryableFailure` without recording a stall + /// or touching state; other failures record the stall and return `Parked`. + pub async fn accept_block(&self, block: &Block, l1_slot: Slot) -> Result { + 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) = 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) = apply_block_to_scratch(block, &mut scratch) { + if err.is_retryable() { + return Ok(AcceptOutcome::RetryableFailure(err)); + } + 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], l1_slot.into_inner(), &scratch) + .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) + } +} + +/// Checks that `block` is the valid continuation of `tip`: hash integrity, +/// then block-id continuity, then `prev_block_hash` linkage. A `None` tip +/// (cold store) expects the genesis block. +fn validate_against_tip(tip: Option<&Tip>, block: &Block) -> Result<(), BlockIngestError> { + let computed = block.recompute_hash(); + if computed != block.header.hash { + return Err(BlockIngestError::HashMismatch { + computed, + header: block.header.hash, + }); + } + + match tip { + None => { + if block.header.block_id != GENESIS_BLOCK_ID { + return Err(BlockIngestError::UnexpectedBlockId { + expected: GENESIS_BLOCK_ID, + got: block.header.block_id, + }); + } + } + Some(tip) => { + let expected = tip + .block_id + .checked_add(1) + .expect("block id should not overflow"); + if block.header.block_id != expected { + return Err(BlockIngestError::UnexpectedBlockId { + expected, + got: block.header.block_id, + }); + } + if block.header.prev_block_hash != tip.hash { + return Err(BlockIngestError::BrokenChainLink { + expected_prev: tip.hash, + got_prev: block.header.prev_block_hash, + }); + } + } + } + Ok(()) +} + +/// Applies a block's transactions to `state`, mapping every failure to a +/// [`BlockIngestError`] so the caller can park rather than crash. Operates on a +/// scratch state; the caller commits only on `Ok`. +fn apply_block_to_scratch(block: &Block, state: &mut V03State) -> Result<(), BlockIngestError> { + let (clock_tx, user_txs) = block + .body + .transactions + .split_last() + .ok_or(BlockIngestError::EmptyBlock)?; + + let expected_clock = LeeTransaction::Public(clock_invocation(block.header.timestamp)); + if *clock_tx != expected_clock { + return Err(BlockIngestError::InvalidClockTransaction); + } + + let is_genesis = block.header.block_id == GENESIS_BLOCK_ID; + for (tx_index, transaction) in user_txs.iter().enumerate() { + let state_transition = |err: anyhow::Error| BlockIngestError::StateTransition { + tx_index: tx_index.try_into().expect("tx index fits in u64"), + reason: format!("{err:#}"), + }; + if is_genesis { + let LeeTransaction::Public(public_tx) = transaction else { + return Err(BlockIngestError::NonPublicGenesisTransaction); + }; + state + .transition_from_public_transaction( + public_tx, + block.header.block_id, + block.header.timestamp, + ) + .map_err(|err| state_transition(err.into()))?; + } else { + transaction + .clone() + .execute_on_state(state, block.header.block_id, block.header.timestamp) + .map_err(|err| state_transition(err.into()))?; + } + } + + let LeeTransaction::Public(clock_public_tx) = clock_tx else { + return Err(BlockIngestError::InvalidClockTransaction); + }; + state + .transition_from_public_transaction( + clock_public_tx, + block.header.block_id, + block.header.timestamp, + ) + .map_err(|err| BlockIngestError::StateTransition { + tx_index: user_txs.len().try_into().expect("tx index fits in u64"), + reason: format!("{:#}", anyhow::Error::from(err)), + })?; + + Ok(()) +} + +#[cfg(test)] +mod stall_reason_tests { + use common::HashType; + + use super::*; + use crate::{ingest_error::BlockIngestError, stall_reason::StallReason}; + + #[tokio::test] + async fn stall_reason_roundtrips_and_clears() { + 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, - ) -> 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 +449,474 @@ 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 common::{HashType, block::HashableBlockData, test_utils::produce_dummy_block}; + + use super::*; + use crate::ingest_error::BlockIngestError; + + fn signing_key() -> lee::PrivateKey { + lee::PrivateKey::try_new([7_u8; 32]).expect("valid key") + } + + // 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 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; + + 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" + ); + } + + #[tokio::test] + async fn accept_block_snapshots_state_at_breakpoint_interval() { + 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![]); + assert!(matches!( + store.accept_block(&genesis, Slot::from(0)).await.unwrap(), + AcceptOutcome::Applied + )); + let mut prev_hash = genesis.header.hash; + + // Blocks 2..=101: one transfer of 1 each; block 100 crosses the interval. + for i in 0..100_u64 { + let tx = common::test_utils::create_transaction_native_token_transfer( + from, + i.into(), + to, + 1, + &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 + )); + } + + // Snapshot at block 100 = genesis + 99 transfers, written with the block. + let bp1 = store.dbio.get_breakpoint(1).expect("breakpoint 1 present"); + assert_eq!(bp1.get_account_by_id(from).balance, 10000 - 99); + + // The #605 restart: reopening past the boundary must work. + drop(store); + let reopened = IndexerStore::open_db(dir.path()).expect("reopen"); + assert_eq!(reopened.last_block().unwrap(), Some(101)); + } + + #[tokio::test] + async fn transient_apply_failure_returns_retryable_failure_without_stall() { + 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"); + + // Overdraft: rejected during execution → StateTransition → retryable. + let tx = common::test_utils::create_transaction_native_token_transfer( + from, + 0, + to, + 1_000_000_000, + &sign_key, + ); + let block = produce_dummy_block(2, Some(genesis.header.hash), vec![tx]); + let outcome = store.accept_block(&block, Slot::from(0)).await.unwrap(); + + assert!(matches!( + outcome, + AcceptOutcome::RetryableFailure(BlockIngestError::StateTransition { .. }) + )); + assert!( + store.get_stall_reason().unwrap().is_none(), + "retryable failure must not persist a stall" + ); + assert_eq!(store.get_last_block_id().unwrap(), Some(1), "tip frozen"); } } diff --git a/lez/indexer/core/src/chain_consistency.rs b/lez/indexer/core/src/chain_consistency.rs new file mode 100644 index 00000000..00f28822 --- /dev/null +++ b/lez/indexer/core/src/chain_consistency.rs @@ -0,0 +1,548 @@ +//! 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::(&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 { + /// 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. + /// + /// 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 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`. + async fn verify_chain_at_anchor(&self, anchor: &Anchor) -> Result { + 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) -> Option { + 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 + )); + } + + #[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); + 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()); + } +} diff --git a/lez/indexer/core/src/config.rs b/lez/indexer/core/src/config.rs index cb7f3dfe..c069d5cf 100644 --- a/lez/indexer/core/src/config.rs +++ b/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 { diff --git a/lez/indexer/core/src/ingest_error.rs b/lez/indexer/core/src/ingest_error.rs new file mode 100644 index 00000000..299013d7 --- /dev/null +++ b/lez/indexer/core/src/ingest_error.rs @@ -0,0 +1,80 @@ +use common::HashType; +use serde::{Deserialize, Serialize}; + +/// Why the indexer could not apply an L2 block from the channel. +/// +/// Persisted in `RocksDB`, so every variant must have the following +/// traits: `Clone + Serialize + Deserialize`. +#[derive(Debug, Clone, Serialize, Deserialize, thiserror::Error)] +pub enum BlockIngestError { + #[error("Failed to deserialize L2 block: {0}")] + /// Here we store the error string that is derived from [`borsh::from_slice`]'s [`Err`]. + Deserialize(String), + #[error("Unexpected block id: expected {expected}, got {got}")] + UnexpectedBlockId { expected: u64, got: u64 }, + #[error("Broken chain link: expected prev {expected_prev}, got {got_prev}")] + BrokenChainLink { + expected_prev: HashType, + got_prev: HashType, + }, + #[error("Block hash mismatch: computed {computed}, header {header}")] + HashMismatch { + computed: HashType, + header: HashType, + }, + #[error("Block has no transactions")] + EmptyBlock, + #[error("Last transaction must be the public clock invocation for the block timestamp")] + InvalidClockTransaction, + #[error("Genesis block must contain only public transactions")] + NonPublicGenesisTransaction, + #[error("State transition failed at transaction {tx_index}: {reason}")] + StateTransition { + /// Index of the failing transaction within the block body. + tx_index: u64, + /// Reason string from `lee::Error` to `anyhow::Error` to `{:#}`. + /// + /// This is required because `lee::Error` is not `Clone + Serialize + Deserialize`, so we + /// cannot store it directly. + reason: String, + }, +} + +impl BlockIngestError { + /// Whether the failure may be transient rather than a property of the block. + /// + /// FIXME: `StateTransition` is too coarse — its `reason` string mixes genuine + /// state-transition rejections with infra failures (risc0 executor teardown, + /// storage errors). Once it carries a structured cause, narrow this so only + /// infra failures retry. + #[must_use] + pub const fn is_retryable(&self) -> bool { + matches!(self, Self::StateTransition { .. }) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn serializes_and_round_trips_externally_tagged() { + let err = BlockIngestError::UnexpectedBlockId { + expected: 5, + got: 7, + }; + let value = serde_json::to_value(&err).expect("serialize"); + assert_eq!( + value, + serde_json::json!({ "UnexpectedBlockId": { "expected": 5, "got": 7 } }) + ); + let back: BlockIngestError = serde_json::from_value(value).expect("deserialize"); + assert!(matches!( + back, + BlockIngestError::UnexpectedBlockId { + expected: 5, + got: 7 + } + )); + } +} diff --git a/lez/indexer/core/src/lib.rs b/lez/indexer/core/src/lib.rs index 0d595fc0..719caa08 100644 --- a/lez/indexer/core/src/lib.rs +++ b/lez/indexer/core/src/lib.rs @@ -3,27 +3,38 @@ use std::{path::Path, sync::Arc}; use anyhow::Result; use arc_swap::ArcSwap; use common::block::Block; -// ToDo: Remove after testnet +// TODO: Remove after testnet use futures::StreamExt as _; +pub use ingest_error::BlockIngestError; 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, }; +use retry::ApplyRetryGate; +pub use stall_reason::StallReason; use crate::{ - block_store::IndexerStore, + 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; +mod retry; +pub mod stall_reason; pub mod status; +/// Consecutive failed apply attempts of the same block before parking. +const APPLY_RETRY_LIMIT: u32 = 3; + #[derive(Clone)] pub struct IndexerCore { pub zone_indexer: Arc>, + /// 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 +42,42 @@ pub struct IndexerCore { } impl IndexerCore { - pub fn new(config: IndexerConfig, storage_dir: &Path) -> Result { + /// 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 { + 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 { // 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,10 +87,11 @@ 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())), @@ -58,10 +105,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 +134,41 @@ 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) { + *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. + /// + /// Returns `false` if the stall could not be recorded durably; the caller + /// must then hold the cursor and retry instead of advancing past the slot. + fn park_undeserializable(&self, slot: Slot, error: std::io::Error) -> bool { + 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())) + { + error!("Failed to record stall reason: {err:#}"); + self.set_status(IndexerSyncStatus::error(format!("store error: {err:#}"))); + return false; + } + self.set_status(IndexerSyncStatus::stalled(format!( + "failed to deserialize L2 block: {reason}" + ))); + true + } + pub fn subscribe_parse_block_stream(&self) -> impl futures::Stream> + '_ { let poll_interval = self.config.consensus_info_polling_interval; let initial_cursor = self @@ -79,6 +178,7 @@ impl IndexerCore { async_stream::stream! { let mut cursor = initial_cursor; + let mut retry_gate = ApplyRetryGate::new(); if cursor.is_some() { info!("Resuming indexer from cursor {cursor:?}"); @@ -90,8 +190,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 +200,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,46 +211,117 @@ 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) => { + // The stall must be durable before the cursor moves. + if !self.park_undeserializable(slot, error) { + had_cycle_error = true; + break; } + // 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) => { + retry_gate.reset(); + 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); + } + Ok(AcceptOutcome::RetryableFailure(ingest_err)) => { + let attempts = retry_gate.register_failure(block.header.block_id); + if attempts >= APPLY_RETRY_LIMIT { + error!( + "Parked at block {} after {attempts} failed apply attempts: {ingest_err}", + block.header.block_id + ); + // The stall must be durable before the cursor moves. + if let Err(err) = self.store.record_stall( + Some(&block.header), + slot, + ingest_err.clone(), + ) { + error!( + "Failed to record stall reason for block {}: {err:#}", + block.header.block_id + ); + self.set_status(IndexerSyncStatus::error(format!( + "store error: {err:#}" + ))); + had_cycle_error = true; + break; + } + self.set_status(IndexerSyncStatus::stalled(ingest_err.to_string())); + self.advance_cursor(&mut cursor, slot); + retry_gate.reset(); + } else { + error!( + "Failed to apply block {} (attempt {attempts}/{APPLY_RETRY_LIMIT}), will retry: {ingest_err}", + block.header.block_id + ); + self.set_status(IndexerSyncStatus::error(format!( + "apply failed, retrying: {ingest_err}" + ))); + had_cycle_error = true; + break; + } + } + 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; } } diff --git a/lez/indexer/core/src/retry.rs b/lez/indexer/core/src/retry.rs new file mode 100644 index 00000000..d7c5f5a2 --- /dev/null +++ b/lez/indexer/core/src/retry.rs @@ -0,0 +1,64 @@ +//! Retry gate for possibly-transient block-apply failures. + +/// Counts consecutive apply failures per block id so the ingest loop can +/// retry before parking. +/// +/// A failure streak is keyed to one block id: a failure of a different block +/// starts a fresh streak. Reset only on a genuinely applied block — the +/// `AlreadyApplied` replay of the prefix after a retry cycle must not clear +/// the failing block's streak. +pub struct ApplyRetryGate { + failing: Option<(u64, u32)>, +} + +impl ApplyRetryGate { + #[must_use] + pub const fn new() -> Self { + Self { failing: None } + } + + /// Registers a failed apply of `block_id`; returns its consecutive + /// attempt count. + pub const fn register_failure(&mut self, block_id: u64) -> u32 { + let attempts = match self.failing { + Some((id, attempts)) if id == block_id => attempts.saturating_add(1), + _ => 1, + }; + self.failing = Some((block_id, attempts)); + attempts + } + + /// Clears the streak; call when a block actually applies. + pub const fn reset(&mut self) { + self.failing = None; + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn counts_consecutive_failures_of_same_block() { + let mut gate = ApplyRetryGate::new(); + assert_eq!(gate.register_failure(7), 1); + assert_eq!(gate.register_failure(7), 2); + assert_eq!(gate.register_failure(7), 3); + } + + #[test] + fn different_block_starts_fresh_streak() { + let mut gate = ApplyRetryGate::new(); + gate.register_failure(7); + gate.register_failure(7); + assert_eq!(gate.register_failure(8), 1); + } + + #[test] + fn reset_clears_streak() { + let mut gate = ApplyRetryGate::new(); + gate.register_failure(7); + gate.reset(); + assert_eq!(gate.register_failure(7), 1); + } +} diff --git a/lez/indexer/core/src/stall_reason.rs b/lez/indexer/core/src/stall_reason.rs new file mode 100644 index 00000000..c5118fcc --- /dev/null +++ b/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, + pub block_hash: Option, + pub prev_block_hash: Option, + pub l1_slot: Slot, + pub error: BlockIngestError, + pub first_seen: Option, + /// 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, +} diff --git a/lez/indexer/core/src/status.rs b/lez/indexer/core/src/status.rs index 1193e124..a483fde6 100644 --- a/lez/indexer/core/src/status.rs +++ b/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, @@ -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, + pub stall_reason: Option, } #[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)); + } } diff --git a/lez/indexer/ffi/indexer_ffi.h b/lez/indexer/ffi/indexer_ffi.h index 8347ad3c..26857af7 100644 --- a/lez/indexer/ffi/indexer_ffi.h +++ b/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 * diff --git a/lez/indexer/ffi/src/api/lifecycle.rs b/lez/indexer/ffi/src/api/lifecycle.rs index f668f3ee..8a1eafaf 100644 --- a/lez/indexer/ffi/src/api/lifecycle.rs +++ b/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 diff --git a/lez/indexer/ffi/src/api/query.rs b/lez/indexer/ffi/src/api/query.rs index 1943f6d4..dff027fa 100644 --- a/lez/indexer/ffi/src/api/query.rs +++ b/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 /// diff --git a/lez/indexer/service/configs/debug/indexer_config.json b/lez/indexer/service/configs/debug/indexer_config.json index 85227700..be5cf353 100644 --- a/lez/indexer/service/configs/debug/indexer_config.json +++ b/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 } diff --git a/lez/indexer/service/configs/docker/indexer_config.json b/lez/indexer/service/configs/docker/indexer_config.json index f083ca27..ce28af0b 100644 --- a/lez/indexer/service/configs/docker/indexer_config.json +++ b/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 } diff --git a/lez/indexer/service/protocol/Cargo.toml b/lez/indexer/service/protocol/Cargo.toml index 5a4176f5..b3e8f65c 100644 --- a/lez/indexer/service/protocol/Cargo.toml +++ b/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"] diff --git a/lez/indexer/service/protocol/src/convert.rs b/lez/indexer/service/protocol/src/convert.rs index cd0bff7e..55c4dc6c 100644 --- a/lez/indexer/service/protocol/src/convert.rs +++ b/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 for lee_core::program::ValidityWindow { value.0.try_into() } } + +// ============================================================================ +// Indexer status conversions +// ============================================================================ + +impl From 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 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 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 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), + } + } +} diff --git a/lez/indexer/service/protocol/src/lib.rs b/lez/indexer/service/protocol/src/lib.rs index a670dee6..e17d539b 100644 --- a/lez/indexer/service/protocol/src/lib.rs +++ b/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, + pub block_hash: Option, + pub prev_block_hash: Option, + pub l1_slot: u64, + pub error: BlockIngestError, + /// The breaking block's L2 timestamp (`None` for a deserialize break). + pub first_seen: Option, + /// 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, + pub indexed_block_id: Option, + pub stall_reason: Option, +} diff --git a/lez/indexer/service/rpc/src/lib.rs b/lez/indexer/service/rpc/src/lib.rs index 5763fe82..8ea807eb 100644 --- a/lez/indexer/service/rpc/src/lib.rs +++ b/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, ErrorObjectOwned>; + #[method(name = "getStatus")] + async fn get_status(&self) -> Result; + // ToDo: expand healthcheck response into some kind of report #[method(name = "checkHealth")] async fn healthcheck(&self) -> Result<(), ErrorObjectOwned>; diff --git a/lez/indexer/service/src/lib.rs b/lez/indexer/service/src/lib.rs index b1c57163..aa142b38 100644 --- a/lez/indexer/service/src/lib.rs +++ b/lez/indexer/service/src/lib.rs @@ -5,6 +5,7 @@ pub use indexer_core::config::*; use indexer_service_rpc::RpcServer as _; use jsonrpsee::server::{Server, ServerHandle}; use log::{error, info}; +use tokio_util::sync::CancellationToken; pub mod service; @@ -69,9 +70,10 @@ pub async fn run_server( config: IndexerConfig, storage_dir: &Path, port: u16, + shutdown: CancellationToken, ) -> Result { #[cfg(feature = "mock-responses")] - let _ = (config, storage_dir); + let _ = (config, storage_dir, shutdown); let server = Server::builder() .build(SocketAddr::from(([0, 0, 0, 0], port))) @@ -86,7 +88,8 @@ pub async fn run_server( #[cfg(not(feature = "mock-responses"))] let handle = { - let service = service::IndexerService::new(config, storage_dir) + let service = service::IndexerService::new(config, storage_dir, shutdown.child_token()) + .await .context("Failed to initialize indexer service")?; server.start(service.into_rpc()) }; diff --git a/lez/indexer/service/src/main.rs b/lez/indexer/service/src/main.rs index 3e36967d..52f195e9 100644 --- a/lez/indexer/service/src/main.rs +++ b/lez/indexer/service/src/main.rs @@ -34,7 +34,9 @@ async fn main() -> Result<()> { let cancellation_token = listen_for_shutdown_signal(); let config = indexer_service::IndexerConfig::from_path(&config_path)?; - let indexer_handle = indexer_service::run_server(config, data_dir.as_path(), port).await?; + let indexer_handle = + indexer_service::run_server(config, data_dir.as_path(), port, cancellation_token.clone()) + .await?; tokio::select! { () = cancellation_token.cancelled() => { diff --git a/lez/indexer/service/src/mock_service.rs b/lez/indexer/service/src/mock_service.rs index 7bf6c528..70af6239 100644 --- a/lez/indexer/service/src/mock_service.rs +++ b/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 { + 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(()) } diff --git a/lez/indexer/service/src/service.rs b/lez/indexer/service/src/service.rs index 7a8ed90f..09759362 100644 --- a/lez/indexer/service/src/service.rs +++ b/lez/indexer/service/src/service.rs @@ -2,9 +2,11 @@ use std::{path::Path, pin::pin, sync::Arc}; use anyhow::{Context as _, Result, bail}; use arc_swap::ArcSwap; -use futures::{StreamExt as _, never::Never}; +use futures::StreamExt as _; 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}, @@ -12,6 +14,7 @@ use jsonrpsee::{ }; use log::{debug, error, info, warn}; use tokio::sync::mpsc::UnboundedSender; +use tokio_util::sync::CancellationToken; pub struct IndexerService { subscription_service: SubscriptionService, @@ -19,9 +22,13 @@ pub struct IndexerService { } impl IndexerService { - pub fn new(config: IndexerConfig, storage_dir: &Path) -> Result { - let indexer = IndexerCore::new(config, storage_dir)?; - let subscription_service = SubscriptionService::spawn_new(indexer.clone()); + pub async fn new( + config: IndexerConfig, + storage_dir: &Path, + shutdown: CancellationToken, + ) -> Result { + let indexer = IndexerCore::new(config, storage_dir).await?; + let subscription_service = SubscriptionService::spawn_new(indexer.clone(), shutdown); Ok(Self { subscription_service, @@ -149,6 +156,10 @@ impl indexer_service_rpc::RpcServer for IndexerService { Ok(tx_res) } + async fn get_status(&self) -> Result { + Ok(self.indexer.status().into()) + } + async fn healthcheck(&self) -> Result<(), ErrorObjectOwned> { // Checking, that indexer can calculate last state let _ = self @@ -164,15 +175,21 @@ impl indexer_service_rpc::RpcServer for IndexerService { struct SubscriptionService { parts: ArcSwap, indexer: IndexerCore, + /// Cancellation token that is used to signal the subscription service to shut down. + /// + /// NOTE: This will auto-cancel on `Drop`, so if your token is shared with other parts + /// use [`CancellationToken::child_token()`] instead. + shutdown: CancellationToken, } impl SubscriptionService { - pub fn spawn_new(indexer: IndexerCore) -> Self { - let parts = Self::spawn_respond_subscribers_loop(indexer.clone()); + pub fn spawn_new(indexer: IndexerCore, shutdown: CancellationToken) -> Self { + let parts = Self::spawn_respond_subscribers_loop(indexer.clone(), shutdown.clone()); Self { parts: ArcSwap::new(Arc::new(parts)), indexer, + shutdown, } } @@ -184,14 +201,18 @@ impl SubscriptionService { ); // Respawn the subscription service loop if it has finished (either with error or panic) - if guard.handle.is_finished() { + if guard.handle.is_finished() && !self.shutdown.is_cancelled() { drop(guard); - let new_parts = Self::spawn_respond_subscribers_loop(self.indexer.clone()); + let new_parts = Self::spawn_respond_subscribers_loop( + self.indexer.clone(), + self.shutdown.clone(), + ); let old_handle_and_sender = self.parts.swap(Arc::new(new_parts)); let old_parts = Arc::into_inner(old_handle_and_sender) .expect("There should be no other references to the old handle and sender"); match old_parts.handle.await { + Ok(Ok(())) => {} Ok(Err(err)) => { error!( "Subscription service loop has unexpectedly finished with error: {err:#}" @@ -209,7 +230,10 @@ impl SubscriptionService { Ok(()) } - fn spawn_respond_subscribers_loop(indexer: IndexerCore) -> SubscriptionLoopParts { + fn spawn_respond_subscribers_loop( + indexer: IndexerCore, + shutdown: CancellationToken, + ) -> SubscriptionLoopParts { let (new_subscription_sender, mut sub_receiver) = tokio::sync::mpsc::unbounded_channel::>(); @@ -225,6 +249,10 @@ impl SubscriptionService { )] loop { tokio::select! { + () = shutdown.cancelled() => { + info!("Shutdown requested; stopping block ingestion"); + return Ok(()); + } sub = sub_receiver.recv() => { let Some(subscription) = sub else { bail!("Subscription receiver closed unexpectedly"); @@ -253,10 +281,11 @@ impl SubscriptionService { } } }; - let res: anyhow::Result = run_loop.await; - let Err(err) = res; - error!("Subscription service loop has unexpectedly finished with error: {err:#?}"); - Err(err) + let res: anyhow::Result<()> = run_loop.await; + if let Err(err) = &res { + error!("Subscription service loop has unexpectedly finished with error: {err:#?}"); + } + res }); SubscriptionLoopParts { handle, @@ -267,12 +296,13 @@ impl SubscriptionService { impl Drop for SubscriptionService { fn drop(&mut self) { + self.shutdown.cancel(); self.parts.load().handle.abort(); } } struct SubscriptionLoopParts { - handle: tokio::task::JoinHandle>, + handle: tokio::task::JoinHandle>, new_subscription_sender: UnboundedSender>, } diff --git a/lez/sequencer/service/src/main.rs b/lez/sequencer/service/src/main.rs index e78ad502..8b577bb8 100644 --- a/lez/sequencer/service/src/main.rs +++ b/lez/sequencer/service/src/main.rs @@ -24,6 +24,8 @@ async fn main() -> Result<()> { let Args { config_path, port } = Args::parse(); + // TODO: handle this cancellation token more gracefully within Sequencer service + // similar to how we do in Indexer let cancellation_token = listen_for_shutdown_signal(); let config = sequencer_service::SequencerConfig::from_path(&config_path)?; diff --git a/lez/storage/Cargo.toml b/lez/storage/Cargo.toml index 930de77c..18c58e11 100644 --- a/lez/storage/Cargo.toml +++ b/lez/storage/Cargo.toml @@ -13,6 +13,7 @@ lee.workspace = true thiserror.workspace = true borsh.workspace = true +log.workspace = true rocksdb.workspace = true tempfile.workspace = true zstd.workspace = true diff --git a/lez/storage/src/indexer/indexer_cells.rs b/lez/storage/src/indexer/indexer_cells.rs index b19a5510..ef104f1a 100644 --- a/lez/storage/src/indexer/indexer_cells.rs +++ b/lez/storage/src/indexer/indexer_cells.rs @@ -7,9 +7,9 @@ use crate::{ error::DbError, 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, + CF_BREAKPOINT_NAME, CF_HASH_TO_ID, CF_TX_TO_ID, + DB_META_LAST_OBSERVED_L1_LIB_HEADER_ID_IN_DB_KEY, DB_META_STALL_REASON_KEY, + DB_META_TIP_SLOT_KEY, DB_META_ZONE_SDK_INDEXER_CURSOR_KEY, TX_HASH_CELL_NAME, }, }; @@ -36,29 +36,6 @@ impl SimpleWritableCell for LastObservedL1LibHeaderCell { } } -#[derive(Debug, BorshSerialize, BorshDeserialize)] -pub struct LastBreakpointIdCell(pub u64); - -impl SimpleStorableCell for LastBreakpointIdCell { - type KeyParams = (); - - const CELL_NAME: &'static str = DB_META_LAST_BREAKPOINT_ID; - const CF_NAME: &'static str = CF_META_NAME; -} - -impl SimpleReadableCell for LastBreakpointIdCell {} - -impl SimpleWritableCell for LastBreakpointIdCell { - fn value_constructor(&self) -> DbResult> { - borsh::to_vec(&self).map_err(|err| { - DbError::borsh_cast_message( - err, - Some("Failed to serialize last breakpoint id".to_owned()), - ) - }) - } -} - #[derive(BorshDeserialize)] pub struct BreakpointCellOwned(pub V03State); @@ -212,6 +189,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)] @@ -247,6 +245,40 @@ impl SimpleWritableCell for ZoneSdkIndexerCursorCellRef<'_> { } } +/// Opaque JSON bytes for the indexer's persisted `Option`. +#[derive(BorshDeserialize)] +pub struct StallReasonCellOwned(pub Vec); + +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> { + 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::{ diff --git a/lez/storage/src/indexer/mod.rs b/lez/storage/src/indexer/mod.rs index 87d586fb..0955ef26 100644 --- a/lez/storage/src/indexer/mod.rs +++ b/lez/storage/src/indexer/mod.rs @@ -5,6 +5,7 @@ use common::{ transaction::{LeeTransaction, clock_invocation}, }; use lee::{GENESIS_BLOCK_ID, V03State}; +use log::warn; use rocksdb::{ BoundColumnFamily, ColumnFamilyDescriptor, DBWithThreadMode, MultiThreaded, Options, }; @@ -20,10 +21,12 @@ pub mod write_non_atomic; /// Key base for storing metainformation about id of last observed L1 lib header in db. pub const DB_META_LAST_OBSERVED_L1_LIB_HEADER_ID_IN_DB_KEY: &str = "last_observed_l1_lib_header_in_db"; -/// Key base for storing metainformation about the last breakpoint. -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` 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"; @@ -84,9 +87,10 @@ impl RocksDBIO { let dbio = Self { db }; - // First breakpoint setup - dbio.put_breakpoint(0, initial_state)?; - dbio.put_meta_last_breakpoint_id(0)?; + // Seed the genesis snapshot once; reopening must not clobber it. + if dbio.get_breakpoint_opt(0)?.is_none() { + dbio.put_breakpoint(0, initial_state)?; + } Ok(dbio) } @@ -152,8 +156,29 @@ impl RocksDBIO { )); } - let br_id = closest_breakpoint_id(block_id); - let mut state = self.get_breakpoint(br_id)?; + // walk down to the nearest snapshot that exists + let target = closest_breakpoint_id(block_id); + let mut br_id = target; + let mut state = loop { + match self.get_breakpoint_opt(br_id)? { + Some(state) => break state, + None if br_id == 0 => { + return Err(DbError::db_interaction_error( + "Breakpoint 0 is missing".to_owned(), + )); + } + None => { + br_id = br_id + .checked_sub(1) + .expect("breakpoint_id > 0 checked above"); + } + } + }; + if br_id < target { + warn!( + "Breakpoint {target} missing; replaying from breakpoint {br_id} for block {block_id}" + ); + } let start = u64::from(BREAKPOINT_INTERVAL) .checked_mul(br_id) @@ -211,20 +236,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:?}" diff --git a/lez/storage/src/indexer/read_once.rs b/lez/storage/src/indexer/read_once.rs index 6e79adc4..3fbf6026 100644 --- a/lez/storage/src/indexer/read_once.rs +++ b/lez/storage/src/indexer/read_once.rs @@ -3,8 +3,8 @@ use crate::{ DBIO as _, cells::shared_cells::{BlockCell, FirstBlockCell, FirstBlockSetCell, LastBlockCell}, indexer::indexer_cells::{ - AccNumTxCell, BlockHashToBlockIdMapCell, BreakpointCellOwned, LastBreakpointIdCell, - LastObservedL1LibHeaderCell, TxHashToBlockIdMapCell, ZoneSdkIndexerCursorCellOwned, + AccNumTxCell, BlockHashToBlockIdMapCell, BreakpointCellOwned, LastObservedL1LibHeaderCell, + StallReasonCellOwned, TipSlotCell, TxHashToBlockIdMapCell, ZoneSdkIndexerCursorCellOwned, }, }; @@ -31,8 +31,8 @@ impl RocksDBIO { Ok(self.get_opt::(())?.is_some()) } - pub fn get_meta_last_breakpoint_id(&self) -> DbResult> { - self.get_opt::(()) + pub fn get_meta_tip_slot_in_db(&self) -> DbResult> { + self.get_opt::(()) .map(|opt| opt.map(|cell| cell.0)) } @@ -49,6 +49,11 @@ impl RocksDBIO { self.get::(br_id).map(|cell| cell.0) } + pub fn get_breakpoint_opt(&self, br_id: u64) -> DbResult> { + self.get_opt::(br_id) + .map(|opt| opt.map(|cell| cell.0)) + } + // Mappings pub fn get_block_id_by_hash(&self, hash: [u8; 32]) -> DbResult> { @@ -73,4 +78,8 @@ impl RocksDBIO { .get_opt::(())? .map(|cell| cell.0)) } + + pub fn get_stall_reason_bytes(&self) -> DbResult>> { + Ok(self.get_opt::(())?.map(|cell| cell.0)) + } } diff --git a/lez/storage/src/indexer/tests.rs b/lez/storage/src/indexer/tests.rs index 44df23d5..d87aaf1c 100644 --- a/lez/storage/src/indexer/tests.rs +++ b/lez/storage/src/indexer/tests.rs @@ -49,16 +49,16 @@ fn initial_state() -> lee::V03State { #[test] fn start_db() { + let initial_state = initial_state(); let temp_dir = tempdir().unwrap(); let temdir_path = temp_dir.path(); - let dbio = RocksDBIO::open_or_create(temdir_path, &initial_state()).unwrap(); + let dbio = RocksDBIO::open_or_create(temdir_path, &initial_state).unwrap(); let last_id = dbio.get_meta_last_block_id_in_db().unwrap(); let first_id = dbio.get_meta_first_block_id_in_db().unwrap(); let is_first_set = dbio.get_meta_is_first_block_set().unwrap(); let last_observed_l1_header = dbio.get_meta_last_observed_l1_lib_header_in_db().unwrap(); - let last_br_id = dbio.get_meta_last_breakpoint_id().unwrap(); let last_block = dbio.get_block(1).unwrap(); let breakpoint = dbio.get_breakpoint(0).unwrap(); let final_state = dbio.final_state().unwrap(); @@ -67,7 +67,6 @@ fn start_db() { assert_eq!(first_id, None); assert_eq!(last_observed_l1_header, None); assert!(!is_first_set); - assert_eq!(last_br_id, Some(0)); // TODO: Will be None after we remove hardcoded testnet state assert!(last_block.is_none()); assert_eq!( breakpoint.get_account_by_id(acc1()), @@ -81,13 +80,15 @@ fn start_db() { #[test] fn one_block_insertion() { + let initial_state = initial_state(); let temp_dir = tempdir().unwrap(); let temdir_path = temp_dir.path(); - let dbio = RocksDBIO::open_or_create(temdir_path, &initial_state()).unwrap(); + 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, &initial_state) + .unwrap(); let prev_hash = genesis_block.header.hash; let from = acc1(); @@ -98,7 +99,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, &initial_state).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(); @@ -107,7 +108,6 @@ fn one_block_insertion() { .unwrap() .unwrap(); let is_first_set = dbio.get_meta_is_first_block_set().unwrap(); - let last_br_id = dbio.get_meta_last_breakpoint_id().unwrap(); let last_block = dbio.get_block(last_id).unwrap().unwrap(); let breakpoint = dbio.get_breakpoint(0).unwrap(); let final_state = dbio.final_state().unwrap(); @@ -116,7 +116,6 @@ fn one_block_insertion() { assert_eq!(first_id, Some(1)); assert_eq!(last_observed_l1_header, [1; 32]); assert!(is_first_set); - assert_eq!(last_br_id, Some(0)); assert_eq!(last_block.header.hash, block.header.hash); assert_eq!( breakpoint.get_account_by_id(acc1()).balance @@ -131,17 +130,44 @@ fn one_block_insertion() { } #[test] -fn new_breakpoint() { +fn put_block_records_tip_inscription_slot() { + let initial_state = initial_state(); let temp_dir = tempdir().unwrap(); - let temdir_path = temp_dir.path(); + let dbio = RocksDBIO::open_or_create(temp_dir.path(), &initial_state).unwrap(); - let dbio = RocksDBIO::open_or_create(temdir_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, &initial_state) + .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, &initial_state) + .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, &initial_state) + .unwrap(); + assert_eq!(dbio.get_meta_tip_slot_in_db().unwrap(), Some(1_005)); +} + +#[test] +fn put_block_stores_breakpoint_in_same_batch() { + let initial_state = initial_state(); + let temp_dir = tempdir().unwrap(); + let dbio = RocksDBIO::open_or_create(temp_dir.path(), &initial_state).unwrap(); let from = acc1(); let to = acc2(); let sign_key = acc1_sign_key(); - for i in 1..=BREAKPOINT_INTERVAL + 1 { + // Chain blocks 1..=BREAKPOINT_INTERVAL. The snapshot is scheduled internally + // by put_block at the boundary block; every call passes the same recognizable + // marker state (the initial one), proving it's stored verbatim rather than + // recomputed. + for i in 1..=BREAKPOINT_INTERVAL { let prev_hash = dbio.get_meta_last_block_id_in_db().unwrap().map(|last_id| { let last_block = dbio.get_block(last_id).unwrap().unwrap(); last_block.header.hash @@ -155,51 +181,70 @@ 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, &initial_state).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(); - let is_first_set = dbio.get_meta_is_first_block_set().unwrap(); - let last_br_id = dbio.get_meta_last_breakpoint_id().unwrap(); - let last_block = dbio.get_block(last_id).unwrap().unwrap(); - let prev_breakpoint = dbio.get_breakpoint(0).unwrap(); - let breakpoint = dbio.get_breakpoint(1).unwrap(); - let final_state = dbio.final_state().unwrap(); + let bp1 = dbio.get_breakpoint(1).unwrap(); + assert_eq!(bp1.get_account_by_id(acc1()).balance, 10000); + assert_eq!(bp1.get_account_by_id(acc2()).balance, 20000); + // Only the boundary block schedules a write: breakpoint 0 must be the only other one. + assert_eq!( + dbio.get_breakpoint(0) + .unwrap() + .get_account_by_id(acc1()) + .balance, + 10000 + ); +} - assert_eq!(last_id, 101); - assert_eq!(first_id, Some(1)); - assert!(is_first_set); - assert_eq!(last_br_id, Some(1)); - assert_ne!(last_block.header.hash, genesis_block().header.hash); +#[test] +fn state_replay_falls_back_over_missing_breakpoints() { + let initial_state = initial_state(); + let temp_dir = tempdir().unwrap(); + let dbio = RocksDBIO::open_or_create(temp_dir.path(), &initial_state).unwrap(); + + let from = acc1(); + let to = acc2(); + let sign_key = acc1_sign_key(); + + for i in 1..=u64::from(BREAKPOINT_INTERVAL) + 1 { + let prev_hash = dbio.get_meta_last_block_id_in_db().unwrap().map(|last_id| { + let last_block = dbio.get_block(last_id).unwrap().unwrap(); + last_block.header.hash + }); + let transfer_tx = common::test_utils::create_transaction_native_token_transfer( + from, + (i - 1).into(), + to, + 1, + &sign_key, + ); + let block = produce_dummy_block(i, prev_hash, vec![transfer_tx]); + dbio.put_block(&block, [0; 32], 0, &initial_state).unwrap(); + } + + // Simulate a store whose boundary snapshot was lost (#605). + dbio.delete_breakpoint(1).unwrap(); + assert!(dbio.get_breakpoint_opt(1).unwrap().is_none()); + let final_state = dbio.final_state().unwrap(); assert_eq!( - prev_breakpoint.get_account_by_id(acc1()).balance - - final_state.get_account_by_id(acc1()).balance, - 101 + 10000 - final_state.get_account_by_id(acc1()).balance, + u128::from(BREAKPOINT_INTERVAL) + 1 ); assert_eq!( - final_state.get_account_by_id(acc2()).balance - - prev_breakpoint.get_account_by_id(acc2()).balance, - 101 - ); - assert_eq!( - breakpoint.get_account_by_id(acc1()).balance - - final_state.get_account_by_id(acc1()).balance, - 1 - ); - assert_eq!( - final_state.get_account_by_id(acc2()).balance - - breakpoint.get_account_by_id(acc2()).balance, - 1 + final_state.get_account_by_id(acc2()).balance - 20000, + u128::from(BREAKPOINT_INTERVAL) + 1 ); } #[test] fn simple_maps() { + let initial_state = initial_state(); let temp_dir = tempdir().unwrap(); let temdir_path = temp_dir.path(); - let dbio = RocksDBIO::open_or_create(temdir_path, &initial_state()).unwrap(); + let dbio = RocksDBIO::open_or_create(temdir_path, &initial_state).unwrap(); let from = acc1(); let to = acc2(); @@ -211,7 +256,7 @@ fn simple_maps() { let control_hash1 = block.header.hash; - dbio.put_block(&block, [1; 32]).unwrap(); + dbio.put_block(&block, [1; 32], 0, &initial_state).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 +268,7 @@ fn simple_maps() { let control_hash2 = block.header.hash; - dbio.put_block(&block, [2; 32]).unwrap(); + dbio.put_block(&block, [2; 32], 0, &initial_state).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 +280,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, &initial_state).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 +292,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, &initial_state).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(); @@ -268,12 +313,13 @@ fn simple_maps() { #[test] fn block_batch() { + let initial_state = initial_state(); let temp_dir = tempdir().unwrap(); let temdir_path = temp_dir.path(); let mut block_res = vec![]; - let dbio = RocksDBIO::open_or_create(temdir_path, &initial_state()).unwrap(); + let dbio = RocksDBIO::open_or_create(temdir_path, &initial_state).unwrap(); let from = acc1(); let to = acc2(); @@ -284,7 +330,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, &initial_state).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 +341,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, &initial_state).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 +352,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, &initial_state).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 +363,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, &initial_state).unwrap(); let block_hashes_mem: Vec<[u8; 32]> = block_res.into_iter().map(|bl| bl.header.hash.0).collect(); @@ -356,10 +402,11 @@ fn block_batch() { #[test] fn account_map() { + let initial_state = initial_state(); let temp_dir = tempdir().unwrap(); let temdir_path = temp_dir.path(); - let dbio = RocksDBIO::open_or_create(temdir_path, &initial_state()).unwrap(); + let dbio = RocksDBIO::open_or_create(temdir_path, &initial_state).unwrap(); let from = acc1(); let to = acc2(); @@ -376,7 +423,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, &initial_state).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 +438,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, &initial_state).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 +453,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, &initial_state).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 +465,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, &initial_state).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(); @@ -431,3 +478,15 @@ fn account_map() { assert_eq!(acc1_tx_limited_hashes.as_slice(), &tx_hash_res[1..5]); } + +#[test] +fn reopen_preserves_seeded_breakpoint() { + let initial_state = initial_state(); + let temp_dir = tempdir().unwrap(); + { + let dbio = RocksDBIO::open_or_create(temp_dir.path(), &initial_state).unwrap(); + assert!(dbio.get_breakpoint_opt(0).unwrap().is_some()); + } // drop releases the RocksDB lock + let dbio = RocksDBIO::open_or_create(temp_dir.path(), &initial_state).unwrap(); + assert!(dbio.get_breakpoint_opt(0).unwrap().is_some()); +} diff --git a/lez/storage/src/indexer/write_atomic.rs b/lez/storage/src/indexer/write_atomic.rs index ec28f8ec..729bcc02 100644 --- a/lez/storage/src/indexer/write_atomic.rs +++ b/lez/storage/src/indexer/write_atomic.rs @@ -2,13 +2,13 @@ use std::collections::HashMap; use rocksdb::WriteBatch; -use super::{BREAKPOINT_INTERVAL, Block, DbError, DbResult, RocksDBIO}; +use super::{BREAKPOINT_INTERVAL, Block, DbError, DbResult, RocksDBIO, V03State}; use crate::{ DBIO as _, cells::shared_cells::{FirstBlockCell, FirstBlockSetCell, LastBlockCell}, indexer::indexer_cells::{ - AccNumTxCell, BlockHashToBlockIdMapCell, LastBreakpointIdCell, LastObservedL1LibHeaderCell, - TxHashToBlockIdMapCell, + AccNumTxCell, BlockHashToBlockIdMapCell, BreakpointCellRef, LastObservedL1LibHeaderCell, + TipSlotCell, TxHashToBlockIdMapCell, }, }; @@ -131,21 +131,28 @@ impl RocksDBIO { self.put_batch(&LastObservedL1LibHeaderCell(l1_lib_header), (), write_batch) } - pub fn put_meta_last_breakpoint_id_batch( + pub fn put_meta_tip_slot_in_db_batch( &self, - br_id: u64, + l1_slot: u64, write_batch: &mut WriteBatch, ) -> DbResult<()> { - self.put_batch(&LastBreakpointIdCell(br_id), (), write_batch) + 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, `Slot`, + /// and (at interval-boundary blocks) a snapshot of `post_state`, the block's + /// post-application state. + pub fn put_block( + &self, + block: &Block, + l1_lib_header: [u8; 32], + l1_slot: u64, + post_state: &V03State, + ) -> 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)?; @@ -208,18 +216,23 @@ impl RocksDBIO { self.put_account_transactions_dependant(acc_id, &tx_hashes, &mut write_batch)?; } - self.db.write(write_batch).map_err(|rerr| { - DbError::rocksdb_cast_message(rerr, Some("Failed to write batch".to_owned())) - })?; - if block .header .block_id .is_multiple_of(BREAKPOINT_INTERVAL.into()) { - self.put_next_breakpoint()?; + let br_id = block + .header + .block_id + .checked_div(BREAKPOINT_INTERVAL.into()) + .expect("Breakpoint interval is not zero"); + self.put_batch(&BreakpointCellRef(post_state), br_id, &mut write_batch)?; } + self.db.write(write_batch).map_err(|rerr| { + DbError::rocksdb_cast_message(rerr, Some("Failed to write batch".to_owned())) + })?; + Ok(()) } } diff --git a/lez/storage/src/indexer/write_non_atomic.rs b/lez/storage/src/indexer/write_non_atomic.rs index 7ddab1dd..a2c178ec 100644 --- a/lez/storage/src/indexer/write_non_atomic.rs +++ b/lez/storage/src/indexer/write_non_atomic.rs @@ -1,9 +1,11 @@ -use super::{BREAKPOINT_INTERVAL, DbError, DbResult, RocksDBIO, V03State}; +use super::{DbResult, RocksDBIO, V03State}; +#[cfg(test)] +use crate::error::DbError; use crate::{ DBIO as _, cells::shared_cells::{FirstBlockSetCell, LastBlockCell}, indexer::indexer_cells::{ - BreakpointCellRef, LastBreakpointIdCell, LastObservedL1LibHeaderCell, + BreakpointCellRef, LastObservedL1LibHeaderCell, StallReasonCellRef, ZoneSdkIndexerCursorCellRef, }, }; @@ -23,10 +25,6 @@ impl RocksDBIO { self.put(&LastObservedL1LibHeaderCell(l1_lib_header), ()) } - pub fn put_meta_last_breakpoint_id(&self, br_id: u64) -> DbResult<()> { - self.put(&LastBreakpointIdCell(br_id), ()) - } - pub fn put_meta_is_first_block_set(&self) -> DbResult<()> { self.put(&FirstBlockSetCell(true), ()) } @@ -35,32 +33,25 @@ 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<()> { self.put(&BreakpointCellRef(breakpoint), br_id) } - pub fn put_next_breakpoint(&self) -> DbResult<()> { - let last_block = self.get_meta_last_block_id_in_db()?.unwrap_or(0); - let next_breakpoint_id = self - .get_meta_last_breakpoint_id()? - .unwrap_or(0) - .checked_add(1) - .expect("Breakpoint Id will be lesser than u64::MAX"); - let block_to_break_id = next_breakpoint_id - .checked_mul(u64::from(BREAKPOINT_INTERVAL)) - .expect("Reached maximum breakpoint id"); - - if block_to_break_id <= last_block { - let next_breakpoint = self.calculate_state_for_id(block_to_break_id)?; - - self.put_breakpoint(next_breakpoint_id, &next_breakpoint)?; - self.put_meta_last_breakpoint_id(next_breakpoint_id) - } else { - Err(DbError::db_interaction_error( - "Breakpoint not yet achieved".to_owned(), - )) - } + /// Deletes a breakpoint snapshot. Test-only fault injection for simulating + /// stores whose boundary snapshot was lost. + #[cfg(test)] + pub(crate) fn delete_breakpoint(&self, br_id: u64) -> DbResult<()> { + let key = borsh::to_vec(&br_id).map_err(|err| { + DbError::borsh_cast_message(err, Some("Failed to serialize breakpoint id".to_owned())) + })?; + self.db + .delete_cf(&self.breakpoint_column(), key) + .map_err(|rerr| DbError::rocksdb_cast_message(rerr, None)) } } diff --git a/test_fixtures/Cargo.toml b/test_fixtures/Cargo.toml index 6e87dcaa..88526dea 100644 --- a/test_fixtures/Cargo.toml +++ b/test_fixtures/Cargo.toml @@ -32,4 +32,5 @@ serde_json.workspace = true tempfile.workspace = true testcontainers = { version = "0.27.3", features = ["docker-compose"] } tokio = { workspace = true, features = ["rt-multi-thread", "macros"] } +tokio-util.workspace = true url.workspace = true diff --git a/test_fixtures/src/config.rs b/test_fixtures/src/config.rs index dca5d445..044871f1 100644 --- a/test_fixtures/src/config.rs +++ b/test_fixtures/src/config.rs @@ -210,6 +210,7 @@ pub fn indexer_config(bedrock_addr: SocketAddr) -> Result { auth: None, }, channel_id: bedrock_channel_id(), + allow_chain_reset: false, }) } diff --git a/test_fixtures/src/setup.rs b/test_fixtures/src/setup.rs index de5ee58d..2827ff2c 100644 --- a/test_fixtures/src/setup.rs +++ b/test_fixtures/src/setup.rs @@ -124,10 +124,15 @@ pub async fn setup_indexer(bedrock_addr: SocketAddr) -> Result<(IndexerHandle, T let indexer_config = config::indexer_config(bedrock_addr).context("Failed to create Indexer config")?; - indexer_service::run_server(indexer_config, temp_indexer_dir.path(), 0) - .await - .context("Failed to run Indexer Service") - .map(|handle| (handle, temp_indexer_dir)) + indexer_service::run_server( + indexer_config, + temp_indexer_dir.path(), + 0, + tokio_util::sync::CancellationToken::new(), + ) + .await + .context("Failed to run Indexer Service") + .map(|handle| (handle, temp_indexer_dir)) } pub async fn setup_sequencer(