logos-blockchain/logos-execution-zone
1
0
Fork 0
mirror of https://github.com/logos-blockchain/logos-execution-zone.git synced 2026-06-29 18:39:30 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor(indexer): use "stall reason" instead of "chain breaker"

Browse Source
This commit is contained in:
erhant 2026-06-26 15:34:17 +03:00
parent 4cc0d17d95
commit 00939ae6d6
9 changed files with 85 additions and 88 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -14,7 +14,7 @@ use logos_blockchain_zone_sdk::Slot;
use storage::indexer::RocksDBIO;
use tokio::sync::RwLock;
use crate::{chain_breaker::ChainBreaker, ingest_error::BlockIngestError};
use crate::{ingest_error::BlockIngestError, stall_reason::StallReason};
struct Tip {
block_id: u64,
@ -25,7 +25,7 @@ struct Tip {
pub enum AcceptOutcome {
/// Chained and applied; tip and L1 read cursor both advance.
Applied,
/// Did not chain or failed to apply; tip stays frozen, breaker recorded.
/// Did not chain or failed to apply; tip stays frozen, stall recorded.
Parked(BlockIngestError),
}
@ -129,18 +129,18 @@ impl IndexerStore {
Ok(())
}
pub fn get_chain_breaker(&self) -> Result<Option<ChainBreaker>> {
let Some(bytes) = self.dbio.get_chain_breaker_bytes()? else {
pub fn get_stall_reason(&self) -> Result<Option<StallReason>> {
let Some(bytes) = self.dbio.get_stall_reason_bytes()? else {
return Ok(None);
};
let breaker: Option<ChainBreaker> =
serde_json::from_slice(&bytes).context("Failed to deserialize stored chain breaker")?;
Ok(breaker)
let stall: Option<StallReason> =
serde_json::from_slice(&bytes).context("Failed to deserialize stored stall reason")?;
Ok(stall)
}
pub fn set_chain_breaker(&self, breaker: &Option<ChainBreaker>) -> Result<()> {
let bytes = serde_json::to_vec(breaker).context("Failed to serialize chain breaker")?;
self.dbio.put_chain_breaker_bytes(&bytes)?;
pub fn set_stall_reason(&self, stall: &Option<StallReason>) -> Result<()> {
let bytes = serde_json::to_vec(stall).context("Failed to serialize stall reason")?;
self.dbio.put_stall_reason_bytes(&bytes)?;
Ok(())
}
@ -219,20 +219,20 @@ impl IndexerStore {
Ok(None)
}
/// Records the chain breaker: the first break is stored verbatim; subsequent
/// Records the stall reason: the first break is stored verbatim; subsequent
/// breaks only bump `orphans_since`, preserving the original cause.
fn record_break(
fn record_stall(
&self,
header: Option<&BlockHeader>,
l1_slot: serde_json::Value,
error: BlockIngestError,
) -> Result<()> {
let breaker = match self.get_chain_breaker()? {
let stall = match self.get_stall_reason()? {
Some(mut existing) => {
existing.orphans_since = existing.orphans_since.saturating_add(1);
existing
}
None => ChainBreaker {
None => StallReason {
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),
@ -242,35 +242,35 @@ impl IndexerStore {
orphans_since: 0,
},
};
self.set_chain_breaker(&Some(breaker))
self.set_stall_reason(&Some(stall))
}
/// Records a breaker for an inscription that could not even be parsed.
pub fn record_deserialize_break(
/// 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<()> {
self.record_break(None, l1_slot, BlockIngestError::Deserialize(error))
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 breaker and returns `Parked` without touching state.
/// 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> {
if let Some(err) = self.acceptance_error(block)? {
self.record_break(Some(&block.header), l1_slot, err.clone())?;
self.record_stall(Some(&block.header), l1_slot, err.clone())?;
return Ok(AcceptOutcome::Parked(err));
}
// 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) {
self.record_break(Some(&block.header), l1_slot, err.clone())?;
self.record_stall(Some(&block.header), l1_slot, err.clone())?;
return Ok(AcceptOutcome::Parked(err));
}
@ -278,13 +278,13 @@ impl IndexerStore {
stored.bedrock_status = BedrockStatus::Finalized;
if let Err(err) = self.dbio.put_block(&stored, [0_u8; 32]) {
let ingest_err = BlockIngestError::Storage(err.to_string());
self.record_break(Some(&block.header), l1_slot, ingest_err.clone())?;
self.record_stall(Some(&block.header), l1_slot, ingest_err.clone())?;
return Ok(AcceptOutcome::Parked(ingest_err));
}
// Commit in-memory state (infallible) only after the DB write succeeded.
*self.current_state.write().await = scratch;
self.set_chain_breaker(&None)?;
self.set_stall_reason(&None)?;
Ok(AcceptOutcome::Applied)
}
@ -407,20 +407,20 @@ fn apply_block_to_scratch(block: &Block, state: &mut V03State) -> Result<(), Blo
}
#[cfg(test)]
mod chain_breaker_tests {
mod stall_reason_tests {
use common::HashType;
use super::*;
use crate::{chain_breaker::ChainBreaker, ingest_error::BlockIngestError};
use crate::{ingest_error::BlockIngestError, stall_reason::StallReason};
#[tokio::test]
async fn chain_breaker_roundtrips_and_clears() {
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_chain_breaker().expect("get").is_none());
assert!(store.get_stall_reason().expect("get").is_none());
let breaker = ChainBreaker {
let stall = StallReason {
block_id: Some(7),
block_hash: Some(HashType([1_u8; 32])),
prev_block_hash: Some(HashType([2_u8; 32])),
@ -429,17 +429,15 @@ mod chain_breaker_tests {
first_seen: Some(99),
orphans_since: 3,
};
store
.set_chain_breaker(&Some(breaker))
.expect("set breaker");
store.set_stall_reason(&Some(stall)).expect("set stall");
let got = store.get_chain_breaker().expect("get").expect("present");
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(_)));
store.set_chain_breaker(&None).expect("clear");
assert!(store.get_chain_breaker().expect("get").is_none());
store.set_stall_reason(&None).expect("clear");
assert!(store.get_stall_reason().expect("get").is_none());
}
}
@ -606,9 +604,9 @@ mod accept_tests {
got: 2
})
));
let breaker = store.get_chain_breaker().expect("get").expect("present");
assert_eq!(breaker.block_id, Some(2));
assert_eq!(breaker.orphans_since, 0);
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]
@ -645,22 +643,22 @@ mod accept_tests {
.await
.expect("accept");
let breaker = store.get_chain_breaker().expect("get").expect("present");
assert_eq!(breaker.block_id, Some(2), "first breaker preserved");
assert_eq!(breaker.orphans_since, 1, "second break counted as orphan");
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_breaker_without_header() {
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_deserialize_break(serde_json::Value::Null, "bad bytes".to_owned())
.record_deserialize_stall(serde_json::Value::Null, "bad bytes".to_owned())
.expect("record");
let breaker = store.get_chain_breaker().expect("get").expect("present");
assert_eq!(breaker.block_id, None);
assert!(matches!(breaker.error, BlockIngestError::Deserialize(_)));
let stall = store.get_stall_reason().expect("get").expect("present");
assert_eq!(stall.block_id, None);
assert!(matches!(stall.error, BlockIngestError::Deserialize(_)));
}
}

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

@ -2,7 +2,7 @@ use common::HashType;
use serde::{Deserialize, Serialize};
/// Why the indexer could not apply an L2 block from the channel. Stored inside a
/// [`crate::chain_breaker::ChainBreaker`] and surfaced on the status snapshot.
/// [`crate::stall_reason::StallReason`] and surfaced on the status snapshot.
#[derive(Debug, Clone, Serialize, Deserialize, thiserror::Error)]
#[serde(rename_all = "camelCase")]
pub enum BlockIngestError {

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

@ -2,7 +2,6 @@ use std::{path::Path, sync::Arc};
use anyhow::Result;
use arc_swap::ArcSwap;
pub use chain_breaker::ChainBreaker;
use common::block::Block;
// ToDo: Remove after testnet
use futures::StreamExt as _;
@ -12,6 +11,7 @@ use logos_blockchain_core::header::HeaderId;
use logos_blockchain_zone_sdk::{
CommonHttpClient, ZoneMessage, adapter::NodeHttpClient, indexer::ZoneIndexer,
};
pub use stall_reason::StallReason;
use crate::{
block_store::IndexerStore,
@ -19,9 +19,9 @@ use crate::{
status::{IndexerStatus, IndexerSyncStatus},
};
pub mod block_store;
pub mod chain_breaker;
pub mod config;
pub mod ingest_error;
pub mod stall_reason;
pub mod status;
#[derive(Clone)]
@ -62,11 +62,11 @@ impl IndexerCore {
pub fn status(&self) -> IndexerStatus {
let sync = IndexerSyncStatus::clone(&self.status.load());
let indexed_block_id = self.store.get_last_block_id().ok().flatten();
let chain_breaker = self.store.get_chain_breaker().ok().flatten();
let stall_reason = self.store.get_stall_reason().ok().flatten();
IndexerStatus {
sync,
indexed_block_id,
chain_breaker,
stall_reason,
}
}

7
lez/indexer/core/src/chain_breaker.rs → lez/indexer/core/src/stall_reason.rs
View File

@ -5,19 +5,20 @@ use crate::ingest_error::BlockIngestError;
/// Diagnostic record of the first block that broke the L2 chain.
///
/// Later non-chaining blocks (orphans, since the tip is frozen) only bump `orphans_since`.
///
/// 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.
/// `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 ChainBreaker {
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 error: BlockIngestError,
pub first_seen: Option<u64>,
/// Number of later non-chaining blocks (orphans, since the tip is frozen).
///
/// TODO: We could store a different "branch" of blocks following this sta
pub orphans_since: u64,
}

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

@ -1,6 +1,6 @@
use serde::Serialize;
use crate::chain_breaker::ChainBreaker;
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".
@ -15,8 +15,8 @@ pub enum IndexerSyncState {
CaughtUp,
/// The last cycle failed (e.g. the L1 node is unreachable). See `last_error`.
Error,
/// Parked on a chain breaker: the validated tip is frozen awaiting a valid
/// continuation. See `last_error` and the snapshot's `chain_breaker`.
/// Parked on a stall reason: the validated tip is frozen awaiting a valid
/// continuation. See `last_error` and the snapshot's `stall_reason`.
Stalled,
}
@ -62,8 +62,8 @@ impl IndexerSyncStatus {
}
}
/// Parked on a chain breaker; `reason` mirrors the breaker's error message.
/// The full breaker is attached to the [`IndexerStatus`] snapshot.
/// 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,
@ -83,7 +83,7 @@ pub struct IndexerStatus {
#[serde(flatten)]
pub sync: IndexerSyncStatus,
pub indexed_block_id: Option<u64>,
pub chain_breaker: Option<ChainBreaker>,
pub stall_reason: Option<StallReason>,
}
#[cfg(test)]
@ -95,7 +95,7 @@ mod tests {
let status = IndexerStatus {
sync: IndexerSyncStatus::error("boom".to_owned()),
indexed_block_id: Some(7),
chain_breaker: None,
stall_reason: None,
};
let value = serde_json::to_value(&status).expect("serialize");
assert_eq!(
@ -104,7 +104,7 @@ mod tests {
"state": "error",
"lastError": "boom",
"indexedBlockId": 7,
"chainBreaker": null,
"stallReason": null,
})
);
}
@ -119,13 +119,13 @@ mod tests {
}
#[test]
fn stalled_status_serializes_with_breaker() {
use crate::{chain_breaker::ChainBreaker, ingest_error::BlockIngestError};
fn stalled_status_serializes_with_stall_reason() {
use crate::{ingest_error::BlockIngestError, stall_reason::StallReason};
let status = IndexerStatus {
sync: IndexerSyncStatus::stalled("broken chain link".to_owned()),
indexed_block_id: Some(41),
chain_breaker: Some(ChainBreaker {
stall_reason: Some(StallReason {
block_id: Some(42),
block_hash: None,
prev_block_hash: None,
@ -139,6 +139,6 @@ mod tests {
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["chainBreaker"]["orphansSince"], serde_json::json!(2));
assert_eq!(value["stallReason"]["orphansSince"], serde_json::json!(2));
}
}

24
lez/storage/src/indexer/indexer_cells.rs
View File

@ -7,8 +7,8 @@ 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_CHAIN_BREAKER_KEY,
DB_META_LAST_BREAKPOINT_ID, DB_META_LAST_OBSERVED_L1_LIB_HEADER_ID_IN_DB_KEY,
CF_BREAKPOINT_NAME, CF_HASH_TO_ID, CF_TX_TO_ID, DB_META_LAST_BREAKPOINT_ID,
DB_META_LAST_OBSERVED_L1_LIB_HEADER_ID_IN_DB_KEY, DB_META_STALL_REASON_KEY,
DB_META_ZONE_SDK_INDEXER_CURSOR_KEY, TX_HASH_CELL_NAME,
},
};
@ -247,36 +247,36 @@ impl SimpleWritableCell for ZoneSdkIndexerCursorCellRef<'_> {
}
}
/// Opaque JSON bytes for the indexer's persisted `Option<ChainBreaker>`.
/// Opaque JSON bytes for the indexer's persisted `Option<StallReason>`.
/// Serialized via `serde_json` by the caller (mirrors the zone-sdk cursor cell).
#[derive(BorshDeserialize)]
pub struct ChainBreakerCellOwned(pub Vec<u8>);
pub struct StallReasonCellOwned(pub Vec<u8>);
impl SimpleStorableCell for ChainBreakerCellOwned {
impl SimpleStorableCell for StallReasonCellOwned {
type KeyParams = ();
const CELL_NAME: &'static str = DB_META_CHAIN_BREAKER_KEY;
const CELL_NAME: &'static str = DB_META_STALL_REASON_KEY;
const CF_NAME: &'static str = CF_META_NAME;
}
impl SimpleReadableCell for ChainBreakerCellOwned {}
impl SimpleReadableCell for StallReasonCellOwned {}
#[derive(BorshSerialize)]
pub struct ChainBreakerCellRef<'bytes>(pub &'bytes [u8]);
pub struct StallReasonCellRef<'bytes>(pub &'bytes [u8]);
impl SimpleStorableCell for ChainBreakerCellRef<'_> {
impl SimpleStorableCell for StallReasonCellRef<'_> {
type KeyParams = ();
const CELL_NAME: &'static str = DB_META_CHAIN_BREAKER_KEY;
const CELL_NAME: &'static str = DB_META_STALL_REASON_KEY;
const CF_NAME: &'static str = CF_META_NAME;
}
impl SimpleWritableCell for ChainBreakerCellRef<'_> {
impl SimpleWritableCell for StallReasonCellRef<'_> {
fn value_constructor(&self) -> DbResult<Vec<u8>> {
borsh::to_vec(&self).map_err(|err| {
DbError::borsh_cast_message(
err,
Some("Failed to serialize chain breaker cell".to_owned()),
Some("Failed to serialize stall reason cell".to_owned()),
)
})
}

4
lez/storage/src/indexer/mod.rs
View File

@ -24,8 +24,8 @@ pub const DB_META_LAST_OBSERVED_L1_LIB_HEADER_ID_IN_DB_KEY: &str =
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<ChainBreaker>` diagnostic record (opaque JSON bytes).
pub const DB_META_CHAIN_BREAKER_KEY: &str = "chain_breaker";
/// Key base for storing the persisted `Option<StallReason>` diagnostic record (opaque JSON bytes).
pub const DB_META_STALL_REASON_KEY: &str = "stall_reason";
/// Cell name for a breakpoint.
pub const BREAKPOINT_CELL_NAME: &str = "breakpoint";

10
lez/storage/src/indexer/read_once.rs
View File

@ -3,8 +3,8 @@ use crate::{
DBIO as _,
cells::shared_cells::{BlockCell, FirstBlockCell, FirstBlockSetCell, LastBlockCell},
indexer::indexer_cells::{
AccNumTxCell, BlockHashToBlockIdMapCell, BreakpointCellOwned, ChainBreakerCellOwned,
LastBreakpointIdCell, LastObservedL1LibHeaderCell, TxHashToBlockIdMapCell,
AccNumTxCell, BlockHashToBlockIdMapCell, BreakpointCellOwned, LastBreakpointIdCell,
LastObservedL1LibHeaderCell, StallReasonCellOwned, TxHashToBlockIdMapCell,
ZoneSdkIndexerCursorCellOwned,
},
};
@ -75,9 +75,7 @@ impl RocksDBIO {
.map(|cell| cell.0))
}
pub fn get_chain_breaker_bytes(&self) -> DbResult<Option<Vec<u8>>> {
Ok(self
.get_opt::<ChainBreakerCellOwned>(())?
.map(|cell| cell.0))
pub fn get_stall_reason_bytes(&self) -> DbResult<Option<Vec<u8>>> {
Ok(self.get_opt::<StallReasonCellOwned>(())?.map(|cell| cell.0))
}
}

6
lez/storage/src/indexer/write_non_atomic.rs
View File

@ -3,7 +3,7 @@ use crate::{
DBIO as _,
cells::shared_cells::{FirstBlockSetCell, LastBlockCell},
indexer::indexer_cells::{
BreakpointCellRef, ChainBreakerCellRef, LastBreakpointIdCell, LastObservedL1LibHeaderCell,
BreakpointCellRef, LastBreakpointIdCell, LastObservedL1LibHeaderCell, StallReasonCellRef,
ZoneSdkIndexerCursorCellRef,
},
};
@ -35,8 +35,8 @@ impl RocksDBIO {
self.put(&ZoneSdkIndexerCursorCellRef(bytes), ())
}
pub fn put_chain_breaker_bytes(&self, bytes: &[u8]) -> DbResult<()> {
self.put(&ChainBreakerCellRef(bytes), ())
pub fn put_stall_reason_bytes(&self, bytes: &[u8]) -> DbResult<()> {
self.put(&StallReasonCellRef(bytes), ())
}
// State