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

Merge 9b2b4a3d8ca714fa089df634a230d1e0764dcdd7 into 2921438e932becd0ec69acd6eaa92317620106e8

Browse Source
This commit is contained in:
erhant 2026-07-09 18:28:37 +03:00 committed by GitHub
commit 480d293385
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
32 changed files with 1965 additions and 304 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
Cargo.lock generated
View File

@ -3863,6 +3863,7 @@ dependencies = [
"storage",
"tempfile",
"testnet_initial_state",
"thiserror 2.0.18",
"tokio",
"url",
]
@ -3910,6 +3911,7 @@ dependencies = [
"base64",
"common",
"hex",
"indexer_core",
"lee",
"lee_core",
"schemars 1.2.1",

54
integration_tests/tests/indexer_stall.rs Normal file
View File

@ -0,0 +1,54 @@
#![expect(
clippy::tests_outside_test_module,
reason = "We don't care about these in tests"
)]
use std::time::Duration;
use anyhow::{Context as _, Result};
use indexer_service_protocol::IndexerSyncState;
use indexer_service_rpc::RpcClient as _;
use integration_tests::{TestContext, wait_for_indexer_to_catch_up};
use log::info;
const CAUGHT_UP_STATUS_TIMEOUT: Duration = Duration::from_secs(60);
/// Test that the indexer status RPC reports caught-up with no stall after a clean run.
///
/// The sequencer keeps producing blocks while we assert, so the status is polled until a
/// `CaughtUp` snapshot is observed and the indexed tip is checked as a lower bound.
///
/// TODO: Integration-level park testing (publishing a bad block to force a stall) is a follow-up
/// needing fault injection support in the test harness.
#[tokio::test]
async fn indexer_status_rpc_reports_caught_up_with_no_stall() -> Result<()> {
let ctx = TestContext::new().await?;
let indexer_tip = wait_for_indexer_to_catch_up(&ctx).await?;
let status = tokio::time::timeout(CAUGHT_UP_STATUS_TIMEOUT, async {
loop {
let status = ctx.indexer_client().get_status().await?;
if status.state == IndexerSyncState::CaughtUp {
return anyhow::Ok(status);
}
info!("Waiting for caught-up indexer status, got {status:?}");
tokio::time::sleep(Duration::from_millis(500)).await;
}
})
.await
.context("Timed out waiting for indexer status to report caught-up")??;
assert!(
status.stall_reason.is_none(),
"indexer should have no stall reason after a clean run, got {status:?}"
);
// test for >= here because the sequencer keeps producing blocks while we assert,
// so the indexed tip may be ahead of the tip we observed when we waited for caught-up.
assert!(
status.indexed_block_id >= Some(indexer_tip),
"status indexed_block_id should be at least the caught-up tip {indexer_tip}, got {status:?}"
);
Ok(())
}

56
lez/common/src/block.rs
View File

