diff --git a/Cargo.lock b/Cargo.lock index efb8cd2a..2d4d3dbc 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3913,6 +3913,7 @@ dependencies = [ "base64", "common", "hex", + "indexer_core", "lee", "lee_core", "schemars 1.2.1", diff --git a/integration_tests/tests/bridge.rs b/integration_tests/tests/bridge.rs index 9a76bad6..a8fff35e 100644 --- a/integration_tests/tests/bridge.rs +++ b/integration_tests/tests/bridge.rs @@ -572,18 +572,18 @@ async fn indexer_status_rpc_reports_caught_up_with_no_stall() -> anyhow::Result< let status = ctx.indexer_client().get_status().await?; assert_eq!( - status["state"], - serde_json::json!("caught_up"), - "indexer should be caught up, got {status}" + status.state, + indexer_service_protocol::IndexerSyncState::CaughtUp, + "indexer should be caught up, got {status:?}" ); assert!( - status["stallReason"].is_null(), - "indexer should have no stall reason after a clean run, got {status}" + status.stall_reason.is_none(), + "indexer should have no stall reason after a clean run, got {status:?}" ); assert_eq!( - status["indexedBlockId"], - serde_json::json!(indexer_tip), - "status indexedBlockId should equal the caught-up tip" + status.indexed_block_id, + Some(indexer_tip), + "status indexed_block_id should equal the caught-up tip" ); Ok(()) diff --git a/lez/indexer/core/src/block_store.rs b/lez/indexer/core/src/block_store.rs index f5fd08b4..3625396c 100644 --- a/lez/indexer/core/src/block_store.rs +++ b/lez/indexer/core/src/block_store.rs @@ -171,8 +171,8 @@ impl IndexerStore { .get_account_by_id(*account_id)) } - /// The last successfully applied block as `{block_id, hash}`, or `None` before - /// any block is stored (cold start). Read fresh from the store each call. + /// 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); @@ -186,51 +186,12 @@ impl IndexerStore { })) } - /// Returns `Some(err)` if `block` is not the valid continuation of the tip: - /// hash integrity, then block-id continuity, then `prev_block_hash` linkage. - fn acceptance_error(&self, block: &Block) -> Result> { - let computed = block.recompute_hash(); - if computed != block.header.hash { - return Ok(Some(BlockIngestError::HashMismatch { - computed, - header: block.header.hash, - })); - } - - match self.validated_tip()? { - None => { - if block.header.block_id != GENESIS_BLOCK_ID { - return Ok(Some(BlockIngestError::UnexpectedBlockId { - expected: GENESIS_BLOCK_ID, - got: block.header.block_id, - })); - } - } - Some(tip) => { - let expected = tip.block_id.checked_add(1).expect("block id overflow"); - if block.header.block_id != expected { - return Ok(Some(BlockIngestError::UnexpectedBlockId { - expected, - got: block.header.block_id, - })); - } - if block.header.prev_block_hash != tip.hash { - return Ok(Some(BlockIngestError::BrokenChainLink { - expected_prev: tip.hash, - got_prev: block.header.prev_block_hash, - })); - } - } - } - Ok(None) - } - /// Records the stall reason: the first break is stored verbatim; subsequent /// breaks only bump `orphans_since`, preserving the original cause. fn record_stall( &self, header: Option<&BlockHeader>, - l1_slot: serde_json::Value, + l1_slot: Slot, error: BlockIngestError, ) -> Result<()> { let stall = match self.get_stall_reason()? { @@ -252,32 +213,26 @@ impl IndexerStore { } /// Records a stall for an inscription that could not even be parsed. - pub fn record_deserialize_stall( - &self, - l1_slot: serde_json::Value, - error: String, - ) -> Result<()> { + pub fn record_deserialize_stall(&self, l1_slot: Slot, error: String) -> Result<()> { self.record_stall(None, l1_slot, BlockIngestError::Deserialize(error)) } /// Validates `block` against the tip and, if it chains, applies it atomically /// (scratch clone, commit only on full success) and advances the tip. On any /// failure records the stall and returns `Parked` without touching state. - pub async fn accept_block( - &self, - block: &Block, - l1_slot: serde_json::Value, - ) -> Result { + pub async fn accept_block(&self, block: &Block, l1_slot: Slot) -> Result { + let tip = self.validated_tip()?; + // idempotent edge case: re-delivery of the current tip // (e.g. after a crash that left the L1 cursor behind the applied tip) - if let Some(tip) = self.validated_tip()? + if let Some(tip) = &tip && block.header.block_id == tip.block_id && block.header.hash == tip.hash { return Ok(AcceptOutcome::AlreadyApplied); } - if let Some(err) = self.acceptance_error(block)? { + 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)); } @@ -307,6 +262,46 @@ impl IndexerStore { } } +/// 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 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`. @@ -380,7 +375,7 @@ mod stall_reason_tests { block_id: Some(7), block_hash: Some(HashType([1_u8; 32])), prev_block_hash: Some(HashType([2_u8; 32])), - l1_slot: serde_json::Value::Null, + l1_slot: Slot::from(42), error: BlockIngestError::StateTransition("boom".to_owned()), first_seen: Some(99), orphans_since: 3, @@ -393,7 +388,7 @@ mod stall_reason_tests { 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, serde_json::Value::Null); + assert_eq!(got.l1_slot, Slot::from(42)); assert_eq!(got.first_seen, Some(99)); store.set_stall_reason(&None).expect("clear"); @@ -434,10 +429,7 @@ mod tests { let genesis = produce_dummy_block(1, None, vec![]); let mut prev_hash = genesis.header.hash; assert!(matches!( - store - .accept_block(&genesis, serde_json::Value::Null) - .await - .unwrap(), + store.accept_block(&genesis, Slot::from(0)).await.unwrap(), AcceptOutcome::Applied )); @@ -447,10 +439,7 @@ mod tests { let block = produce_dummy_block(i + 2, Some(prev_hash), vec![tx]); prev_hash = block.header.hash; assert!(matches!( - store - .accept_block(&block, serde_json::Value::Null) - .await - .unwrap(), + store.accept_block(&block, Slot::from(0)).await.unwrap(), AcceptOutcome::Applied )); } @@ -480,19 +469,13 @@ mod tests { let genesis = produce_dummy_block(1, None, vec![]); let mut prev_hash = genesis.header.hash; - store - .accept_block(&genesis, serde_json::Value::Null) - .await - .unwrap(); + store.accept_block(&genesis, Slot::from(0)).await.unwrap(); 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, serde_json::Value::Null) - .await - .unwrap(); + store.accept_block(&block, Slot::from(0)).await.unwrap(); } // State at block N is inclusive of block N. @@ -547,7 +530,7 @@ mod accept_tests { let block = valid_hash_block(2, HashType([0_u8; 32])); let outcome = store - .accept_block(&block, serde_json::Value::Null) + .accept_block(&block, Slot::from(0)) .await .expect("accept"); @@ -572,7 +555,7 @@ mod accept_tests { block.header.timestamp = 999; // invalidates the stored hash let outcome = store - .accept_block(&block, serde_json::Value::Null) + .accept_block(&block, Slot::from(0)) .await .expect("accept"); assert!(matches!( @@ -588,12 +571,12 @@ mod accept_tests { let first = valid_hash_block(2, HashType([0_u8; 32])); store - .accept_block(&first, serde_json::Value::Null) + .accept_block(&first, Slot::from(0)) .await .expect("accept"); let second = valid_hash_block(3, HashType([0_u8; 32])); store - .accept_block(&second, serde_json::Value::Null) + .accept_block(&second, Slot::from(0)) .await .expect("accept"); @@ -608,7 +591,7 @@ mod accept_tests { let store = IndexerStore::open_db(dir.path()).expect("open store"); store - .record_deserialize_stall(serde_json::Value::Null, "bad bytes".to_owned()) + .record_deserialize_stall(Slot::from(0), "bad bytes".to_owned()) .expect("record"); let stall = store.get_stall_reason().expect("get").expect("present"); @@ -624,20 +607,14 @@ mod accept_tests { // Genesis (block 1, clock-only) applies and advances the tip. let genesis = produce_dummy_block(1, None, vec![]); assert!(matches!( - store - .accept_block(&genesis, serde_json::Value::Null) - .await - .unwrap(), + 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, serde_json::Value::Null) - .await - .unwrap(), + store.accept_block(&bad, Slot::from(0)).await.unwrap(), AcceptOutcome::Parked(BlockIngestError::UnexpectedBlockId { expected: 2, got: 3 @@ -656,10 +633,7 @@ mod accept_tests { // 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, serde_json::Value::Null) - .await - .unwrap(), + store.accept_block(&next, Slot::from(0)).await.unwrap(), AcceptOutcome::Applied )); assert!( @@ -687,7 +661,7 @@ mod accept_tests { let genesis = produce_dummy_block(1, None, vec![]); store - .accept_block(&genesis, serde_json::Value::Null) + .accept_block(&genesis, Slot::from(0)) .await .expect("accept genesis"); @@ -697,20 +671,14 @@ mod accept_tests { ); let block = produce_dummy_block(2, Some(genesis.header.hash), vec![tx]); assert!(matches!( - store - .accept_block(&block, serde_json::Value::Null) - .await - .unwrap(), + 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, serde_json::Value::Null) - .await - .unwrap(), + store.accept_block(&block, Slot::from(0)).await.unwrap(), AcceptOutcome::AlreadyApplied )); assert_eq!( diff --git a/lez/indexer/core/src/ingest_error.rs b/lez/indexer/core/src/ingest_error.rs index dce82b9c..1235120a 100644 --- a/lez/indexer/core/src/ingest_error.rs +++ b/lez/indexer/core/src/ingest_error.rs @@ -4,7 +4,6 @@ use serde::{Deserialize, Serialize}; /// Why the indexer could not apply an L2 block from the channel. Stored inside a /// [`crate::stall_reason::StallReason`] and surfaced on the status snapshot. #[derive(Debug, Clone, Serialize, Deserialize, thiserror::Error)] -#[serde(rename_all = "camelCase")] pub enum BlockIngestError { #[error("failed to deserialize L2 block: {0}")] Deserialize(String), @@ -37,7 +36,7 @@ mod tests { let value = serde_json::to_value(&err).expect("serialize"); assert_eq!( value, - serde_json::json!({ "unexpectedBlockId": { "expected": 5, "got": 7 } }) + serde_json::json!({ "UnexpectedBlockId": { "expected": 5, "got": 7 } }) ); let back: BlockIngestError = serde_json::from_value(value).expect("deserialize"); assert!(matches!( diff --git a/lez/indexer/core/src/lib.rs b/lez/indexer/core/src/lib.rs index d1feca64..d24488ee 100644 --- a/lez/indexer/core/src/lib.rs +++ b/lez/indexer/core/src/lib.rs @@ -42,7 +42,35 @@ 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.chain_identity_outcome().await? { + ChainIdentityOutcome::Consistent => Ok(core), + ChainIdentityOutcome::Mismatch { detail } if config.allow_chain_reset => { + warn!( + "Chain reset detected ({detail}). 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) + } + ChainIdentityOutcome::Mismatch { detail } => Err(anyhow::anyhow!( + "Indexer store at {} holds a different chain than the channel now serves \ + ({detail}). 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)); @@ -62,37 +90,6 @@ impl IndexerCore { }) } - /// 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 `allow_reset`, in which case wipe the store and re-index from - /// scratch. Used at service/FFI startup in place of `new`. - pub async fn new_with_chain_check( - config: IndexerConfig, - storage_dir: &Path, - allow_reset: bool, - ) -> Result { - let home = storage_dir.join(format!("rocksdb-{}", config.channel_id)); - let core = Self::new(config.clone(), storage_dir)?; - match core.chain_identity_outcome().await? { - ChainIdentityOutcome::Consistent => Ok(core), - ChainIdentityOutcome::Mismatch { detail } if allow_reset => { - warn!( - "Chain reset detected ({detail}). 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::new(config, storage_dir) - } - ChainIdentityOutcome::Mismatch { detail } => Err(anyhow::anyhow!( - "Indexer store at {} holds a different chain than the channel now serves \ - ({detail}). Run `just clean`, point at a fresh storage dir, or set \ - `allow_chain_reset` in the indexer config.", - home.display() - )), - } - } - /// Detects when the local store belongs to a different chain than the connected L1 /// (e.g. a wiped/restarted bedrock) so startup can reset instead of silently /// diverging. Mostly a dev convenience; won't trigger during normal live indexing. @@ -183,6 +180,28 @@ 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. + fn park_undeserializable(&self, slot: Slot, error: &impl std::fmt::Display) { + error!("Failed to deserialize L2 block from zone-sdk: {error}"); + if let Err(err) = self.store.record_deserialize_stall(slot, error.to_string()) { + warn!("Failed to record stall reason: {err:#}"); + } + self.set_status(IndexerSyncStatus::stalled(format!( + "failed to deserialize L2 block: {error}" + ))); + } + pub fn subscribe_parse_block_stream(&self) -> impl futures::Stream> + '_ { let poll_interval = self.config.consensus_info_polling_interval; let initial_cursor = self @@ -228,38 +247,29 @@ impl IndexerCore { ZoneMessage::Deposit(_) | ZoneMessage::Withdraw(_) => continue, }; - let l1_slot = serde_json::to_value(slot).unwrap_or(serde_json::Value::Null); - 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}"); - if let Err(err) = - self.store.record_deserialize_stall(l1_slot, e.to_string()) - { + Err(error) => { + error!("Failed to deserialize L2 block from zone-sdk: {error}"); + if let Err(err) = self.store.record_deserialize_stall(slot, error.to_string()) { warn!("Failed to record stall reason: {err:#}"); } self.set_status(IndexerSyncStatus::stalled(format!( - "failed to deserialize L2 block: {e}" + "failed to deserialize L2 block: {error}" ))); - // Advance the L1 read cursor past the broken inscription; - // the validated tip stays frozen. - cursor = Some(slot); - if let Err(err) = self.store.set_zone_cursor(&slot) { - warn!("Failed to persist indexer cursor: {err:#}"); - } + + + // L1 proceeds regardless + self.advance_cursor(&mut cursor, slot); continue; } }; - match self.store.accept_block(&block, l1_slot).await { + match self.store.accept_block(&block, slot).await { Ok(AcceptOutcome::Applied) => { info!("Indexed L2 block {}", block.header.block_id); self.set_status(IndexerSyncStatus::syncing()); - cursor = Some(slot); - if let Err(err) = self.store.set_zone_cursor(&slot) { - warn!("Failed to persist indexer cursor: {err:#}"); - } + self.advance_cursor(&mut cursor, slot); yield Ok(block); } Ok(AcceptOutcome::AlreadyApplied) => { @@ -267,10 +277,7 @@ impl IndexerCore { "Skipping already-applied block {}", block.header.block_id ); - cursor = Some(slot); - if let Err(err) = self.store.set_zone_cursor(&slot) { - warn!("Failed to persist indexer cursor: {err:#}"); - } + self.advance_cursor(&mut cursor, slot); } Ok(AcceptOutcome::Parked(ingest_err)) => { error!( @@ -278,15 +285,12 @@ impl IndexerCore { block.header.block_id ); self.set_status(IndexerSyncStatus::stalled(ingest_err.to_string())); - // Advance the L1 read cursor; tip stays frozen, no yield. - cursor = Some(slot); - if let Err(err) = self.store.set_zone_cursor(&slot) { - warn!("Failed to persist indexer cursor: {err:#}"); - } + // L1 proceeds regardless + self.advance_cursor(&mut cursor, slot); } Err(err) => { // Infrastructure error (DB read/write), not a bad block. - // Keep the cursor put; re-poll the same position next cycle. + // will re-poll from the same cursor next cycle. error!( "Store error applying block {}: {err:#}", block.header.block_id diff --git a/lez/indexer/core/src/stall_reason.rs b/lez/indexer/core/src/stall_reason.rs index fdad9861..c5118fcc 100644 --- a/lez/indexer/core/src/stall_reason.rs +++ b/lez/indexer/core/src/stall_reason.rs @@ -1,4 +1,5 @@ use common::HashType; +use logos_blockchain_zone_sdk::Slot; use serde::{Deserialize, Serialize}; use crate::ingest_error::BlockIngestError; @@ -6,15 +7,14 @@ 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 zone-sdk cursor position captured as JSON. +/// 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)] -#[serde(rename_all = "camelCase")] pub struct StallReason { pub block_id: Option, pub block_hash: Option, pub prev_block_hash: Option, - pub l1_slot: serde_json::Value, + pub l1_slot: Slot, pub error: BlockIngestError, pub first_seen: Option, /// Number of later non-chaining blocks (orphans, since the tip is frozen). diff --git a/lez/indexer/core/src/status.rs b/lez/indexer/core/src/status.rs index d4f6d918..939abec5 100644 --- a/lez/indexer/core/src/status.rs +++ b/lez/indexer/core/src/status.rs @@ -5,7 +5,6 @@ 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, @@ -23,7 +22,6 @@ pub enum IndexerSyncState { /// 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, @@ -78,7 +76,6 @@ 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, @@ -101,10 +98,10 @@ mod tests { assert_eq!( value, serde_json::json!({ - "state": "error", - "lastError": "boom", - "indexedBlockId": 7, - "stallReason": null, + "state": "Error", + "last_error": "boom", + "indexed_block_id": 7, + "stall_reason": null, }) ); } @@ -114,12 +111,14 @@ 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 { @@ -129,16 +128,16 @@ mod tests { block_id: Some(42), block_hash: None, prev_block_hash: None, - l1_slot: serde_json::Value::Null, + l1_slot: Slot::from(0), error: BlockIngestError::StateTransition("boom".to_owned()), 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["lastError"], serde_json::json!("broken chain link")); - assert_eq!(value["indexedBlockId"], serde_json::json!(41)); - assert_eq!(value["stallReason"]["orphansSince"], serde_json::json!(2)); + 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 1a71718c..8a1eafaf 100644 --- a/lez/indexer/ffi/src/api/lifecycle.rs +++ b/lez/indexer/ffi/src/api/lifecycle.rs @@ -112,13 +112,8 @@ unsafe fn setup_indexer( unsafe { Runtime::from_borrowed(caller.as_ref()) } }; - let allow_reset = config.allow_chain_reset; let core = runtime - .block_on(IndexerCore::new_with_chain_check( - config, - &storage_dir, - allow_reset, - )) + .block_on(IndexerCore::new(config, &storage_dir)) .map_err(|e| { log::error!("Could not initialize indexer core: {e}"); OperationStatus::InitializationError 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/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..f497574d 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,85 @@ 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::StateTransition(msg) => Self::StateTransition(msg), + } + } +} + +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..6ea06a81 100644 --- a/lez/indexer/service/protocol/src/lib.rs +++ b/lez/indexer/service/protocol/src/lib.rs @@ -363,3 +363,66 @@ 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, + }, + StateTransition(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 108ee6c2..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}; @@ -70,7 +72,7 @@ pub trait Rpc { ) -> Result, ErrorObjectOwned>; #[method(name = "getStatus")] - async fn get_status(&self) -> Result; + async fn get_status(&self) -> Result; // ToDo: expand healthcheck response into some kind of report #[method(name = "checkHealth")] diff --git a/lez/indexer/service/src/mock_service.rs b/lez/indexer/service/src/mock_service.rs index c92ebb0f..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,7 +325,7 @@ impl indexer_service_rpc::RpcServer for MockIndexerService { .collect()) } - async fn get_status(&self) -> Result { + async fn get_status(&self) -> Result { let indexed_block_id = self .state .read() @@ -335,12 +335,12 @@ impl indexer_service_rpc::RpcServer for MockIndexerService { .rev() .find(|block| block.bedrock_status == BedrockStatus::Finalized) .map(|block| block.header.block_id); - Ok(serde_json::json!({ - "state": "caught_up", - "lastError": null, - "indexedBlockId": indexed_block_id, - "stallReason": null, - })) + Ok(IndexerStatus { + state: IndexerSyncState::CaughtUp, + last_error: None, + indexed_block_id, + stall_reason: None, + }) } async fn healthcheck(&self) -> Result<(), ErrorObjectOwned> { diff --git a/lez/indexer/service/src/service.rs b/lez/indexer/service/src/service.rs index 0d1d8692..767560b1 100644 --- a/lez/indexer/service/src/service.rs +++ b/lez/indexer/service/src/service.rs @@ -4,7 +4,9 @@ use anyhow::{Context as _, Result, bail}; use arc_swap::ArcSwap; use futures::{StreamExt as _, never::Never}; use indexer_core::{IndexerCore, config::IndexerConfig}; -use indexer_service_protocol::{Account, AccountId, Block, BlockId, HashType, Transaction}; +use indexer_service_protocol::{ + Account, AccountId, Block, BlockId, HashType, IndexerStatus, Transaction, +}; use jsonrpsee::{ SubscriptionSink, core::{Serialize, SubscriptionResult, async_trait}, @@ -20,8 +22,7 @@ pub struct IndexerService { impl IndexerService { pub async fn new(config: IndexerConfig, storage_dir: &Path) -> Result { - let allow_reset = config.allow_chain_reset; - let indexer = IndexerCore::new_with_chain_check(config, storage_dir, allow_reset).await?; + let indexer = IndexerCore::new(config, storage_dir).await?; let subscription_service = SubscriptionService::spawn_new(indexer.clone()); Ok(Self { @@ -150,14 +151,8 @@ impl indexer_service_rpc::RpcServer for IndexerService { Ok(tx_res) } - async fn get_status(&self) -> Result { - serde_json::to_value(self.indexer.status()).map_err(|err| { - ErrorObjectOwned::owned( - ErrorCode::InternalError.code(), - "failed to serialize indexer status".to_owned(), - Some(format!("{err:#}")), - ) - }) + async fn get_status(&self) -> Result { + Ok(self.indexer.status().into()) } async fn healthcheck(&self) -> Result<(), ErrorObjectOwned> {