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

lez/indexer/core/src/block_store.rs

@@ -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),
 }
 
@@ -134,18 +134,18 @@ impl IndexerStore {
         Ok(())
     }
 
-    pub fn get_chain_breaker(&self) -> Result<Option<ChainBreaker>> {
+    pub fn get_stall_reason(&self) -> Result<Option<StallReason>> {
-        let Some(bytes) = self.dbio.get_chain_breaker_bytes()? else {
+        let Some(bytes) = self.dbio.get_stall_reason_bytes()? else {
             return Ok(None);
         };
-        let breaker: Option<ChainBreaker> =
+        let stall: Option<StallReason> =
-            serde_json::from_slice(&bytes).context("Failed to deserialize stored chain breaker")?;
+            serde_json::from_slice(&bytes).context("Failed to deserialize stored stall reason")?;
-        Ok(breaker)
+        Ok(stall)
     }
 
-    pub fn set_chain_breaker(&self, breaker: &Option<ChainBreaker>) -> Result<()> {
+    pub fn set_stall_reason(&self, stall: &Option<StallReason>) -> Result<()> {
-        let bytes = serde_json::to_vec(breaker).context("Failed to serialize chain breaker")?;
+        let bytes = serde_json::to_vec(stall).context("Failed to serialize stall reason")?;
-        self.dbio.put_chain_breaker_bytes(&bytes)?;
+        self.dbio.put_stall_reason_bytes(&bytes)?;
         Ok(())
     }
 
@@ -224,20 +224,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),
@@ -247,35 +247,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.
+    /// Records a stall for an inscription that could not even be parsed.
-    pub fn record_deserialize_break(
+    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));
         }
 
@@ -283,13 +283,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)
     }
 
@@ -412,20 +412,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])),
@@ -434,17 +434,15 @@ mod chain_breaker_tests {
             first_seen: Some(99),
             orphans_since: 3,
         };
-        store
+        store.set_stall_reason(&Some(stall)).expect("set stall");
-            .set_chain_breaker(&Some(breaker))
-            .expect("set breaker");
 
-        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");
+        store.set_stall_reason(&None).expect("clear");
-        assert!(store.get_chain_breaker().expect("get").is_none());
+        assert!(store.get_stall_reason().expect("get").is_none());
     }
 }
 
@@ -636,9 +634,9 @@ mod accept_tests {
                 got: 2
             })
         ));
-        let breaker = store.get_chain_breaker().expect("get").expect("present");
+        let stall = store.get_stall_reason().expect("get").expect("present");
-        assert_eq!(breaker.block_id, Some(2));
+        assert_eq!(stall.block_id, Some(2));
-        assert_eq!(breaker.orphans_since, 0);
+        assert_eq!(stall.orphans_since, 0);
     }
 
     #[tokio::test]
@@ -675,22 +673,22 @@ mod accept_tests {
             .await
             .expect("accept");
 
-        let breaker = store.get_chain_breaker().expect("get").expect("present");
+        let stall = store.get_stall_reason().expect("get").expect("present");
-        assert_eq!(breaker.block_id, Some(2), "first breaker preserved");
+        assert_eq!(stall.block_id, Some(2), "first stall preserved");
-        assert_eq!(breaker.orphans_since, 1, "second break counted as orphan");
+        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");
+        let stall = store.get_stall_reason().expect("get").expect("present");
-        assert_eq!(breaker.block_id, None);
+        assert_eq!(stall.block_id, None);
-        assert!(matches!(breaker.error, BlockIngestError::Deserialize(_)));
+        assert!(matches!(stall.error, BlockIngestError::Deserialize(_)));
     }
 }

lez/indexer/core/src/ingest_error.rs

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

lez/indexer/core/src/lib.rs

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

lez/indexer/core/src/stall_reason.rs

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

lez/indexer/core/src/status.rs

@@ -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
+    /// Parked on a stall reason: the validated tip is frozen awaiting a valid
-    /// continuation. See `last_error` and the snapshot's `chain_breaker`.
+    /// 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.
+    /// Parked on a stall reason; `reason` mirrors the stall's error message.
-    /// The full breaker is attached to the [`IndexerStatus`] snapshot.
+    /// 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() {
+    fn stalled_status_serializes_with_stall_reason() {
-        use crate::{chain_breaker::ChainBreaker, ingest_error::BlockIngestError};
+        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));
     }
 }

lez/storage/src/indexer/indexer_cells.rs

@@ -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,
+        CF_BREAKPOINT_NAME, CF_HASH_TO_ID, CF_TX_TO_ID, DB_META_LAST_BREAKPOINT_ID,
-        DB_META_LAST_BREAKPOINT_ID, DB_META_LAST_OBSERVED_L1_LIB_HEADER_ID_IN_DB_KEY,
+        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()),
             )
         })
     }

lez/storage/src/indexer/mod.rs

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

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, ChainBreakerCellOwned,
+        AccNumTxCell, BlockHashToBlockIdMapCell, BreakpointCellOwned, LastBreakpointIdCell,
-        LastBreakpointIdCell, LastObservedL1LibHeaderCell, TxHashToBlockIdMapCell,
+        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>>> {
+    pub fn get_stall_reason_bytes(&self) -> DbResult<Option<Vec<u8>>> {
-        Ok(self
+        Ok(self.get_opt::<StallReasonCellOwned>(())?.map(|cell| cell.0))
-            .get_opt::<ChainBreakerCellOwned>(())?
-            .map(|cell| cell.0))
     }
 }

lez/storage/src/indexer/write_non_atomic.rs

@@ -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<()> {
+    pub fn put_stall_reason_bytes(&self, bytes: &[u8]) -> DbResult<()> {
-        self.put(&ChainBreakerCellRef(bytes), ())
+        self.put(&StallReasonCellRef(bytes), ())
     }
 
     // State