@ -55,6 +55,21 @@ pub struct Block {
pub bedrock_status: BedrockStatus,
}
impl Block {
/// Recomputes the hash from this block's contents, for integrity verification
/// against the value stored in `header.hash`.
#[must_use]
pub fn recompute_hash(&self) -> BlockHash {
HashableBlockData {
block_id: self.header.block_id,
prev_block_hash: self.header.prev_block_hash,
timestamp: self.header.timestamp,
transactions: self.body.transactions.clone(),
}
.compute_hash()
}
}
impl Serialize for Block {
fn serialize<S: serde::Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
crate::borsh_base64::serialize(self, serializer)
@ -76,11 +91,13 @@ pub struct HashableBlockData {
}
impl HashableBlockData {
/// Domain-separated hash of the block contents: `SHA256(PREFIX || borsh(self))`.
/// The single source of truth for both producing and verifying a block hash.
#[must_use]
pub fn into_pending_block(self, signing_key: &lee::PrivateKey) -> Block {
pub fn compute_hash(&self) -> BlockHash {
const PREFIX: &[u8; 32] = b"/LEE/v0.3/Message/Block/\x00\x00\x00\x00\x00\x00\x00\x00";
let data_bytes = borsh::to_vec(&self).unwrap();
let data_bytes = borsh::to_vec(self).unwrap();
let mut bytes = Vec::with_capacity(
PREFIX
.len()
@ -89,8 +106,12 @@ impl HashableBlockData {
);
bytes.extend_from_slice(PREFIX);
bytes.extend_from_slice(&data_bytes);
OwnHasher::hash(&bytes)
}
let hash = OwnHasher::hash(&bytes);
#[must_use]
pub fn into_pending_block(self, signing_key: &lee::PrivateKey) -> Block {
let hash = self.compute_hash();
let signature = lee::Signature::new(signing_key, &hash.0);
Block {
header: BlockHeader {
@ -132,4 +153,33 @@ mod tests {
let block_from_bytes = borsh::from_slice::<HashableBlockData>(&bytes).unwrap();
assert_eq!(hashable, block_from_bytes);
}
#[test]
fn recompute_hash_matches_header_for_well_formed_block() {
let key = lee::PrivateKey::try_new([7_u8; 32]).expect("valid key");
let block = HashableBlockData {
block_id: 5,
prev_block_hash: HashType([9_u8; 32]),
timestamp: 42,
transactions: vec![test_utils::produce_dummy_empty_transaction()],
}
.into_pending_block(&key);
assert_eq!(block.recompute_hash(), block.header.hash);
}
#[test]
fn recompute_hash_detects_tampering() {
let key = lee::PrivateKey::try_new([7_u8; 32]).expect("valid key");
let block = HashableBlockData {
block_id: 5,
prev_block_hash: HashType([9_u8; 32]),
timestamp: 42,
transactions: vec![test_utils::produce_dummy_empty_transaction()],
}
.into_pending_block(&key);
let mut tampered = block;
tampered.header.timestamp = 99; // header changed; stale hash no longer matches
assert_ne!(tampered.recompute_hash(), tampered.header.hash);
}
}

19
lez/common/src/transaction.rs
View File

@ -92,9 +92,8 @@ impl LeeTransaction {
Ok(diff)
}
/// Computes the validated state diff without enforcing the system-account
/// restriction. Shared by [`Self::validate_on_state`] and
/// [`Self::execute_without_system_accounts_check_on_state`].
/// Computes the validated state diff. Shared by [`Self::validate_on_state`]
/// (which adds the system-account guards) and [`Self::execute_on_state`].
fn compute_state_diff(
&self,
state: &V03State,
@ -129,16 +128,12 @@ impl LeeTransaction {
Ok(self)
}
/// Similar to [`Self::execute_check_on_state`], but skips the system-account guard.
/// Executes the transaction against the current state and applies the resulting diff,
/// without the system-account guards enforced by [`Self::execute_check_on_state`].
///
/// FIXME: HOT FIX (testnet v0.2): the indexer replays blocks the sequencer already
/// accepted, including sequencer-generated deposit transactions that
/// legitimately modify the bridge account. The `TransactionOrigin::Sequencer`
/// tag that lets the sequencer bypass the guard is not carried in the block,
/// so the indexer cannot yet distinguish deposit txs from user txs.
///
/// REMOVE ME when the indexer can authenticate deposit transactions.
pub fn execute_without_system_accounts_check_on_state(
/// The indexer replays blocks the sequencer already validated and inscribed on Bedrock,
/// so it trusts those inscriptions and re-derives state without re-validating them.
pub fn execute_on_state(
self,
state: &mut V03State,
block_id: BlockId,

3
lez/configs/docker-all-in-one/indexer_config.json
View File

@ -3,5 +3,6 @@
"bedrock_config": {
"addr": "http://logos-blockchain-node-0:18080"
},
"channel_id": "0101010101010101010101010101010101010101010101010101010101010101"
"channel_id": "0101010101010101010101010101010101010101010101010101010101010101",
"allow_chain_reset": true
}

1
lez/indexer/core/Cargo.toml
View File

@ -29,6 +29,7 @@ futures.workspace = true
url.workspace = true
logos-blockchain-core.workspace = true
serde_json.workspace = true
thiserror.workspace = true
async-stream.workspace = true
tokio.workspace = true

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

@ -2,17 +2,35 @@ use std::{path::Path, sync::Arc};
use anyhow::{Context as _, Result};
use common::{
block::{BedrockStatus, Block},
HashType,
block::{BedrockStatus, Block, BlockHeader},
transaction::{LeeTransaction, clock_invocation},
};
use lee::{Account, AccountId, V03State};
use lee::{Account, AccountId, GENESIS_BLOCK_ID, V03State};
use lee_core::BlockId;
use log::info;
use log::warn;
use logos_blockchain_core::header::HeaderId;
use logos_blockchain_zone_sdk::Slot;
use storage::indexer::RocksDBIO;
use tokio::sync::RwLock;
use crate::{ingest_error::BlockIngestError, stall_reason::StallReason};
struct Tip {
block_id: u64,
hash: HashType,
}
/// Outcome of feeding a parsed L2 block to the validated tip.
pub enum AcceptOutcome {
/// Chained and applied; tip and L1 read cursor both advance.
Applied,
/// A duplicate re-delivery of the current tip. Just L2 advances.
AlreadyApplied,
/// Did not chain or failed to apply; tip stays frozen, stall recorded.
Parked(BlockIngestError),
}
#[derive(Clone)]
pub struct IndexerStore {
dbio: Arc<RocksDBIO>,
@ -118,6 +136,36 @@ impl IndexerStore {
Ok(())
}
/// The L1 inscription slot of the validated tip, written atomically with it
/// by [`Self::accept_block`]. `None` on a cold store or one written before
/// the slot was recorded.
pub fn get_tip_slot(&self) -> Result<Option<Slot>> {
Ok(self.dbio.get_meta_tip_slot_in_db()?.map(Slot::from))
}
pub fn get_stall_reason(&self) -> Result<Option<StallReason>> {
let Some(bytes) = self.dbio.get_stall_reason_bytes()? else {
return Ok(None);
};
let stall: Option<StallReason> =
serde_json::from_slice(&bytes).context("Failed to deserialize stored stall reason")?;
Ok(stall)
}
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(())
}
/// Clears a recorded stall marker if one is present, skipping the write otherwise.
fn clear_stall_if_present(&self) -> Result<()> {
if self.get_stall_reason()?.is_some() {
self.set_stall_reason(&None)?;
}
Ok(())
}
/// Recalculation of final state directly from DB.
///
/// Used for indexer healthcheck.
@ -139,137 +187,247 @@ impl IndexerStore {
.get_account_by_id(*account_id))
}
pub async fn put_block(&self, mut block: Block, l1_header: HeaderId) -> Result<()> {
info!("Applying block {}", block.header.block_id);
{
let mut state_guard = self.current_state.write().await;
/// The last successfully applied block, or `None` on a cold store.
/// Read fresh from the store each call.
fn validated_tip(&self) -> Result<Option<Tip>> {
let Some(block_id) = self.dbio.get_meta_last_block_id_in_db()? else {
return Ok(None);
};
let Some(block) = self.dbio.get_block(block_id)? else {
return Ok(None);
};
Ok(Some(Tip {
block_id,
hash: block.header.hash,
}))
}
let (clock_tx, user_txs) = block
.body
.transactions
.split_last()
.ok_or_else(|| anyhow::anyhow!("Block has no transactions"))?;
anyhow::ensure!(
*clock_tx == LeeTransaction::Public(clock_invocation(block.header.timestamp)),
"Last transaction in block must be the clock invocation for the block timestamp"
);
let is_genesis = block.header.block_id == 1;
for transaction in user_txs {
if is_genesis {
let genesis_tx = match transaction {
LeeTransaction::Public(public_tx) => public_tx,
LeeTransaction::PrivacyPreserving(_)
| LeeTransaction::ProgramDeployment(_) => {
anyhow::bail!("Genesis block should contain only public transactions")
}
};
state_guard
.transition_from_public_transaction(
genesis_tx,
block.header.block_id,
block.header.timestamp,
)
.context("Failed to execute genesis public transaction")?;
} else {
transaction
.clone()
.transaction_stateless_check()?
// FIXME: HOT FIX (testnet v0.2): does not check for system account updates due to
// sequencer-generated deposit tx'es;
// CHANGE ME back to `execute_check_on_state` when the indexer can authenticate deposit transactions
.execute_without_system_accounts_check_on_state(
&mut state_guard,
block.header.block_id,
block.header.timestamp,
)?;
}
/// Record the stall reason.
///
/// - First stall is stored verbatim
/// - Subsequent stalls only bump `orphans_since`, preserving the original cause.
pub fn record_stall(
&self,
header: Option<&BlockHeader>,
l1_slot: Slot,
error: BlockIngestError,
) -> Result<()> {
let stall = match self.get_stall_reason()? {
Some(mut existing) => {
existing.orphans_since = existing.orphans_since.saturating_add(1);
existing
}
None => StallReason {
// need to map out of `header` because they are not ser/de
block_id: header.map(|h| h.block_id),
block_hash: header.map(|h| h.hash),
prev_block_hash: header.map(|h| h.prev_block_hash),
first_seen: header.map(|h| h.timestamp),
l1_slot,
error,
orphans_since: 0,
},
};
self.set_stall_reason(&Some(stall))
}
// Apply the clock invocation directly (it is expected to modify clock accounts).
let LeeTransaction::Public(clock_public_tx) = clock_tx else {
anyhow::bail!("Clock invocation must be a public transaction");
};
state_guard.transition_from_public_transaction(
clock_public_tx,
block.header.block_id,
block.header.timestamp,
)?;
/// Validates `block` against the tip and, if it chains, applies it atomically
/// (scratch clone, commit only on full success) and advances the tip. On any
/// failure records the stall and returns `Parked` without touching state.
pub async fn accept_block(&self, block: &Block, l1_slot: Slot) -> Result<AcceptOutcome> {
let tip = self.validated_tip()?;
// Re-delivery of an already-applied block is idempotent, not a divergence
if let Some(tip) = &tip
&& block.header.block_id <= tip.block_id
&& let Some(stored) = self.get_block_at_id(block.header.block_id)?
&& stored.header.hash == block.header.hash
{
return Ok(AcceptOutcome::AlreadyApplied);
}
// ToDo: Currently we are fetching only finalized blocks
// if it changes, the following lines need to be updated
// to represent correct block finality
block.bedrock_status = BedrockStatus::Finalized;
if let Err(err) = validate_against_tip(tip.as_ref(), block) {
self.record_stall(Some(&block.header), l1_slot, err.clone())?;
return Ok(AcceptOutcome::Parked(err));
}
info!("Putting block {} into DB", block.header.block_id);
Ok(self.dbio.put_block(&block, l1_header.into())?)
// TODO: we use scratch state to be atomic, but need to revisit how expensive a clone is
let mut scratch = self.current_state.read().await.clone();
if let Err(err) = apply_block_to_scratch(block, &mut scratch) {
self.record_stall(Some(&block.header), l1_slot, err.clone())?;
return Ok(AcceptOutcome::Parked(err));
}
let mut stored = block.clone();
stored.bedrock_status = BedrockStatus::Finalized;
self.dbio
.put_block(&stored, [0_u8; 32], l1_slot.into_inner())
.context("Failed to persist accepted block")?;
// Commit in-memory state (infallible) only after the DB write succeeded.
*self.current_state.write().await = scratch;
// Best-effort: the block is durably applied, so a failed stall clear must not
// fail the apply. It self-heals on the next clear.
if let Err(err) = self.clear_stall_if_present() {
warn!("Failed to clear stall marker after applying block: {err:#}");
}
Ok(AcceptOutcome::Applied)
}
}
/// Checks that `block` is the valid continuation of `tip`: hash integrity,
/// then block-id continuity, then `prev_block_hash` linkage. A `None` tip
/// (cold store) expects the genesis block.
fn validate_against_tip(tip: Option<&Tip>, block: &Block) -> Result<(), BlockIngestError> {
let computed = block.recompute_hash();
if computed != block.header.hash {
return Err(BlockIngestError::HashMismatch {
computed,
header: block.header.hash,
});
}
match tip {
None => {
if block.header.block_id != GENESIS_BLOCK_ID {
return Err(BlockIngestError::UnexpectedBlockId {
expected: GENESIS_BLOCK_ID,
got: block.header.block_id,
});
}
}
Some(tip) => {
let expected = tip
.block_id
.checked_add(1)
.expect("block id should not overflow");
if block.header.block_id != expected {
return Err(BlockIngestError::UnexpectedBlockId {
expected,
got: block.header.block_id,
});
}
if block.header.prev_block_hash != tip.hash {
return Err(BlockIngestError::BrokenChainLink {
expected_prev: tip.hash,
got_prev: block.header.prev_block_hash,
});
}
}
}
Ok(())
}
/// Applies a block's transactions to `state`, mapping every failure to a
/// [`BlockIngestError`] so the caller can park rather than crash. Operates on a
/// scratch state; the caller commits only on `Ok`.
fn apply_block_to_scratch(block: &Block, state: &mut V03State) -> Result<(), BlockIngestError> {
let (clock_tx, user_txs) = block
.body
.transactions
.split_last()
.ok_or(BlockIngestError::EmptyBlock)?;
let expected_clock = LeeTransaction::Public(clock_invocation(block.header.timestamp));
if *clock_tx != expected_clock {
return Err(BlockIngestError::InvalidClockTransaction);
}
let is_genesis = block.header.block_id == GENESIS_BLOCK_ID;
for (tx_index, transaction) in user_txs.iter().enumerate() {
let state_transition = |err: anyhow::Error| BlockIngestError::StateTransition {
tx_index: tx_index.try_into().expect("tx index fits in u64"),
reason: format!("{err:#}"),
};
if is_genesis {
let LeeTransaction::Public(public_tx) = transaction else {
return Err(BlockIngestError::NonPublicGenesisTransaction);
};
state
.transition_from_public_transaction(
public_tx,
block.header.block_id,
block.header.timestamp,
)
.map_err(|err| state_transition(err.into()))?;
} else {
transaction
.clone()
.execute_on_state(state, block.header.block_id, block.header.timestamp)
.map_err(|err| state_transition(err.into()))?;
}
}
let LeeTransaction::Public(clock_public_tx) = clock_tx else {
return Err(BlockIngestError::InvalidClockTransaction);
};
state
.transition_from_public_transaction(
clock_public_tx,
block.header.block_id,
block.header.timestamp,
)
.map_err(|err| BlockIngestError::StateTransition {
tx_index: user_txs.len().try_into().expect("tx index fits in u64"),
reason: format!("{:#}", anyhow::Error::from(err)),
})?;
Ok(())
}
#[cfg(test)]
mod stall_reason_tests {
use common::HashType;
use super::*;
use crate::{ingest_error::BlockIngestError, stall_reason::StallReason};
#[tokio::test]
async fn stall_reason_roundtrips_and_clears() {
let dir = tempfile::tempdir().expect("tempdir");
let store = IndexerStore::open_db(dir.path()).expect("open store");
assert!(store.get_stall_reason().expect("get").is_none());
let stall = StallReason {
block_id: Some(7),
block_hash: Some(HashType([1_u8; 32])),
prev_block_hash: Some(HashType([2_u8; 32])),
l1_slot: Slot::from(42),
error: BlockIngestError::StateTransition {
tx_index: 0,
reason: "boom".to_owned(),
},
first_seen: Some(99),
orphans_since: 3,
};
store.set_stall_reason(&Some(stall)).expect("set stall");
let got = store.get_stall_reason().expect("get").expect("present");
assert_eq!(got.block_id, Some(7));
assert_eq!(got.orphans_since, 3);
assert!(matches!(
got.error,
BlockIngestError::StateTransition { .. }
));
assert_eq!(got.block_hash, Some(HashType([1_u8; 32])));
assert_eq!(got.prev_block_hash, Some(HashType([2_u8; 32])));
assert_eq!(got.l1_slot, Slot::from(42));
assert_eq!(got.first_seen, Some(99));
store.set_stall_reason(&None).expect("clear");
assert!(store.get_stall_reason().expect("get").is_none());
}
}
#[cfg(test)]
mod tests {
use common::{HashType, block::HashableBlockData};
use common::test_utils::{create_transaction_native_token_transfer, produce_dummy_block};
use tempfile::tempdir;
use testnet_initial_state::initial_pub_accounts_private_keys;
use super::*;
struct TestFixture {
storage: IndexerStore,
from: AccountId,
to: AccountId,
_home: tempfile::TempDir,
}
#[expect(
clippy::arithmetic_side_effects,
reason = "test helper with bounded inputs"
)]
async fn store_with_transfer_blocks(
block_count: u64,
prev_hash: Option<common::HashType>,
) -> TestFixture {
let home = tempdir().unwrap();
let storage = IndexerStore::open_db(home.path()).unwrap();
let initial_accounts = initial_pub_accounts_private_keys();
let from = initial_accounts[0].account_id;
let to = initial_accounts[1].account_id;
let sign_key = initial_accounts[0].pub_sign_key.clone();
let mut prev_hash = prev_hash;
for i in 0..block_count {
let tx = common::test_utils::create_transaction_native_token_transfer(
from,
u128::from(i),
to,
10,
&sign_key,
);
let block_id = i + 1;
let next_block = common::test_utils::produce_dummy_block(block_id, prev_hash, vec![tx]);
prev_hash = Some(next_block.header.hash);
storage
.put_block(
next_block,
HeaderId::from([u8::try_from(i + 1).unwrap(); 32]),
)
.await
.unwrap();
}
TestFixture {
storage,
from,
to,
_home: home,
}
}
#[test]
fn correct_startup() {
let home = tempdir().unwrap();
@ -282,75 +440,388 @@ mod tests {
}
#[tokio::test]
async fn state_transition() {
async fn accept_block_applies_transfers_and_advances_tip() {
let home = tempdir().unwrap();
let storage = IndexerStore::open_db(home.as_ref()).unwrap();
let store = IndexerStore::open_db(home.as_ref()).unwrap();
let initial_accounts = initial_pub_accounts_private_keys();
let from = initial_accounts[0].account_id;
let to = initial_accounts[1].account_id;
let sign_key = initial_accounts[0].pub_sign_key.clone();
let clock_tx = LeeTransaction::Public(clock_invocation(0));
let genesis_block_data = HashableBlockData {
block_id: 1,
prev_block_hash: HashType::default(),
timestamp: 0,
transactions: vec![clock_tx],
};
let genesis_block = genesis_block_data
.into_pending_block(&common::test_utils::sequencer_sign_key_for_testing());
let mut prev_hash = Some(genesis_block.header.hash);
storage
.put_block(genesis_block, HeaderId::from([0_u8; 32]))
.await
.unwrap();
// Genesis (block 1): clock-only.
let genesis = produce_dummy_block(1, None, vec![]);
let mut prev_hash = genesis.header.hash;
assert!(matches!(
store.accept_block(&genesis, Slot::from(0)).await.unwrap(),
AcceptOutcome::Applied
));
for i in 0..10_u128 {
let tx = common::test_utils::create_transaction_native_token_transfer(
from, i, to, 10, &sign_key,
);
let block_id = u64::try_from(i + 1).unwrap();
let next_block = common::test_utils::produce_dummy_block(block_id, prev_hash, vec![tx]);
prev_hash = Some(next_block.header.hash);
storage
.put_block(
next_block,
HeaderId::from([u8::try_from(i + 1).unwrap(); 32]),
)
.await
.unwrap();
// Blocks 2..=11: one native transfer of 10 each (nonces 0..=9).
for i in 0..10_u64 {
let tx = create_transaction_native_token_transfer(from, i.into(), to, 10, &sign_key);
let block = produce_dummy_block(i + 2, Some(prev_hash), vec![tx]);
prev_hash = block.header.hash;
assert!(matches!(
store.accept_block(&block, Slot::from(0)).await.unwrap(),
AcceptOutcome::Applied
));
}
let acc1_val = storage.account_current_state(&from).await.unwrap();
let acc2_val = storage.account_current_state(&to).await.unwrap();
assert_eq!(acc1_val.balance, 9900);
assert_eq!(acc2_val.balance, 20100);
assert_eq!(
store.account_current_state(&from).await.unwrap().balance,
9900
);
assert_eq!(
store.account_current_state(&to).await.unwrap().balance,
20100
);
// Tip advanced to the last applied block; a clean run leaves no stall.
assert_eq!(store.get_last_block_id().unwrap(), Some(11));
assert!(store.get_stall_reason().unwrap().is_none());
}
#[tokio::test]
async fn account_state_at_block() {
let TestFixture {
storage,
from,
to,
_home,
} = store_with_transfer_blocks(10, None).await;
async fn account_state_at_block_reflects_history() {
let home = tempdir().unwrap();
let store = IndexerStore::open_db(home.as_ref()).unwrap();
let acc1_at_1 = storage.account_state_at_block(&from, 1).unwrap();
let acc2_at_1 = storage.account_state_at_block(&to, 1).unwrap();
assert_eq!(acc1_at_1.balance, 9990);
assert_eq!(acc2_at_1.balance, 20010);
let initial_accounts = initial_pub_accounts_private_keys();
let from = initial_accounts[0].account_id;
let to = initial_accounts[1].account_id;
let sign_key = initial_accounts[0].pub_sign_key.clone();
let acc1_at_5 = storage.account_state_at_block(&from, 5).unwrap();
let acc2_at_5 = storage.account_state_at_block(&to, 5).unwrap();
assert_eq!(acc1_at_5.balance, 9950);
assert_eq!(acc2_at_5.balance, 20050);
let genesis = produce_dummy_block(1, None, vec![]);
let mut prev_hash = genesis.header.hash;
store.accept_block(&genesis, Slot::from(0)).await.unwrap();
let acc1_at_9 = storage.account_state_at_block(&from, 9).unwrap();
let acc2_at_9 = storage.account_state_at_block(&to, 9).unwrap();
assert_eq!(acc1_at_9.balance, 9910);
assert_eq!(acc2_at_9.balance, 20090);
for i in 0..10_u64 {
let tx = create_transaction_native_token_transfer(from, i.into(), to, 10, &sign_key);
let block = produce_dummy_block(i + 2, Some(prev_hash), vec![tx]);
prev_hash = block.header.hash;
store.accept_block(&block, Slot::from(0)).await.unwrap();
}
// State at block N is inclusive of block N.
// Block 1 (genesis, clock-only): no transfers yet.
assert_eq!(
store.account_state_at_block(&from, 1).unwrap().balance,
10000
);
assert_eq!(store.account_state_at_block(&to, 1).unwrap().balance, 20000);
// Through block 5: 4 transfers applied (blocks 2..=5).
assert_eq!(
store.account_state_at_block(&from, 5).unwrap().balance,
9960
);
assert_eq!(store.account_state_at_block(&to, 5).unwrap().balance, 20040);
// Through block 9: 8 transfers applied (blocks 2..=9).
assert_eq!(
store.account_state_at_block(&from, 9).unwrap().balance,
9920
);
assert_eq!(store.account_state_at_block(&to, 9).unwrap().balance, 20080);
}
}
#[cfg(test)]
mod accept_tests {
use common::{HashType, block::HashableBlockData, test_utils::produce_dummy_block};
use super::*;
use crate::ingest_error::BlockIngestError;
fn signing_key() -> lee::PrivateKey {
lee::PrivateKey::try_new([7_u8; 32]).expect("valid key")
}
// A block with a correct hash but empty body — enough to exercise the
// acceptance checks (id/link/hash), which run before any state application.
fn valid_hash_block(block_id: u64, prev: HashType) -> common::block::Block {
HashableBlockData {
block_id,
prev_block_hash: prev,
timestamp: 0,
transactions: vec![],
}
.into_pending_block(&signing_key())
}
#[tokio::test]
async fn non_genesis_first_block_parks_with_unexpected_id() {
let dir = tempfile::tempdir().expect("tempdir");
let store = IndexerStore::open_db(dir.path()).expect("open store");
let block = valid_hash_block(2, HashType([0_u8; 32]));
let outcome = store
.accept_block(&block, Slot::from(0))
.await
.expect("accept");
assert!(matches!(
outcome,
AcceptOutcome::Parked(BlockIngestError::UnexpectedBlockId {
expected: 1,
got: 2
})
));
let stall = store.get_stall_reason().expect("get").expect("present");
assert_eq!(stall.block_id, Some(2));
assert_eq!(stall.orphans_since, 0);
}
#[tokio::test]
async fn hash_mismatch_parks() {
let dir = tempfile::tempdir().expect("tempdir");
let store = IndexerStore::open_db(dir.path()).expect("open store");
let mut block = valid_hash_block(1, HashType([0_u8; 32]));
block.header.timestamp = 999; // invalidates the stored hash
let outcome = store
.accept_block(&block, Slot::from(0))
.await
.expect("accept");
assert!(matches!(
outcome,
AcceptOutcome::Parked(BlockIngestError::HashMismatch { .. })
));
}
#[tokio::test]
async fn second_break_bumps_orphan_count_and_keeps_first() {
let dir = tempfile::tempdir().expect("tempdir");
let store = IndexerStore::open_db(dir.path()).expect("open store");
let first = valid_hash_block(2, HashType([0_u8; 32]));
store
.accept_block(&first, Slot::from(0))
.await
.expect("accept");
let second = valid_hash_block(3, HashType([0_u8; 32]));
store
.accept_block(&second, Slot::from(0))
.await
.expect("accept");
let stall = store.get_stall_reason().expect("get").expect("present");
assert_eq!(stall.block_id, Some(2), "first stall preserved");
assert_eq!(stall.orphans_since, 1, "second break counted as orphan");
}
#[tokio::test]
async fn deserialize_break_records_stall_without_header() {
let dir = tempfile::tempdir().expect("tempdir");
let store = IndexerStore::open_db(dir.path()).expect("open store");
store
.record_stall(
None,
Slot::from(0),
BlockIngestError::Deserialize("bad bytes".to_owned()),
)
.expect("record");
let stall = store.get_stall_reason().expect("get").expect("present");
assert_eq!(stall.block_id, None);
assert!(matches!(stall.error, BlockIngestError::Deserialize(_)));
}
#[tokio::test]
async fn parks_then_recovers_on_valid_continuation() {
let dir = tempfile::tempdir().expect("tempdir");
let store = IndexerStore::open_db(dir.path()).expect("open store");
// Genesis (block 1, clock-only) applies and advances the tip.
let genesis = produce_dummy_block(1, None, vec![]);
assert!(matches!(
store.accept_block(&genesis, Slot::from(0)).await.unwrap(),
AcceptOutcome::Applied
));
// A block that skips ahead (id 3 while the tip is 1) parks the indexer.
let bad = produce_dummy_block(3, Some(genesis.header.hash), vec![]);
assert!(matches!(
store.accept_block(&bad, Slot::from(0)).await.unwrap(),
AcceptOutcome::Parked(BlockIngestError::UnexpectedBlockId {
expected: 2,
got: 3
})
));
assert!(
store.get_stall_reason().unwrap().is_some(),
"indexer should be parked after the bad block"
);
assert_eq!(
store.get_last_block_id().unwrap(),
Some(1),
"validated tip must stay frozen at genesis while parked"
);
// The valid continuation (block 2 chaining on genesis) recovers the chain.
let next = produce_dummy_block(2, Some(genesis.header.hash), vec![]);
assert!(matches!(
store.accept_block(&next, Slot::from(0)).await.unwrap(),
AcceptOutcome::Applied
));
assert!(
store.get_stall_reason().unwrap().is_none(),
"stall reason must clear on recovery"
);
assert_eq!(
store.get_last_block_id().unwrap(),
Some(2),
"tip must advance to the recovered block"
);
}
#[tokio::test]
async fn accept_block_records_tip_inscription_slot() {
let dir = tempfile::tempdir().expect("tempdir");
let store = IndexerStore::open_db(dir.path()).expect("open store");
assert_eq!(store.get_tip_slot().expect("get"), None);
let genesis = produce_dummy_block(1, None, vec![]);
store
.accept_block(&genesis, Slot::from(1_000))
.await
.expect("accept");
assert_eq!(store.get_tip_slot().expect("get"), Some(Slot::from(1_000)));
let block2 = produce_dummy_block(2, Some(genesis.header.hash), vec![]);
store
.accept_block(&block2, Slot::from(1_005))
.await
.expect("accept");
assert_eq!(store.get_tip_slot().expect("get"), Some(Slot::from(1_005)));
// A parked block freezes the tip, so its slot must not advance either.
let bad = produce_dummy_block(4, Some(block2.header.hash), vec![]);
assert!(matches!(
store.accept_block(&bad, Slot::from(1_010)).await.unwrap(),
AcceptOutcome::Parked(_)
));
assert_eq!(store.get_tip_slot().expect("get"), Some(Slot::from(1_005)));
// Neither must a re-delivered old block move it.
assert!(matches!(
store
.accept_block(&genesis, Slot::from(1_015))
.await
.unwrap(),
AcceptOutcome::AlreadyApplied
));
assert_eq!(store.get_tip_slot().expect("get"), Some(Slot::from(1_005)));
}
#[tokio::test]
async fn redelivered_tip_block_is_idempotent_not_parked() {
use testnet_initial_state::initial_pub_accounts_private_keys;
let dir = tempfile::tempdir().expect("tempdir");
let store = IndexerStore::open_db(dir.path()).expect("open store");
let accounts = initial_pub_accounts_private_keys();
let from = accounts[0].account_id;
let to = accounts[1].account_id;
let sign_key = accounts[0].pub_sign_key.clone();
let genesis = produce_dummy_block(1, None, vec![]);
store
.accept_block(&genesis, Slot::from(0))
.await
.expect("accept genesis");
// Block 2: a single transfer of 10.
let tx = common::test_utils::create_transaction_native_token_transfer(
from, 0, to, 10, &sign_key,
);
let block = produce_dummy_block(2, Some(genesis.header.hash), vec![tx]);
assert!(matches!(
store.accept_block(&block, Slot::from(0)).await.unwrap(),
AcceptOutcome::Applied
));
let balance_after = store.account_current_state(&from).await.unwrap().balance;
// Re-deliver the exact same block: idempotent skip, no state change, no park.
assert!(matches!(
store.accept_block(&block, Slot::from(0)).await.unwrap(),
AcceptOutcome::AlreadyApplied
));
assert_eq!(
store.account_current_state(&from).await.unwrap().balance,
balance_after,
"re-delivered block must not be applied twice"
);
assert_eq!(
store.get_last_block_id().unwrap(),
Some(2),
"tip must stay at the already-applied block"
);
assert!(
store.get_stall_reason().unwrap().is_none(),
"a benign duplicate must not park the indexer"
);
}
#[tokio::test]
async fn redelivered_block_below_tip_is_idempotent_not_parked() {
use testnet_initial_state::initial_pub_accounts_private_keys;
let dir = tempfile::tempdir().expect("tempdir");
let store = IndexerStore::open_db(dir.path()).expect("open store");
let accounts = initial_pub_accounts_private_keys();
let from = accounts[0].account_id;
let to = accounts[1].account_id;
let sign_key = accounts[0].pub_sign_key.clone();
// Build a short chain: genesis (1) -> block 2 -> block 3, so the tip is 3.
let genesis = produce_dummy_block(1, None, vec![]);
store
.accept_block(&genesis, Slot::from(0))
.await
.expect("accept genesis");
let tx2 = common::test_utils::create_transaction_native_token_transfer(
from, 0, to, 10, &sign_key,
);
let block2 = produce_dummy_block(2, Some(genesis.header.hash), vec![tx2]);
assert!(matches!(
store.accept_block(&block2, Slot::from(0)).await.unwrap(),
AcceptOutcome::Applied
));
let tx3 = common::test_utils::create_transaction_native_token_transfer(
from, 1, to, 10, &sign_key,
);
let block3 = produce_dummy_block(3, Some(block2.header.hash), vec![tx3]);
assert!(matches!(
store.accept_block(&block3, Slot::from(0)).await.unwrap(),
AcceptOutcome::Applied
));
let balance_after = store.account_current_state(&from).await.unwrap().balance;
// Re-deliver block 2 (id below the tip): a re-delivery, not a divergence.
assert!(matches!(
store.accept_block(&block2, Slot::from(0)).await.unwrap(),
AcceptOutcome::AlreadyApplied
));
assert_eq!(
store.account_current_state(&from).await.unwrap().balance,
balance_after,
"re-delivered block below the tip must not be applied again"
);
assert_eq!(
store.get_last_block_id().unwrap(),
Some(3),
"tip must stay at the current head"
);
assert!(
store.get_stall_reason().unwrap().is_none(),
"a benign re-delivery must not park the indexer"
);
}
}

548
lez/indexer/core/src/chain_consistency.rs Normal file
View File

@ -0,0 +1,548 @@
//! Startup check that the local store still belongs to the chain the
//! connected channel serves.
use anyhow::Result;
use common::{HashType, block::Block};
use futures::StreamExt as _;
use log::warn;
use logos_blockchain_zone_sdk::{Slot, ZoneMessage, adapter::Node as _};
use crate::IndexerCore;
/// Upper bound on the channel reads of the startup consistency check.
const CHANNEL_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(60);
/// Result of comparing the indexer's stored chain against the channel.
pub enum ChainConsistency {
/// Channel still serves our anchor block (the stored tip position, or the
/// parked block while stalled).
Consistent,
/// We could not determine the outcome due to one of:
///
/// - cold store (no anchor to compare at)
/// - the channel served only blocks newer than the anchor
/// - or the channel read was inconclusive (timeout / error / empty stream)
///
/// NOTE: None of these prove a reset, so the caller proceeds.
/// A genuine divergence is still caught later when the ingest loop tries to apply and parks.
Inconclusive,
/// Positive evidence that the channel is a different chain than the store.
///
/// Details in [`ChainMismatch`], and impl's Display trait.
Inconsistent(ChainMismatch),
}
/// The evidence behind a [`ChainConsistency::Inconsistent`].
pub enum ChainMismatch {
/// The channel serves a different block at the anchor's id.
Block {
ours: (u64, HashType),
channel: (u64, HashType),
},
/// The channel serves a block at/below the anchor's id past the anchor
/// slot; on the same chain those ids live at earlier slots.
ReinscribedBlock {
channel: (u64, HashType),
slot: Slot,
anchor_slot: Slot,
},
/// The channel has content past the anchor slot but no longer the
/// inscription we anchored on.
AnchorSlotChanged { anchor_slot: Slot },
/// The channel does not exist on the connected chain.
ChannelMissing,
/// The channel's history ends before the anchor slot.
ChannelBehindAnchor { tip_slot: Slot, anchor_slot: Slot },
}
impl std::fmt::Display for ChainMismatch {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::Block { ours, channel } => write!(
f,
"stored block {} {} != channel block {} {}",
ours.0, ours.1, channel.0, channel.1
),
Self::ReinscribedBlock {
channel,
slot,
anchor_slot,
} => write!(
f,
"channel re-serves block {} {} at slot {} past our anchor slot {}",
channel.0,
channel.1,
slot.into_inner(),
anchor_slot.into_inner()
),
Self::AnchorSlotChanged { anchor_slot } => write!(
f,
"channel content at slot {} no longer includes the inscription we parked on",
anchor_slot.into_inner()
),
Self::ChannelMissing => write!(f, "channel does not exist on the connected chain"),
Self::ChannelBehindAnchor {
tip_slot,
anchor_slot,
} => write!(
f,
"channel tip slot {} is behind our anchor slot {}",
tip_slot.into_inner(),
anchor_slot.into_inner()
),
}
}
}
/// A block that must still be inscribed at `slot` if the channel is the chain
/// the store was built from: the tip at the read cursor, or the recorded
/// parked block while stalled.
struct Anchor {
slot: Slot,
/// The anchor block's `(id, hash)`.
///
/// `None` when parked on an undeserializable inscription (no header was recorded).
block: Option<(u64, HashType)>,
}
impl Anchor {
/// Probes a channel message read at/after the anchor slot.
/// See [`IndexerCore::verify_chain_at_anchor`].
pub fn probe_anchor_slot(&self, msg: &ZoneMessage, slot: Slot) -> AnchorProbe {
if slot < self.slot {
return AnchorProbe::KeepLooking;
}
let Some((anchor_id, anchor_hash)) = self.block else {
// Anchored on an undeserializable inscription: any message still
// present at that slot means the history is intact.
return if slot == self.slot {
AnchorProbe::SameChain
} else {
AnchorProbe::Mismatch(ChainMismatch::AnchorSlotChanged {
anchor_slot: self.slot,
})
};
};
let ZoneMessage::Block(zone_block) = msg else {
return AnchorProbe::KeepLooking;
};
let Ok(block) = borsh::from_slice::<Block>(&zone_block.data) else {
return AnchorProbe::KeepLooking;
};
let (id, hash) = (block.header.block_id, block.header.hash);
if id == anchor_id {
return if hash == anchor_hash {
AnchorProbe::SameChain
} else {
AnchorProbe::Mismatch(ChainMismatch::Block {
ours: (anchor_id, anchor_hash),
channel: (id, hash),
})
};
}
if id > anchor_id {
return AnchorProbe::Bail;
}
if slot == self.slot {
// Older ids can share the anchor's slot on the same chain.
return AnchorProbe::KeepLooking;
}
// An id below the anchor served past the anchor slot is impossible on the
// same chain, even if the content is identical (deterministic genesis).
AnchorProbe::Mismatch(ChainMismatch::ReinscribedBlock {
channel: (id, hash),
slot,
anchor_slot: self.slot,
})
}
}
/// What a single channel message tells the anchored consistency check.
enum AnchorProbe {
/// The anchor is still in place: same chain.
SameChain,
Mismatch(ChainMismatch),
/// Only newer ids past the anchor: plausible on the same chain, so stop
/// scanning without a verdict.
Bail,
KeepLooking,
}
#[expect(
clippy::multiple_inherent_impl,
reason = "split for clarity & isolation of relevant code"
)]
impl IndexerCore {
/// Verifies whether the channel still serves the same chain the store was built from.
/// This may change frequently during development where we reset the chain from time to
/// time in devnet/testnet, but we do not expect [`ChainConsistency::Inconsistent`] in
/// production.
///
/// To compare the chains, we use an [`Anchor`] block that is either the parked L2 block
/// while stalled, or the tip L2 block at its own inscription L1 slot.
pub(crate) async fn verify_chain_consistency(&self) -> Result<ChainConsistency> {
let Some(anchor) = self.get_startup_anchor()? else {
// empty or cold store: nothing to compare
return Ok(ChainConsistency::Inconclusive);
};
self.verify_chain_at_anchor(&anchor).await
}
/// Builds the anchor for the startup check.
///
/// - If stalled, returns the recorded _parked_ block
/// - If not stalled, returns the validated tip at its _own_ inscription slot.
/// - If the store is empty, returns `None`.
fn get_startup_anchor(&self) -> Result<Option<Anchor>> {
if let Some(stall) = self.store.get_stall_reason()? {
return Ok(Some(Anchor {
slot: stall.l1_slot,
block: stall.block_id.zip(stall.block_hash),
}));
}
// not stalled, so anchor on the tip at its own inscription slot
let Some(slot) = self.store.get_tip_slot()?.or(self.store.get_zone_cursor()?) else {
return Ok(None);
};
let Some(tip_id) = self.store.get_last_block_id()? else {
return Ok(None);
};
let Some(tip) = self.store.get_block_at_id(tip_id)? else {
return Ok(None);
};
Ok(Some(Anchor {
slot,
block: Some((tip_id, tip.header.hash)),
}))
}
/// Verifies the channel still carries the anchor block at its slot.
///
/// The anchor was finalized at `anchor.slot`, so the same chain must still
/// serve it there, while a reset chain re-inscribes its content only at
/// later wall-clock slots.
///
/// Only positive evidence of a different chain yields `Inconsistent`.
/// Absence of data stays `Inconclusive`.
async fn verify_chain_at_anchor(&self, anchor: &Anchor) -> Result<ChainConsistency> {
match self.node.channel_state(self.config.channel_id).await {
Ok(state) => {
if let Some(mismatch) = frontier_verdict(anchor.slot, state.map(|s| s.tip_slot)) {
return Ok(ChainConsistency::Inconsistent(mismatch));
}
}
Err(err) => {
warn!("Failed to read channel state for the consistency check: {err:#}");
}
}
// `next_messages` is exclusive, so `slot - 1` includes the anchor slot.
let Some(from_slot) = anchor.slot.into_inner().checked_sub(1) else {
return Ok(ChainConsistency::Inconclusive);
};
let scan = async {
let stream = self
.zone_indexer
.next_messages(Some(Slot::from(from_slot)))
.await?;
let mut stream = std::pin::pin!(stream);
while let Some((msg, slot)) = stream.next().await {
match anchor.probe_anchor_slot(&msg, slot) {
AnchorProbe::SameChain => return Ok(Some(ChainConsistency::Consistent)),
AnchorProbe::Mismatch(mismatch) => {
return Ok(Some(ChainConsistency::Inconsistent(mismatch)));
}
AnchorProbe::Bail => return Ok(Some(ChainConsistency::Inconclusive)),
AnchorProbe::KeepLooking => { /* dont do anything */ }
}
}
Ok::<_, anyhow::Error>(None)
};
match tokio::time::timeout(CHANNEL_READ_TIMEOUT, scan).await {
Ok(Ok(Some(outcome))) => Ok(outcome),
Ok(Ok(None)) => Ok(ChainConsistency::Inconclusive),
Ok(Err(err)) => {
warn!(
"Failed to read the anchor slot for the consistency check; proceeding: {err:#}"
);
Ok(ChainConsistency::Inconclusive)
}
Err(_elapsed) => {
warn!("Timed out reading the anchor slot for the consistency check; proceeding");
Ok(ChainConsistency::Inconclusive)
}
}
}
}
/// Checks the channel frontier against the anchor slot.
///
/// The anchor block was finalized at `anchor_slot`, so on the same chain the
/// channel tip can never be behind it, and the channel must exist.
fn frontier_verdict(anchor_slot: Slot, channel_tip_slot: Option<Slot>) -> Option<ChainMismatch> {
match channel_tip_slot {
None => Some(ChainMismatch::ChannelMissing),
Some(tip_slot) if tip_slot < anchor_slot => Some(ChainMismatch::ChannelBehindAnchor {
tip_slot,
anchor_slot,
}),
Some(_) => None,
}
}
#[cfg(test)]
mod tests {
use std::time::Duration;
use common::block::HashableBlockData;
use logos_blockchain_core::mantle::ops::channel::{MsgId, inscribe::Inscription};
use logos_blockchain_zone_sdk::ZoneBlock;
use super::*;
use crate::{
BlockIngestError,
block_store::AcceptOutcome,
config::{ChannelId, ClientConfig, IndexerConfig},
};
fn unreachable_core(dir: &std::path::Path) -> IndexerCore {
let config = IndexerConfig {
consensus_info_polling_interval: Duration::from_secs(1),
bedrock_config: ClientConfig {
addr: "http://localhost:1".parse().expect("url"),
auth: None,
},
channel_id: ChannelId::from([1; 32]),
allow_chain_reset: false,
};
IndexerCore::open(config, dir).expect("open core")
}
fn test_block(block_id: u64, timestamp: u64) -> Block {
HashableBlockData {
block_id,
prev_block_hash: HashType([0; 32]),
timestamp,
transactions: vec![],
}
.into_pending_block(&lee::PrivateKey::try_new([7; 32]).expect("valid key"))
}
fn block_msg(block: &Block) -> ZoneMessage {
let bytes = borsh::to_vec(block).expect("serialize");
ZoneMessage::Block(ZoneBlock {
id: MsgId::from([0_u8; 32]),
data: Inscription::try_from(bytes.as_slice()).expect("inscription"),
})
}
fn anchor_for(block: &Block, slot: Slot) -> Anchor {
Anchor {
slot,
block: Some((block.header.block_id, block.header.hash)),
}
}
#[tokio::test]
async fn cold_store_is_inconclusive() {
// An empty store has no cursor, so there is nothing to compare: the check
// must be Inconclusive (not Consistent), and it returns before any L1 read.
let dir = tempfile::tempdir().expect("tempdir");
let core = unreachable_core(dir.path());
assert!(matches!(
core.verify_chain_consistency().await.expect("verify"),
ChainConsistency::Inconclusive
));
}
#[tokio::test]
async fn parked_store_with_unreachable_node_is_inconclusive() {
// Network failure is not evidence of a reset: a parked store must stay
// parked (Inconclusive), not error out or trip the wipe path.
let dir = tempfile::tempdir().expect("tempdir");
let core = unreachable_core(dir.path());
let parked = test_block(5, 42);
core.store
.record_stall(
Some(&parked.header),
Slot::from(1_000),
BlockIngestError::EmptyBlock,
)
.expect("record stall");
assert!(matches!(
core.verify_chain_consistency().await.expect("verify"),
ChainConsistency::Inconclusive
));
}
#[tokio::test]
async fn caught_up_store_with_unreachable_node_is_inconclusive() {
let dir = tempfile::tempdir().expect("tempdir");
let core = unreachable_core(dir.path());
let genesis = common::test_utils::produce_dummy_block(1, None, vec![]);
assert!(matches!(
core.store
.accept_block(&genesis, Slot::from(1_000))
.await
.expect("accept"),
AcceptOutcome::Applied
));
core.store
.set_zone_cursor(&Slot::from(1_000))
.expect("set cursor");
assert!(matches!(
core.verify_chain_consistency().await.expect("verify"),
ChainConsistency::Inconclusive
));
}
#[tokio::test]
async fn startup_anchor_prefers_tip_slot_over_lagging_cursor() {
// Cursor persist failures are warn-only, so the read cursor can lag the
// tip by several blocks. The anchor must pair the tip with its own
// inscription slot; pairing it with the stale cursor would make the scan
// misread the chain's intermediate blocks as re-inscriptions.
let dir = tempfile::tempdir().expect("tempdir");
let core = unreachable_core(dir.path());
let genesis = common::test_utils::produce_dummy_block(1, None, vec![]);
core.store
.accept_block(&genesis, Slot::from(1_000))
.await
.expect("accept");
let block2 = common::test_utils::produce_dummy_block(2, Some(genesis.header.hash), vec![]);
core.store
.accept_block(&block2, Slot::from(1_005))
.await
.expect("accept");
let block3 = common::test_utils::produce_dummy_block(3, Some(block2.header.hash), vec![]);
core.store
.accept_block(&block3, Slot::from(1_010))
.await
.expect("accept");
// Cursor last persisted at the genesis slot: two blocks behind the tip.
core.store
.set_zone_cursor(&Slot::from(1_000))
.expect("set cursor");
let anchor = core.get_startup_anchor().expect("anchor").expect("present");
assert_eq!(anchor.slot, Slot::from(1_010));
assert_eq!(anchor.block, Some((3, block3.header.hash)));
}
#[test]
fn probe_finds_anchor_block_at_slot() {
let tip = test_block(5, 42);
let anchor = anchor_for(&tip, Slot::from(1_000));
assert!(matches!(
anchor.probe_anchor_slot(&block_msg(&tip), Slot::from(1_000)),
AnchorProbe::SameChain
));
}
#[test]
fn probe_flags_different_block_at_anchor_id() {
let tip = test_block(5, 42);
let anchor = anchor_for(&tip, Slot::from(1_000));
// Same id, different content (timestamp) => different hash.
let other = test_block(5, 43);
assert!(matches!(
anchor.probe_anchor_slot(&block_msg(&other), Slot::from(1_000)),
AnchorProbe::Mismatch(ChainMismatch::Block { .. })
));
}
#[test]
fn probe_flags_old_id_reinscribed_past_the_anchor_slot() {
// A reset chain re-inscribes from genesis at later slots. Even if the
// content is byte-identical (deterministic genesis), an id at/below
// the anchor past the anchor slot is impossible on the same chain.
let tip = test_block(5, 42);
let anchor = anchor_for(&tip, Slot::from(1_000));
let genesis = test_block(1, 0);
assert!(matches!(
anchor.probe_anchor_slot(&block_msg(&genesis), Slot::from(1_001)),
AnchorProbe::Mismatch(ChainMismatch::ReinscribedBlock { .. })
));
}
#[test]
fn probe_skips_older_blocks_sharing_the_anchor_slot() {
let tip = test_block(5, 42);
let anchor = anchor_for(&tip, Slot::from(1_000));
let earlier = test_block(4, 41);
assert!(matches!(
anchor.probe_anchor_slot(&block_msg(&earlier), Slot::from(1_000)),
AnchorProbe::KeepLooking
));
}
#[test]
fn probe_bails_on_newer_ids_past_the_anchor() {
// Blocks newer than the anchor are plausible on the same chain (e.g.
// published while we were down), so they must never count as evidence.
let tip = test_block(5, 42);
let anchor = anchor_for(&tip, Slot::from(1_000));
let newer = test_block(6, 43);
assert!(matches!(
anchor.probe_anchor_slot(&block_msg(&newer), Slot::from(1_001)),
AnchorProbe::Bail
));
}
#[test]
fn probe_skips_undeserializable_inscriptions() {
let tip = test_block(5, 42);
let anchor = anchor_for(&tip, Slot::from(1_000));
let garbage = ZoneMessage::Block(ZoneBlock {
id: MsgId::from([0_u8; 32]),
data: Inscription::try_from(&[1_u8, 2, 3][..]).expect("inscription"),
});
assert!(matches!(
anchor.probe_anchor_slot(&garbage, Slot::from(1_000)),
AnchorProbe::KeepLooking
));
}
#[test]
fn probe_accepts_any_message_for_a_headerless_anchor() {
// A deserialize park records no header: any message still present at
// the anchor slot means the history is intact.
let anchor = Anchor {
slot: Slot::from(1_000),
block: None,
};
let garbage = ZoneMessage::Block(ZoneBlock {
id: MsgId::from([0_u8; 32]),
data: Inscription::try_from(&[1_u8, 2, 3][..]).expect("inscription"),
});
assert!(matches!(
anchor.probe_anchor_slot(&garbage, Slot::from(1_000)),
AnchorProbe::SameChain
));
assert!(matches!(
anchor.probe_anchor_slot(&garbage, Slot::from(1_001)),
AnchorProbe::Mismatch(ChainMismatch::AnchorSlotChanged { .. })
));
}
#[test]
fn frontier_flags_missing_channel_and_short_history() {
assert!(matches!(
frontier_verdict(Slot::from(1_000), None),
Some(ChainMismatch::ChannelMissing)
));
assert!(matches!(
frontier_verdict(Slot::from(1_000), Some(Slot::from(999))),
Some(ChainMismatch::ChannelBehindAnchor { .. })
));
assert!(frontier_verdict(Slot::from(1_000), Some(Slot::from(1_000))).is_none());
assert!(frontier_verdict(Slot::from(1_000), Some(Slot::from(2_000))).is_none());
}
}

7
lez/indexer/core/src/config.rs
View File

@ -20,6 +20,13 @@ pub struct IndexerConfig {
pub consensus_info_polling_interval: Duration,
pub bedrock_config: ClientConfig,
pub channel_id: ChannelId,
/// Whether to wipe the indexer store and re-index from scratch when the startup
/// chain-identity check finds the channel serving a different block than the one
/// stored at the same id.
///
/// Defaults to `false`: on mismatch the indexer refuses to start.
#[serde(default)]
pub allow_chain_reset: bool,
}
impl IndexerConfig {

67
lez/indexer/core/src/ingest_error.rs Normal file
View File

@ -0,0 +1,67 @@
use common::HashType;
use serde::{Deserialize, Serialize};
/// Why the indexer could not apply an L2 block from the channel.
///
/// Persisted in `RocksDB`, so every variant must have the following
/// traits: `Clone + Serialize + Deserialize`.
#[derive(Debug, Clone, Serialize, Deserialize, thiserror::Error)]
pub enum BlockIngestError {
#[error("Failed to deserialize L2 block: {0}")]
/// Here we store the error string that is derived from [`borsh::from_slice`]'s [`Err`].
Deserialize(String),
#[error("Unexpected block id: expected {expected}, got {got}")]
UnexpectedBlockId { expected: u64, got: u64 },
#[error("Broken chain link: expected prev {expected_prev}, got {got_prev}")]
BrokenChainLink {
expected_prev: HashType,
got_prev: HashType,
},
#[error("Block hash mismatch: computed {computed}, header {header}")]
HashMismatch {
computed: HashType,
header: HashType,
},
#[error("Block has no transactions")]
EmptyBlock,
#[error("Last transaction must be the public clock invocation for the block timestamp")]
InvalidClockTransaction,
#[error("Genesis block must contain only public transactions")]
NonPublicGenesisTransaction,
#[error("State transition failed at transaction {tx_index}: {reason}")]
StateTransition {
/// Index of the failing transaction within the block body.
tx_index: u64,
/// Reason string from `lee::Error` to `anyhow::Error` to `{:#}`.
///
/// This is required because `lee::Error` is not `Clone + Serialize + Deserialize`, so we
/// cannot store it directly.
reason: String,
},
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn serializes_and_round_trips_externally_tagged() {
let err = BlockIngestError::UnexpectedBlockId {
expected: 5,
got: 7,
};
let value = serde_json::to_value(&err).expect("serialize");
assert_eq!(
value,
serde_json::json!({ "UnexpectedBlockId": { "expected": 5, "got": 7 } })
);
let back: BlockIngestError = serde_json::from_value(value).expect("deserialize");
assert!(matches!(
back,
BlockIngestError::UnexpectedBlockId {
expected: 5,
got: 7
}
));
}
}

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

@ -3,27 +3,33 @@ use std::{path::Path, sync::Arc};
use anyhow::Result;
use arc_swap::ArcSwap;
use common::block::Block;
// ToDo: Remove after testnet
// TODO: Remove after testnet
use futures::StreamExt as _;
pub use ingest_error::BlockIngestError;
use log::{error, info, warn};
use logos_blockchain_core::header::HeaderId;
use logos_blockchain_zone_sdk::{
CommonHttpClient, ZoneMessage, adapter::NodeHttpClient, indexer::ZoneIndexer,
CommonHttpClient, Slot, ZoneMessage, adapter::NodeHttpClient, indexer::ZoneIndexer,
};
pub use stall_reason::StallReason;
use crate::{
block_store::IndexerStore,
block_store::{AcceptOutcome, IndexerStore},
chain_consistency::ChainConsistency,
config::IndexerConfig,
status::{IndexerStatus, IndexerSyncStatus},
};
pub mod block_store;
pub mod chain_consistency;
pub mod config;
pub mod ingest_error;
pub mod stall_reason;
pub mod status;
#[derive(Clone)]
pub struct IndexerCore {
pub zone_indexer: Arc<ZoneIndexer<NodeHttpClient>>,
/// Direct node handle for queries outside `ZoneIndexer`'s streaming API.
pub node: NodeHttpClient,
pub config: IndexerConfig,
pub store: IndexerStore,
/// Live ingestion status; updated by the ingest stream, read by `status`.
@ -31,7 +37,42 @@ 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.verify_chain_consistency().await? {
// `Inconclusive` is deliberately treated the same as `Consistent`.
//
// We could not prove a reset, so proceed from the cursor without wiping
// a possibly-valid store. A genuinely divergent chain is still caught
// later when the ingest loop tries to apply and parks.
ChainConsistency::Consistent | ChainConsistency::Inconclusive => Ok(core),
ChainConsistency::Inconsistent(mismatch) if config.allow_chain_reset => {
warn!(
"Chain reset detected ({mismatch}). Wiping indexer store at {} and \
re-indexing.",
home.display()
);
drop(core); // sole owner before the ingest task is spawned → closes the DB
storage::indexer::RocksDBIO::destroy(&home)?;
Self::open(config, storage_dir)
}
ChainConsistency::Inconsistent(mismatch) => Err(anyhow::anyhow!(
"Indexer store at {} holds a different chain than the channel now serves \
({mismatch}). Delete the indexer storage directory, point at a fresh one, or \
set `allow_chain_reset` in the indexer config.",
home.display()
)),
}
}
/// Opens the store and builds the core without the chain-identity check.
fn open(config: IndexerConfig, storage_dir: &Path) -> Result<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));
@ -41,10 +82,11 @@ impl IndexerCore {
CommonHttpClient::new(basic_auth),
config.bedrock_config.addr.clone(),
);
let zone_indexer = ZoneIndexer::new(config.channel_id, node);
let zone_indexer = ZoneIndexer::new(config.channel_id, node.clone());
Ok(Self {
zone_indexer: Arc::new(zone_indexer),
node,
config,
store: IndexerStore::open_db(&home)?,
status: Arc::new(ArcSwap::from_pointee(IndexerSyncStatus::starting())),
@ -58,10 +100,27 @@ impl IndexerCore {
#[must_use]
pub fn status(&self) -> IndexerStatus {
let sync = IndexerSyncStatus::clone(&self.status.load());
let indexed_block_id = self.store.get_last_block_id().ok().flatten();
// Log-and-fall-back rather than collapsing a store error into the same
// `None` as "legitimately absent": a DB read failure must not silently
// masquerade as "no tip yet" / "no stall recorded" in the snapshot.
let indexed_block_id = match self.store.get_last_block_id() {
Ok(id) => id,
Err(err) => {
warn!("Failed to read last indexed block id for status: {err:#}");
None
}
};
let stall_reason = match self.store.get_stall_reason() {
Ok(reason) => reason,
Err(err) => {
warn!("Failed to read stall reason for status: {err:#}");
None
}
};
IndexerStatus {
sync,
indexed_block_id,
stall_reason,
}
}
@ -70,6 +129,35 @@ 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: std::io::Error) {
let error = anyhow::Error::new(error);
// use `:#` to get the entire error chain
let reason = format!("{error:#}");
error!("Failed to deserialize L2 block from zone-sdk: {reason}");
if let Err(err) =
self.store
.record_stall(None, slot, BlockIngestError::Deserialize(reason.clone()))
{
warn!("Failed to record stall reason: {err:#}");
}
self.set_status(IndexerSyncStatus::stalled(format!(
"failed to deserialize L2 block: {reason}"
)));
}
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
@ -90,8 +178,6 @@ impl IndexerCore {
let stream = match self.zone_indexer.next_messages(cursor).await {
Ok(s) => s,
Err(err) => {
// `next_messages` reads L1 consensus info internally, so
// this also covers an unreachable/misconfigured L1 node.
error!("Failed to start zone-sdk next_messages stream: {err}");
self.set_status(IndexerSyncStatus::error(format!(
"cannot reach L1 / read channel: {err}"
@ -102,11 +188,8 @@ impl IndexerCore {
};
let mut stream = std::pin::pin!(stream);
// Flip to Syncing on the first message of this cycle (not merely on
// a successful poll) so the steady-state CaughtUp status doesn't
// flicker. Until then the state stays Starting (cold-start scan of
// empty L1 history) or CaughtUp (idle).
let mut announced_syncing = false;
let mut had_cycle_error = false;
while let Some((msg, slot)) = stream.next().await {
if !announced_syncing {
@ -116,46 +199,74 @@ impl IndexerCore {
let zone_block = match msg {
ZoneMessage::Block(b) => b,
// Non-block messages don't carry a cursor position; the
// next ZoneBlock advances past them implicitly.
// FIXME: will be handled in prep of decentralized sequencers
ZoneMessage::Deposit(_) | ZoneMessage::Withdraw(_) => continue,
};
let block: Block = match borsh::from_slice(&zone_block.data) {
Ok(b) => b,
Err(e) => {
error!("Failed to deserialize L2 block from zone-sdk: {e}");
// Advance past the broken inscription so we don't
// re-process it on restart.
cursor = Some(slot);
if let Err(err) = self.store.set_zone_cursor(&slot) {
warn!("Failed to persist indexer cursor: {err:#}");
}
Err(error) => {
self.park_undeserializable(slot, error);
// L1 proceeds regardless
self.advance_cursor(&mut cursor, slot);
continue;
}
};
info!("Indexed L2 block {}", block.header.block_id);
// TODO: Remove l1_header placeholder once storage layer
// no longer requires it. Zone-sdk handles L1 tracking internally.
let placeholder_l1_header = HeaderId::from([0_u8; 32]);
if let Err(err) = self.store.put_block(block.clone(), placeholder_l1_header).await {
error!("Failed to store block {}: {err:#}", block.header.block_id);
match self.store.accept_block(&block, slot).await {
Ok(AcceptOutcome::Applied) => {
info!("Indexed L2 block {}", block.header.block_id);
self.set_status(IndexerSyncStatus::syncing());
self.advance_cursor(&mut cursor, slot);
yield Ok(block);
}
Ok(AcceptOutcome::AlreadyApplied) => {
info!(
"Skipping already-applied block {}",
block.header.block_id
);
self.advance_cursor(&mut cursor, slot);
}
Ok(AcceptOutcome::Parked(ingest_err)) => {
error!(
"Parked at block {}: {ingest_err}",
block.header.block_id
);
self.set_status(IndexerSyncStatus::stalled(ingest_err.to_string()));
// L1 proceeds regardless
self.advance_cursor(&mut cursor, slot);
}
Err(err) => {
// Infrastructure error (DB read/write), not a bad block.
// will re-poll from the same cursor next cycle.
error!(
"Store error applying block {}: {err:#}",
block.header.block_id
);
self.set_status(IndexerSyncStatus::error(format!(
"store error: {err:#}"
)));
had_cycle_error = true;
break;
}
}
cursor = Some(slot);
if let Err(err) = self.store.set_zone_cursor(&slot) {
warn!("Failed to persist indexer cursor: {err:#}");
}
yield Ok(block);
}
// Stream drained: caught up to LIB as of this cycle. Clears any
// prior error (e.g. a transient L1 disconnect that left no
// backlog, so the `Syncing` branch above never ran). Sleep then
// poll again.
self.set_status(IndexerSyncStatus::caught_up());
if had_cycle_error {
tokio::time::sleep(poll_interval).await;
continue;
}
// Stream drained. Stay Stalled if parked; otherwise we are caught up.
// A store error here must not be collapsed to "no stall recorded":
// that would wrongly flip us to caught-up, so we log and hold state.
match self.store.get_stall_reason() {
Ok(None) => self.set_status(IndexerSyncStatus::caught_up()),
Ok(Some(_)) => {}
Err(err) => {
warn!("Failed to read stall reason after draining stream; not marking caught up: {err:#}");
}
}
tokio::time::sleep(poll_interval).await;
}
}

25
lez/indexer/core/src/stall_reason.rs Normal file
View File

@ -0,0 +1,25 @@
use common::HashType;
use logos_blockchain_zone_sdk::Slot;
use serde::{Deserialize, Serialize};
use crate::ingest_error::BlockIngestError;
/// Diagnostic record of the first block that broke the L2 chain.
///
/// The block-derived fields are `None` for a deserialize break (no header was
/// ever parsed). `l1_slot` is the L1 slot the breaking inscription was read at.
/// `first_seen` is the breaking block's L2 timestamp (`None` for a deserialize break).
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct StallReason {
pub block_id: Option<u64>,
pub block_hash: Option<HashType>,
pub prev_block_hash: Option<HashType>,
pub l1_slot: Slot,
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 break, but for now we
/// just count them.
pub orphans_since: u64,
}

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

@ -1,9 +1,10 @@
use serde::Serialize;
use crate::stall_reason::StallReason;
/// Coarse lifecycle state of the indexer's ingestion loop, so a client can tell
/// "still catching up" apart from "something went wrong".
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum IndexerSyncState {
/// Booted; no ingestion cycle has run yet.
Starting,
@ -13,12 +14,14 @@ pub enum IndexerSyncState {
CaughtUp,
/// The last cycle failed (e.g. the L1 node is unreachable). See `last_error`.
Error,
/// Parked on a stall reason: the validated tip is frozen awaiting a valid
/// continuation. See `last_error` and the snapshot's `stall_reason`.
Stalled,
}
/// Live ingestion status owned by the ingest loop: the coarse `state` plus the
/// reason when it is `Error`.
#[derive(Debug, Clone, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct IndexerSyncStatus {
pub state: IndexerSyncState,
pub last_error: Option<String>,
@ -56,6 +59,15 @@ impl IndexerSyncStatus {
last_error: Some(reason),
}
}
/// Parked on a stall reason; `reason` mirrors the stall's error message.
/// The full stall is attached to the [`IndexerStatus`] snapshot.
pub(crate) const fn stalled(reason: String) -> Self {
Self {
state: IndexerSyncState::Stalled,
last_error: Some(reason),
}
}
}
/// Full status snapshot returned to callers (FFI/RPC): the live [`IndexerSyncStatus`]
@ -64,11 +76,11 @@ impl IndexerSyncStatus {
/// The tip is tracked by the store, not the ingest loop, so it lives here on the
/// returned snapshot rather than inside the shared [`IndexerSyncStatus`].
#[derive(Debug, Clone, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct IndexerStatus {
#[serde(flatten)]
pub sync: IndexerSyncStatus,
pub indexed_block_id: Option<u64>,
pub stall_reason: Option<StallReason>,
}
#[cfg(test)]
@ -80,14 +92,16 @@ mod tests {
let status = IndexerStatus {
sync: IndexerSyncStatus::error("boom".to_owned()),
indexed_block_id: Some(7),
stall_reason: None,
};
let value = serde_json::to_value(&status).expect("serialize");
assert_eq!(
value,
serde_json::json!({
"state": "error",
"lastError": "boom",
"indexedBlockId": 7,
"state": "Error",
"last_error": "boom",
"indexed_block_id": 7,
"stall_reason": null,
})
);
}
@ -97,7 +111,36 @@ mod tests {
let value = serde_json::to_value(IndexerSyncStatus::caught_up()).expect("serialize");
assert_eq!(
value,
serde_json::json!({ "state": "caught_up", "lastError": null })
serde_json::json!({ "state": "CaughtUp", "last_error": null })
);
}
#[test]
fn stalled_status_serializes_with_stall_reason() {
use logos_blockchain_zone_sdk::Slot;
use crate::{ingest_error::BlockIngestError, stall_reason::StallReason};
let status = IndexerStatus {
sync: IndexerSyncStatus::stalled("broken chain link".to_owned()),
indexed_block_id: Some(41),
stall_reason: Some(StallReason {
block_id: Some(42),
block_hash: None,
prev_block_hash: None,
l1_slot: Slot::from(0),
error: BlockIngestError::StateTransition {
tx_index: 0,
reason: String::default(),
},
first_seen: None,
orphans_since: 2,
}),
};
let value = serde_json::to_value(&status).expect("serialize");
assert_eq!(value["state"], serde_json::json!("Stalled"));
assert_eq!(value["last_error"], serde_json::json!("broken chain link"));
assert_eq!(value["indexed_block_id"], serde_json::json!(41));
assert_eq!(value["stall_reason"]["orphans_since"], serde_json::json!(2));
}
}

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
*

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

@ -112,10 +112,12 @@ unsafe fn setup_indexer(
unsafe { Runtime::from_borrowed(caller.as_ref()) }
};
let core = IndexerCore::new(config, &storage_dir).map_err(|e| {
log::error!("Could not initialize indexer core: {e}");
OperationStatus::InitializationError
})?;
let core = runtime
.block_on(IndexerCore::new(config, &storage_dir))
.map_err(|e| {
log::error!("Could not initialize indexer core: {e}");
OperationStatus::InitializationError
})?;
// The block stream writes each parsed block into the store as a side effect
// of being polled, so we spawn a task that simply drains it. There are no

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

11
lez/indexer/service/configs/debug/indexer_config.json
View File

@ -1,7 +1,8 @@
{
"consensus_info_polling_interval": "1s",
"bedrock_config": {
"addr": "http://localhost:18080"
},
"channel_id": "0101010101010101010101010101010101010101010101010101010101010101"
"consensus_info_polling_interval": "1s",
"bedrock_config": {
"addr": "http://localhost:18080"
},
"channel_id": "0101010101010101010101010101010101010101010101010101010101010101",
"allow_chain_reset": true
}

3
lez/indexer/service/configs/docker/indexer_config.json
View File

@ -3,5 +3,6 @@
"bedrock_config": {
"addr": "http://host.docker.internal:18080"
},
"channel_id": "0101010101010101010101010101010101010101010101010101010101010101"
"channel_id": "0101010101010101010101010101010101010101010101010101010101010101",
"allow_chain_reset": true
}

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

102
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,94 @@ 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::EmptyBlock => Self::EmptyBlock,
indexer_core::BlockIngestError::InvalidClockTransaction => {
Self::InvalidClockTransaction
}
indexer_core::BlockIngestError::NonPublicGenesisTransaction => {
Self::NonPublicGenesisTransaction
}
indexer_core::BlockIngestError::StateTransition { tx_index, reason } => {
Self::StateTransition { tx_index, reason }
}
}
}
}
impl From<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),
}
}
}

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

@ -363,3 +363,73 @@ pub enum BedrockStatus {
Safe,
Finalized,
}
/// Coarse lifecycle state of the indexer's ingestion loop, so a client can tell
/// "still catching up" apart from "something went wrong".
#[derive(Debug, Copy, Clone, PartialEq, Eq, Hash, Serialize, Deserialize, JsonSchema)]
pub enum IndexerSyncState {
/// Booted; no ingestion cycle has run yet.
Starting,
/// Streaming finalized messages toward the L1 frontier.
Syncing,
/// Drained the stream up to the last finalized block; idle until new blocks finalize.
CaughtUp,
/// The last cycle failed (e.g. the L1 node is unreachable). See `last_error`.
Error,
/// Parked on a stall reason: the validated tip is frozen awaiting a valid
/// continuation. See `last_error` and `stall_reason`.
Stalled,
}
/// Why the indexer could not apply an L2 block from the channel.
#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize, JsonSchema)]
pub enum BlockIngestError {
Deserialize(String),
UnexpectedBlockId {
expected: u64,
got: u64,
},
BrokenChainLink {
expected_prev: HashType,
got_prev: HashType,
},
HashMismatch {
computed: HashType,
header: HashType,
},
EmptyBlock,
InvalidClockTransaction,
NonPublicGenesisTransaction,
StateTransition {
/// Index of the failing transaction within the block body.
tx_index: u64,
reason: String,
},
}
/// Diagnostic record of the first block that broke the L2 chain.
///
/// The block-derived fields are `None` for a deserialize break (no header was
/// ever parsed). `l1_slot` is the L1 slot the breaking inscription was read at.
#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize, JsonSchema)]
pub struct StallReason {
pub block_id: Option<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>,
}

7
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};
@ -69,6 +71,9 @@ pub trait Rpc {
limit: u64,
) -> Result<Vec<Transaction>, ErrorObjectOwned>;
#[method(name = "getStatus")]
async fn get_status(&self) -> Result<IndexerStatus, ErrorObjectOwned>;
// ToDo: expand healthcheck response into some kind of report
#[method(name = "checkHealth")]
async fn healthcheck(&self) -> Result<(), ErrorObjectOwned>;

1
lez/indexer/service/src/lib.rs
View File

@ -87,6 +87,7 @@ pub async fn run_server(
#[cfg(not(feature = "mock-responses"))]
let handle = {
let service = service::IndexerService::new(config, storage_dir)
.await
.context("Failed to initialize indexer service")?;
server.start(service.into_rpc())
};

26
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,6 +325,24 @@ impl indexer_service_rpc::RpcServer for MockIndexerService {
.collect())
}
async fn get_status(&self) -> Result<IndexerStatus, ErrorObjectOwned> {
let indexed_block_id = self
.state
.read()
.await
.blocks
.iter()
.rev()
.find(|block| block.bedrock_status == BedrockStatus::Finalized)
.map(|block| block.header.block_id);
Ok(IndexerStatus {
state: IndexerSyncState::CaughtUp,
last_error: None,
indexed_block_id,
stall_reason: None,
})
}
async fn healthcheck(&self) -> Result<(), ErrorObjectOwned> {
Ok(())
}

12
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},
@ -19,8 +21,8 @@ pub struct IndexerService {
}
impl IndexerService {
pub fn new(config: IndexerConfig, storage_dir: &Path) -> Result<Self> {
let indexer = IndexerCore::new(config, storage_dir)?;
pub async fn new(config: IndexerConfig, storage_dir: &Path) -> Result<Self> {
let indexer = IndexerCore::new(config, storage_dir).await?;
let subscription_service = SubscriptionService::spawn_new(indexer.clone());
Ok(Self {
@ -149,6 +151,10 @@ impl indexer_service_rpc::RpcServer for IndexerService {
Ok(tx_res)
}
async fn get_status(&self) -> Result<IndexerStatus, ErrorObjectOwned> {
Ok(self.indexer.status().into())
}
async fn healthcheck(&self) -> Result<(), ErrorObjectOwned> {
// Checking, that indexer can calculate last state
let _ = self

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

@ -8,8 +8,8 @@ use crate::{
indexer::{
ACC_NUM_CELL_NAME, BLOCK_HASH_CELL_NAME, BREAKPOINT_CELL_NAME, CF_ACC_META,
CF_BREAKPOINT_NAME, CF_HASH_TO_ID, CF_TX_TO_ID, DB_META_LAST_BREAKPOINT_ID,
DB_META_LAST_OBSERVED_L1_LIB_HEADER_ID_IN_DB_KEY, DB_META_ZONE_SDK_INDEXER_CURSOR_KEY,
TX_HASH_CELL_NAME,
DB_META_LAST_OBSERVED_L1_LIB_HEADER_ID_IN_DB_KEY, DB_META_STALL_REASON_KEY,
DB_META_TIP_SLOT_KEY, DB_META_ZONE_SDK_INDEXER_CURSOR_KEY, TX_HASH_CELL_NAME,
},
};
@ -212,6 +212,27 @@ impl SimpleWritableCell for AccNumTxCell {
}
}
/// The L1 inscription slot of the tip block, written atomically with the tip.
#[derive(Debug, BorshSerialize, BorshDeserialize)]
pub struct TipSlotCell(pub u64);
impl SimpleStorableCell for TipSlotCell {
type KeyParams = ();
const CELL_NAME: &'static str = DB_META_TIP_SLOT_KEY;
const CF_NAME: &'static str = CF_META_NAME;
}
impl SimpleReadableCell for TipSlotCell {}
impl SimpleWritableCell for TipSlotCell {
fn value_constructor(&self) -> DbResult<Vec<u8>> {
borsh::to_vec(&self).map_err(|err| {
DbError::borsh_cast_message(err, Some("Failed to serialize tip slot".to_owned()))
})
}
}
/// Opaque bytes for the zone-sdk indexer cursor `Option<(MsgId, Slot)>`.
/// The caller serializes via `serde_json` (neither type derives borsh).
#[derive(BorshDeserialize)]
@ -247,6 +268,40 @@ impl SimpleWritableCell for ZoneSdkIndexerCursorCellRef<'_> {
}
}
/// Opaque JSON bytes for the indexer's persisted `Option<StallReason>`.
#[derive(BorshDeserialize)]
pub struct StallReasonCellOwned(pub Vec<u8>);
impl SimpleStorableCell for StallReasonCellOwned {
type KeyParams = ();
const CELL_NAME: &'static str = DB_META_STALL_REASON_KEY;
const CF_NAME: &'static str = CF_META_NAME;
}
impl SimpleReadableCell for StallReasonCellOwned {}
#[derive(BorshSerialize)]
pub struct StallReasonCellRef<'bytes>(pub &'bytes [u8]);
impl SimpleStorableCell for StallReasonCellRef<'_> {
type KeyParams = ();
const CELL_NAME: &'static str = DB_META_STALL_REASON_KEY;
const CF_NAME: &'static str = CF_META_NAME;
}
impl SimpleWritableCell for StallReasonCellRef<'_> {
fn value_constructor(&self) -> DbResult<Vec<u8>> {
borsh::to_vec(&self).map_err(|err| {
DbError::borsh_cast_message(
err,
Some("Failed to serialize stall reason cell".to_owned()),
)
})
}
}
#[cfg(test)]
mod uniform_tests {
use crate::{

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

@ -24,6 +24,10 @@ 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<StallReason>` diagnostic record (opaque JSON bytes).
pub const DB_META_STALL_REASON_KEY: &str = "stall_reason";
/// Key base for storing the L1 inscription slot of the tip block.
pub const DB_META_TIP_SLOT_KEY: &str = "tip_slot";
/// Cell name for a breakpoint.
pub const BREAKPOINT_CELL_NAME: &str = "breakpoint";
@ -211,20 +215,7 @@ fn apply_block_transactions(mut block: Block, state: &mut V03State) -> DbResult<
})?;
} else {
transaction
.transaction_stateless_check()
.map_err(|err| {
DbError::db_interaction_error(format!(
"transaction pre check failed with err {err:?}"
))
})?
// FIXME: HOT FIX (testnet v0.2): does not check for system account updates due to
// sequencer-generated deposit tx'es;
// CHANGE ME back to `execute_check_on_state` when the indexer can authenticate deposit transactions
.execute_without_system_accounts_check_on_state(
state,
block.header.block_id,
block.header.timestamp,
)
.execute_on_state(state, block.header.block_id, block.header.timestamp)
.map_err(|err| {
DbError::db_interaction_error(format!(
"transaction execution failed with err {err:?}"

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

@ -4,7 +4,8 @@ use crate::{
cells::shared_cells::{BlockCell, FirstBlockCell, FirstBlockSetCell, LastBlockCell},
indexer::indexer_cells::{
AccNumTxCell, BlockHashToBlockIdMapCell, BreakpointCellOwned, LastBreakpointIdCell,
LastObservedL1LibHeaderCell, TxHashToBlockIdMapCell, ZoneSdkIndexerCursorCellOwned,
LastObservedL1LibHeaderCell, StallReasonCellOwned, TipSlotCell, TxHashToBlockIdMapCell,
ZoneSdkIndexerCursorCellOwned,
},
};
@ -36,6 +37,11 @@ impl RocksDBIO {
.map(|opt| opt.map(|cell| cell.0))
}
pub fn get_meta_tip_slot_in_db(&self) -> DbResult<Option<u64>> {
self.get_opt::<TipSlotCell>(())
.map(|opt| opt.map(|cell| cell.0))
}
// Block
pub fn get_block(&self, block_id: u64) -> DbResult<Option<Block>> {
@ -73,4 +79,8 @@ impl RocksDBIO {
.get_opt::<ZoneSdkIndexerCursorCellOwned>(())?
.map(|cell| cell.0))
}
pub fn get_stall_reason_bytes(&self) -> DbResult<Option<Vec<u8>>> {
Ok(self.get_opt::<StallReasonCellOwned>(())?.map(|cell| cell.0))
}
}

50
lez/storage/src/indexer/tests.rs
View File

@ -87,7 +87,7 @@ fn one_block_insertion() {
let dbio = RocksDBIO::open_or_create(temdir_path, &initial_state()).unwrap();
let genesis_block = genesis_block();
dbio.put_block(&genesis_block, [0; 32]).unwrap();
dbio.put_block(&genesis_block, [0; 32], 0).unwrap();
let prev_hash = genesis_block.header.hash;
let from = acc1();
@ -98,7 +98,7 @@ fn one_block_insertion() {
common::test_utils::create_transaction_native_token_transfer(from, 0, to, 1, &sign_key);
let block = produce_dummy_block(2, Some(prev_hash), vec![transfer_tx]);
dbio.put_block(&block, [1; 32]).unwrap();
dbio.put_block(&block, [1; 32], 0).unwrap();
let last_id = dbio.get_meta_last_block_id_in_db().unwrap().unwrap();
let first_id = dbio.get_meta_first_block_id_in_db().unwrap();
@ -130,6 +130,26 @@ fn one_block_insertion() {
);
}
#[test]
fn put_block_records_tip_inscription_slot() {
let temp_dir = tempdir().unwrap();
let dbio = RocksDBIO::open_or_create(temp_dir.path(), &initial_state()).unwrap();
assert_eq!(dbio.get_meta_tip_slot_in_db().unwrap(), None);
let genesis_block = genesis_block();
dbio.put_block(&genesis_block, [0; 32], 1_000).unwrap();
assert_eq!(dbio.get_meta_tip_slot_in_db().unwrap(), Some(1_000));
let block = produce_dummy_block(2, Some(genesis_block.header.hash), vec![]);
dbio.put_block(&block, [1; 32], 1_005).unwrap();
assert_eq!(dbio.get_meta_tip_slot_in_db().unwrap(), Some(1_005));
// Re-inserting a block at/below the tip must not move the tip slot.
dbio.put_block(&genesis_block, [0; 32], 1_010).unwrap();
assert_eq!(dbio.get_meta_tip_slot_in_db().unwrap(), Some(1_005));
}
#[test]
fn new_breakpoint() {
let temp_dir = tempdir().unwrap();
@ -155,7 +175,7 @@ fn new_breakpoint() {
&sign_key,
);
let block = produce_dummy_block(i.into(), prev_hash, vec![transfer_tx]);
dbio.put_block(&block, [i; 32]).unwrap();
dbio.put_block(&block, [i; 32], 0).unwrap();
}
let last_id = dbio.get_meta_last_block_id_in_db().unwrap().unwrap();
@ -211,7 +231,7 @@ fn simple_maps() {
let control_hash1 = block.header.hash;
dbio.put_block(&block, [1; 32]).unwrap();
dbio.put_block(&block, [1; 32], 0).unwrap();
let last_id = dbio.get_meta_last_block_id_in_db().unwrap().unwrap();
let last_block = dbio.get_block(last_id).unwrap().unwrap();
@ -223,7 +243,7 @@ fn simple_maps() {
let control_hash2 = block.header.hash;
dbio.put_block(&block, [2; 32]).unwrap();
dbio.put_block(&block, [2; 32], 0).unwrap();
let last_id = dbio.get_meta_last_block_id_in_db().unwrap().unwrap();
let last_block = dbio.get_block(last_id).unwrap().unwrap();
@ -235,7 +255,7 @@ fn simple_maps() {
let control_tx_hash1 = transfer_tx.hash();
let block = produce_dummy_block(3, Some(prev_hash), vec![transfer_tx]);
dbio.put_block(&block, [3; 32]).unwrap();
dbio.put_block(&block, [3; 32], 0).unwrap();
let last_id = dbio.get_meta_last_block_id_in_db().unwrap().unwrap();
let last_block = dbio.get_block(last_id).unwrap().unwrap();
@ -247,7 +267,7 @@ fn simple_maps() {
let control_tx_hash2 = transfer_tx.hash();
let block = produce_dummy_block(4, Some(prev_hash), vec![transfer_tx]);
dbio.put_block(&block, [4; 32]).unwrap();
dbio.put_block(&block, [4; 32], 0).unwrap();
let control_block_id1 = dbio.get_block_id_by_hash(control_hash1.0).unwrap().unwrap();
let control_block_id2 = dbio.get_block_id_by_hash(control_hash2.0).unwrap().unwrap();
@ -284,7 +304,7 @@ fn block_batch() {
let block = produce_dummy_block(1, None, vec![transfer_tx]);
block_res.push(block.clone());
dbio.put_block(&block, [1; 32]).unwrap();
dbio.put_block(&block, [1; 32], 0).unwrap();
let last_id = dbio.get_meta_last_block_id_in_db().unwrap().unwrap();
let last_block = dbio.get_block(last_id).unwrap().unwrap();
@ -295,7 +315,7 @@ fn block_batch() {
let block = produce_dummy_block(2, Some(prev_hash), vec![transfer_tx]);
block_res.push(block.clone());
dbio.put_block(&block, [2; 32]).unwrap();
dbio.put_block(&block, [2; 32], 0).unwrap();
let last_id = dbio.get_meta_last_block_id_in_db().unwrap().unwrap();
let last_block = dbio.get_block(last_id).unwrap().unwrap();
@ -306,7 +326,7 @@ fn block_batch() {
let block = produce_dummy_block(3, Some(prev_hash), vec![transfer_tx]);
block_res.push(block.clone());
dbio.put_block(&block, [3; 32]).unwrap();
dbio.put_block(&block, [3; 32], 0).unwrap();
let last_id = dbio.get_meta_last_block_id_in_db().unwrap().unwrap();
let last_block = dbio.get_block(last_id).unwrap().unwrap();
@ -317,7 +337,7 @@ fn block_batch() {
let block = produce_dummy_block(4, Some(prev_hash), vec![transfer_tx]);
block_res.push(block.clone());
dbio.put_block(&block, [4; 32]).unwrap();
dbio.put_block(&block, [4; 32], 0).unwrap();
let block_hashes_mem: Vec<[u8; 32]> =
block_res.into_iter().map(|bl| bl.header.hash.0).collect();
@ -376,7 +396,7 @@ fn account_map() {
let block = produce_dummy_block(1, None, vec![transfer_tx1, transfer_tx2]);
dbio.put_block(&block, [1; 32]).unwrap();
dbio.put_block(&block, [1; 32], 0).unwrap();
let last_id = dbio.get_meta_last_block_id_in_db().unwrap().unwrap();
let last_block = dbio.get_block(last_id).unwrap().unwrap();
@ -391,7 +411,7 @@ fn account_map() {
let block = produce_dummy_block(2, Some(prev_hash), vec![transfer_tx1, transfer_tx2]);
dbio.put_block(&block, [2; 32]).unwrap();
dbio.put_block(&block, [2; 32], 0).unwrap();
let last_id = dbio.get_meta_last_block_id_in_db().unwrap().unwrap();
let last_block = dbio.get_block(last_id).unwrap().unwrap();
@ -406,7 +426,7 @@ fn account_map() {
let block = produce_dummy_block(3, Some(prev_hash), vec![transfer_tx1, transfer_tx2]);
dbio.put_block(&block, [3; 32]).unwrap();
dbio.put_block(&block, [3; 32], 0).unwrap();
let last_id = dbio.get_meta_last_block_id_in_db().unwrap().unwrap();
let last_block = dbio.get_block(last_id).unwrap().unwrap();
@ -418,7 +438,7 @@ fn account_map() {
let block = produce_dummy_block(4, Some(prev_hash), vec![transfer_tx]);
dbio.put_block(&block, [4; 32]).unwrap();
dbio.put_block(&block, [4; 32], 0).unwrap();
let acc1_tx = dbio.get_acc_transactions(*acc1().value(), 0, 7).unwrap();
let acc1_tx_hashes: Vec<[u8; 32]> = acc1_tx.into_iter().map(|tx| tx.hash().0).collect();

16
lez/storage/src/indexer/write_atomic.rs
View File

@ -8,7 +8,7 @@ use crate::{
cells::shared_cells::{FirstBlockCell, FirstBlockSetCell, LastBlockCell},
indexer::indexer_cells::{
AccNumTxCell, BlockHashToBlockIdMapCell, LastBreakpointIdCell, LastObservedL1LibHeaderCell,
TxHashToBlockIdMapCell,
TipSlotCell, TxHashToBlockIdMapCell,
},
};
@ -139,13 +139,20 @@ impl RocksDBIO {
self.put_batch(&LastBreakpointIdCell(br_id), (), write_batch)
}
pub fn put_meta_tip_slot_in_db_batch(
&self,
l1_slot: u64,
write_batch: &mut WriteBatch,
) -> DbResult<()> {
self.put_batch(&TipSlotCell(l1_slot), (), write_batch)
}
pub fn put_meta_is_first_block_set_batch(&self, write_batch: &mut WriteBatch) -> DbResult<()> {
self.put_batch(&FirstBlockSetCell(true), (), write_batch)
}
// Block
pub fn put_block(&self, block: &Block, l1_lib_header: [u8; 32]) -> DbResult<()> {
/// Put a block atomically (via [`WriteBatch`]) along with its L1 header and `Slot`.
pub fn put_block(&self, block: &Block, l1_lib_header: [u8; 32], l1_slot: u64) -> DbResult<()> {
let cf_block = self.block_column();
let last_curr_block = self.get_meta_last_block_id_in_db()?.unwrap_or(0);
let mut write_batch = WriteBatch::default();
@ -163,6 +170,7 @@ impl RocksDBIO {
if block.header.block_id > last_curr_block {
self.put_meta_last_block_in_db_batch(block.header.block_id, &mut write_batch)?;
self.put_meta_last_observed_l1_lib_header_in_db_batch(l1_lib_header, &mut write_batch)?;
self.put_meta_tip_slot_in_db_batch(l1_slot, &mut write_batch)?;
}
if last_curr_block == 0 {
self.put_meta_first_block_in_db_batch(block, &mut write_batch)?;

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, LastBreakpointIdCell, LastObservedL1LibHeaderCell,
BreakpointCellRef, LastBreakpointIdCell, LastObservedL1LibHeaderCell, StallReasonCellRef,
ZoneSdkIndexerCursorCellRef,
},
};
@ -35,6 +35,10 @@ impl RocksDBIO {
self.put(&ZoneSdkIndexerCursorCellRef(bytes), ())
}
pub fn put_stall_reason_bytes(&self, bytes: &[u8]) -> DbResult<()> {
self.put(&StallReasonCellRef(bytes), ())
}
// State
pub fn put_breakpoint(&self, br_id: u64, breakpoint: &V03State) -> DbResult<()> {

1
test_fixtures/src/config.rs
View File

@ -176,6 +176,7 @@ pub fn indexer_config(bedrock_addr: SocketAddr) -> Result<IndexerConfig> {
auth: None,
},
channel_id: bedrock_channel_id(),
allow_chain_reset: false,
})
}