logos-blockchain/logos-execution-zone
1
0
Fork 0
mirror of https://github.com/logos-blockchain/logos-execution-zone.git synced 2026-07-09 15:29:34 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(indexer): address several reviews, use better return types with conversions, some docfixes

Browse Source
This commit is contained in:
erhant 2026-07-02 21:39:30 +03:00
parent 357c92f947
commit 9f2fa83945
16 changed files with 337 additions and 227 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
Cargo.lock generated
View File

@ -3913,6 +3913,7 @@ dependencies = [
"base64",
"common",
"hex",
"indexer_core",
"lee",
"lee_core",
"schemars 1.2.1",

16
integration_tests/tests/bridge.rs
View File

@ -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(())

164
lez/indexer/core/src/block_store.rs
View File

@ -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<Option<Tip>> {
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<Option<BlockIngestError>> {
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<AcceptOutcome> {
pub async fn accept_block(&self, block: &Block, l1_slot: Slot) -> Result<AcceptOutcome> {
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!(

3
lez/indexer/core/src/ingest_error.rs
View File

@ -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!(

126
lez/indexer/core/src/lib.rs
View File

@ -42,7 +42,35 @@ pub struct IndexerCore {
}
impl IndexerCore {
pub fn new(config: IndexerConfig, storage_dir: &Path) -> Result<Self> {
/// 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<Self> {
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<Self> {
// 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<Self> {
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: 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<Item = Result<Block>> + '_ {
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

6
lez/indexer/core/src/stall_reason.rs
View File

@ -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<u64>,
pub block_hash: Option<HashType>,
pub prev_block_hash: Option<HashType>,
pub l1_slot: serde_json::Value,
pub l1_slot: Slot,
pub error: BlockIngestError,
pub first_seen: Option<u64>,
/// Number of later non-chaining blocks (orphans, since the tip is frozen).

25
lez/indexer/core/src/status.rs
View File

@ -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<String>,
@ -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));
}
}

6
lez/indexer/ffi/indexer_ffi.h
View File

@ -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
*

7
lez/indexer/ffi/src/api/lifecycle.rs
View File

@ -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

6
lez/indexer/ffi/src/api/query.rs
View File

@ -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
///

3
lez/indexer/service/protocol/Cargo.toml
View File

@ -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"]

93
lez/indexer/service/protocol/src/convert.rs
View File

@ -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<ValidityWindow> for lee_core::program::ValidityWindow<u64> {
value.0.try_into()
}
}
// ============================================================================
// Indexer status conversions
// ============================================================================
impl From<indexer_core::status::IndexerSyncState> 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<indexer_core::BlockIngestError> 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<indexer_core::StallReason> 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<indexer_core::status::IndexerStatus> 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),
}
}
}

63
lez/indexer/service/protocol/src/lib.rs
View File

@ -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<u64>,
pub block_hash: Option<HashType>,
pub prev_block_hash: Option<HashType>,
pub l1_slot: u64,
pub error: BlockIngestError,
/// The breaking block's L2 timestamp (`None` for a deserialize break).
pub first_seen: Option<Timestamp>,
/// 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<String>,
pub indexed_block_id: Option<BlockId>,
pub stall_reason: Option<StallReason>,
}

6
lez/indexer/service/rpc/src/lib.rs
View File

@ -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<Vec<Transaction>, ErrorObjectOwned>;
#[method(name = "getStatus")]
async fn get_status(&self) -> Result<serde_json::Value, ErrorObjectOwned>;
async fn get_status(&self) -> Result<IndexerStatus, ErrorObjectOwned>;
// ToDo: expand healthcheck response into some kind of report
#[method(name = "checkHealth")]

22
lez/indexer/service/src/mock_service.rs
View File

@ -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<serde_json::Value, ErrorObjectOwned> {
async fn get_status(&self) -> Result<IndexerStatus, ErrorObjectOwned> {
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> {

17
lez/indexer/service/src/service.rs
View File

@ -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<Self> {
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::Value, ErrorObjectOwned> {
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<IndexerStatus, ErrorObjectOwned> {
Ok(self.indexer.status().into())
}
async fn healthcheck(&self) -> Result<(), ErrorObjectOwned> {