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

feat(sequencer): bootstrap state from Bedrock

Browse Source
This commit is contained in:
Daniil Polyakov 2026-07-09 00:22:33 +03:00
parent 0532760f8a
commit 0428dae84d
18 changed files with 958 additions and 395 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

21
Cargo.lock generated
View File

@ -1349,6 +1349,25 @@ dependencies = [
"rand_core 0.10.1",
]
[[package]]
name = "chain_consistency"
version = "0.1.0"
dependencies = [
"anyhow",
"borsh",
"common",
"futures",
"lee",
"lee_core",
"log",
"logos-blockchain-core",
"logos-blockchain-zone-sdk",
"serde",
"serde_json",
"thiserror 2.0.18",
"tokio",
]
[[package]]
name = "chkstk_stub"
version = "0.1.0"
@ -3852,6 +3871,7 @@ dependencies = [
"async-stream",
"authenticated_transfer_core",
"borsh",
"chain_consistency",
"common",
"futures",
"humantime-serde",
@ -8896,6 +8916,7 @@ dependencies = [
"borsh",
"bridge_core",
"bytesize",
"chain_consistency",
"chrono",
"common",
"faucet_core",

2
Cargo.toml
View File

@ -32,6 +32,7 @@ members = [
"lez/wallet",
"lez/wallet-ffi",
"lez/common",
"lez/chain_consistency",
"lez/programs",
"lez/programs/amm",
"lez/programs/associated_token_account",
@ -64,6 +65,7 @@ members = [
lee = { path = "lee/state_machine" }
lee_core = { path = "lee/state_machine/core" }
common = { path = "lez/common" }
chain_consistency = { path = "lez/chain_consistency" }
mempool = { path = "lez/mempool" }
storage = { path = "lez/storage" }
key_protocol = { path = "lee/key_protocol" }

26
lez/chain_consistency/Cargo.toml Normal file
View File

@ -0,0 +1,26 @@
[package]
name = "chain_consistency"
version = "0.1.0"
edition = "2024"
license.workspace = true
[lints]
workspace = true
[dependencies]
common.workspace = true
lee.workspace = true
lee_core.workspace = true
logos-blockchain-zone-sdk.workspace = true
logos-blockchain-core.workspace = true
anyhow.workspace = true
borsh.workspace = true
futures.workspace = true
log.workspace = true
serde.workspace = true
thiserror.workspace = true
tokio.workspace = true
[dev-dependencies]
serde_json.workspace = true

182
lez/chain_consistency/src/apply.rs Normal file
View File

@ -0,0 +1,182 @@
//! Validating and applying channel L2 blocks to a `V03State`.
//!
//! These primitives are shared by the consumers that reconstruct L2 state from
//! the Bedrock channel.
use common::{
HashType,
block::Block,
transaction::{LeeTransaction, clock_invocation},
};
use lee::{GENESIS_BLOCK_ID, V03State};
use lee_core::BlockId;
use serde::{Deserialize, Serialize};
/// Why a channel L2 block could not be validated or applied.
///
/// Persisted in `RocksDB` (via the Indexer's stall reason), so every variant
/// must be `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: BlockId, got: BlockId },
#[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,
},
}
/// The last successfully applied block a candidate must extend.
pub struct Tip {
pub block_id: BlockId,
pub hash: HashType,
}
/// 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.
pub 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`.
pub fn apply_block(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 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
}
));
}
}

310
lez/indexer/core/src/chain_consistency.rs → lez/chain_consistency/src/consistency.rs
View File

@ -1,18 +1,18 @@
//! Startup check that the local store still belongs to the chain the
//! connected channel serves.
//! Startup check that a local store still belongs to the chain the connected
//! channel serves.
use anyhow::Result;
use common::{HashType, block::Block};
use futures::StreamExt as _;
use lee_core::BlockId;
use log::warn;
use logos_blockchain_zone_sdk::{Slot, ZoneMessage, adapter::Node as _};
use crate::IndexerCore;
use logos_blockchain_core::mantle::ops::channel::ChannelId;
use logos_blockchain_zone_sdk::{Slot, ZoneMessage, adapter, indexer::ZoneIndexer};
/// 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.
/// Result of comparing a caller'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).
@ -24,7 +24,7 @@ pub enum ChainConsistency {
/// - 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.
/// A genuine divergence is still caught later when applying the channel history.
Inconclusive,
/// Positive evidence that the channel is a different chain than the store.
///
@ -36,13 +36,13 @@ pub enum ChainConsistency {
pub enum ChainMismatch {
/// The channel serves a different block at the anchor's id.
Block {
ours: (u64, HashType),
channel: (u64, HashType),
ours: (BlockId, HashType),
channel: (BlockId, 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),
channel: (BlockId, HashType),
slot: Slot,
anchor_slot: Slot,
},
@ -97,18 +97,25 @@ impl std::fmt::Display for ChainMismatch {
/// 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 {
pub 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)>,
/// `None` when anchored on an undeserializable inscription (no header was recorded).
block: Option<(BlockId, HashType)>,
}
impl Anchor {
/// Builds an anchor at `slot` on the block `(id, hash)`, or a headerless
/// anchor (`None`) when only the slot is known.
#[must_use]
pub const fn new(slot: Slot, block: Option<(BlockId, HashType)>) -> Self {
Self { slot, block }
}
/// 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 {
/// See [`verify_chain_consistency`].
fn probe_anchor_slot(&self, msg: &ZoneMessage, slot: Slot) -> AnchorProbe {
if slot < self.slot {
return AnchorProbe::KeepLooking;
}
@ -168,104 +175,104 @@ enum AnchorProbe {
KeepLooking,
}
#[expect(
clippy::multiple_inherent_impl,
reason = "split for clarity & isolation of relevant code"
)]
impl IndexerCore {
/// Detects when the local store belongs to a different chain than the connected
/// L1 (e.g. a wiped/restarted bedrock) so startup can reset instead of silently
/// diverging. Best-effort dev convenience; won't trigger during normal live indexing.
///
/// Anchors on a block the same chain must still serve at a known slot: the
/// recorded parked block while stalled, otherwise the tip at the read
/// cursor (only blocks advance the cursor, so when not parked it is the
/// tip's inscription slot). See [`Self::verify_chain_at_anchor`].
pub(crate) async fn verify_chain_consistency(&self) -> Result<ChainConsistency> {
let anchor = if let Some(stall) = self.store.get_stall_reason()? {
Anchor {
slot: stall.l1_slot,
block: stall.block_id.zip(stall.block_hash),
}
} else {
let Some(cursor) = self.store.get_zone_cursor()? else {
// if we have no cursor, the store is empty or cold: nothing to compare
return Ok(ChainConsistency::Inconclusive);
};
let Some(tip_id) = self.store.get_last_block_id()? else {
return Ok(ChainConsistency::Inconclusive);
};
let Some(tip) = self.store.get_block_at_id(tip_id)? else {
return Ok(ChainConsistency::Inconclusive);
};
Anchor {
slot: cursor,
block: Some((tip_id, tip.header.hash)),
}
};
self.verify_chain_at_anchor(&anchor).await
}
/// 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 [`ChainConsistency::Inconsistent`]; absence of data stays
/// [`ChainConsistency::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:#}");
/// Detects when a local store belongs to a different chain than the connected
/// L1 (e.g. a wiped/restarted Bedrock) so startup can react instead of silently
/// diverging.
///
/// 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
/// [`ChainConsistency::Inconsistent`]; absence of data stays
/// [`ChainConsistency::Inconclusive`].
///
/// `node` need only implement the zone-sdk [`adapter::Node`] trait; a throwaway
/// [`ZoneIndexer`] is built internally for the channel read.
pub async fn verify_chain_consistency<N>(
node: &N,
channel_id: ChannelId,
anchor: &Anchor,
) -> Result<ChainConsistency>
where
N: adapter::Node + Clone + Sync,
{
match node.channel_state(channel_id).await {
Ok(state) => {
if let Some(mismatch) = frontier_verdict(anchor.slot, state.map(|s| s.tip_slot)) {
return Ok(ChainConsistency::Inconsistent(mismatch));
}
}
// `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)
}
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 zone_indexer = ZoneIndexer::new(channel_id, node.clone());
let scan = async {
let stream = 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)
}
}
}
/// Classifies an already-fetched slice of channel messages against `anchor`.
///
/// The message-slice counterpart of [`verify_chain_consistency`], for callers
/// that read the channel history themselves (e.g. a sequencer that reads once
/// and both verifies and reconstructs from the same messages). `messages` must
/// be the finalized channel history from the anchor slot onward, in slot order.
///
/// Reuses the same [`Anchor::probe_anchor_slot`] and [`frontier_verdict`] logic
/// as the streaming path, so both consumers share one definition of a mismatch.
#[must_use]
pub fn classify_channel(
anchor: &Anchor,
channel_tip_slot: Option<Slot>,
messages: &[(ZoneMessage, Slot)],
) -> ChainConsistency {
if let Some(mismatch) = frontier_verdict(anchor.slot, channel_tip_slot) {
return ChainConsistency::Inconsistent(mismatch);
}
for (msg, slot) in messages {
match anchor.probe_anchor_slot(msg, *slot) {
AnchorProbe::SameChain => return ChainConsistency::Consistent,
AnchorProbe::Mismatch(mismatch) => return ChainConsistency::Inconsistent(mismatch),
AnchorProbe::Bail => return ChainConsistency::Inconclusive,
AnchorProbe::KeepLooking => {}
}
}
ChainConsistency::Inconclusive
}
/// Checks the channel frontier against the anchor slot.
@ -285,33 +292,13 @@ fn frontier_verdict(anchor_slot: Slot, channel_tip_slot: Option<Slot>) -> Option
#[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 {
fn test_block(block_id: BlockId, timestamp: u64) -> Block {
HashableBlockData {
block_id,
prev_block_hash: HashType([0; 32]),
@ -330,63 +317,7 @@ mod tests {
}
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
));
Anchor::new(slot, Some((block.header.block_id, block.header.hash)))
}
#[test]
@ -467,10 +398,7 @@ mod tests {
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 anchor = Anchor::new(Slot::from(1_000), None);
let garbage = ZoneMessage::Block(ZoneBlock {
id: MsgId::from([0_u8; 32]),
data: Inscription::try_from(&[1_u8, 2, 3][..]).expect("inscription"),

9
lez/chain_consistency/src/lib.rs Normal file
View File

@ -0,0 +1,9 @@
//! Reconstructing and verifying L2 chain state from a Bedrock (L1) channel.
pub use apply::{BlockIngestError, Tip, apply_block, validate_against_tip};
pub use consistency::{
Anchor, ChainConsistency, ChainMismatch, classify_channel, verify_chain_consistency,
};
pub mod apply;
pub mod consistency;

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

@ -13,6 +13,7 @@ testnet = []
[dependencies]
common.workspace = true
chain_consistency.workspace = true
logos-blockchain-zone-sdk.workspace = true
lee.workspace = true
lee_core.workspace = true

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

@ -1,12 +1,12 @@
use std::{path::Path, sync::Arc};
use anyhow::{Context as _, Result};
use chain_consistency::{BlockIngestError, Tip};
use common::{
HashType,
block::{BedrockStatus, Block, BlockHeader},
transaction::{LeeTransaction, clock_invocation},
transaction::LeeTransaction,
};
use lee::{Account, AccountId, GENESIS_BLOCK_ID, V03State};
use lee::{Account, AccountId, V03State};
use lee_core::BlockId;
use log::warn;
use logos_blockchain_core::header::HeaderId;
@ -14,12 +14,7 @@ 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,
}
use crate::stall_reason::StallReason;
/// Outcome of feeding a parsed L2 block to the validated tip.
pub enum AcceptOutcome {
@ -239,14 +234,14 @@ impl IndexerStore {
return Ok(AcceptOutcome::AlreadyApplied);
}
if let Err(err) = validate_against_tip(tip.as_ref(), block) {
if let Err(err) = chain_consistency::validate_against_tip(tip.as_ref(), block) {
self.record_stall(Some(&block.header), l1_slot, err.clone())?;
return Ok(AcceptOutcome::Parked(err));
}
// TODO: we use scratch state to be atomic, but need to revisit how expensive a clone is
let mut scratch = self.current_state.read().await.clone();
if let Err(err) = apply_block_to_scratch(block, &mut scratch) {
if let Err(err) = chain_consistency::apply_block(block, &mut scratch) {
self.record_stall(Some(&block.header), l1_slot, err.clone())?;
return Ok(AcceptOutcome::Parked(err));
}
@ -268,112 +263,12 @@ impl IndexerStore {
}
}
/// Checks that `block` is the valid continuation of `tip`: hash integrity,
/// then block-id continuity, then `prev_block_hash` linkage. A `None` tip
/// (cold store) expects the genesis block.
fn validate_against_tip(tip: Option<&Tip>, block: &Block) -> Result<(), BlockIngestError> {
let computed = block.recompute_hash();
if computed != block.header.hash {
return Err(BlockIngestError::HashMismatch {
computed,
header: block.header.hash,
});
}
match tip {
None => {
if block.header.block_id != GENESIS_BLOCK_ID {
return Err(BlockIngestError::UnexpectedBlockId {
expected: GENESIS_BLOCK_ID,
got: block.header.block_id,
});
}
}
Some(tip) => {
let expected = tip
.block_id
.checked_add(1)
.expect("block id 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};
use crate::stall_reason::StallReason;
#[tokio::test]
async fn stall_reason_roundtrips_and_clears() {
@ -519,10 +414,10 @@ mod tests {
#[cfg(test)]
mod accept_tests {
use chain_consistency::BlockIngestError;
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")

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

@ -1,67 +0,0 @@
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
}
));
}
}

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

@ -2,10 +2,10 @@ use std::{path::Path, sync::Arc};
use anyhow::Result;
use arc_swap::ArcSwap;
use chain_consistency::{Anchor, BlockIngestError, ChainConsistency};
use common::block::Block;
// TODO: Remove after testnet
use futures::StreamExt as _;
pub use ingest_error::BlockIngestError;
use log::{error, info, warn};
use logos_blockchain_zone_sdk::{
CommonHttpClient, Slot, ZoneMessage, adapter::NodeHttpClient, indexer::ZoneIndexer,
@ -14,14 +14,11 @@ pub use stall_reason::StallReason;
use crate::{
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;
@ -93,6 +90,35 @@ impl IndexerCore {
})
}
/// Detects when the local store belongs to a different chain than the connected
/// L1 (e.g. a wiped/restarted bedrock) so startup can reset instead of silently
/// diverging. Best-effort dev convenience; won't trigger during normal live indexing.
///
/// The anchoring/probe logic lives in the shared [`chain_consistency`] crate;
/// here we build the [`Anchor`] from the store (the recorded parked block while
/// stalled, otherwise the tip at the read cursor — only blocks advance the
/// cursor, so when not parked it is the tip's inscription slot) and delegate.
async fn verify_chain_consistency(&self) -> Result<ChainConsistency> {
let anchor = if let Some(stall) = self.store.get_stall_reason()? {
Anchor::new(stall.l1_slot, stall.block_id.zip(stall.block_hash))
} else {
let Some(cursor) = self.store.get_zone_cursor()? else {
// if we have no cursor, the store is empty or cold: nothing to compare
return Ok(ChainConsistency::Inconclusive);
};
let Some(tip_id) = self.store.get_last_block_id()? else {
return Ok(ChainConsistency::Inconclusive);
};
let Some(tip) = self.store.get_block_at_id(tip_id)? else {
return Ok(ChainConsistency::Inconclusive);
};
Anchor::new(cursor, Some((tip_id, tip.header.hash)))
};
chain_consistency::verify_chain_consistency(&self.node, self.config.channel_id, &anchor)
.await
}
/// Snapshot of the current ingestion status (sync state + indexed tip).
///
/// Combines the ingest loop's live status with the L2 tip read fresh from the
@ -272,3 +298,90 @@ impl IndexerCore {
}
}
}
#[cfg(test)]
mod tests {
use std::time::Duration;
use common::{HashType, block::HashableBlockData};
use logos_blockchain_zone_sdk::Slot;
use super::*;
use crate::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"))
}
#[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
));
}
}

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

@ -11,6 +11,7 @@ workspace = true
lee.workspace = true
lee_core.workspace = true
common.workspace = true
chain_consistency.workspace = true
storage.workspace = true
mempool.workspace = true
logos-blockchain-zone-sdk.workspace = true

45
lez/sequencer/core/src/block_publisher.rs
View File

@ -2,14 +2,16 @@ use std::{pin::Pin, sync::Arc, time::Duration};
use anyhow::{Context as _, Result, anyhow};
use common::block::Block;
use futures::StreamExt as _;
use log::{info, warn};
pub use logos_blockchain_core::mantle::ops::channel::MsgId;
use logos_blockchain_core::mantle::ops::channel::{ChannelId, inscribe::Inscription};
pub use logos_blockchain_key_management_system_service::keys::{Ed25519Key, ZkKey};
pub use logos_blockchain_zone_sdk::sequencer::SequencerCheckpoint;
use logos_blockchain_zone_sdk::{
CommonHttpClient,
adapter::NodeHttpClient,
CommonHttpClient, Slot, ZoneMessage,
adapter::{Node as _, NodeHttpClient},
indexer::ZoneIndexer,
sequencer::{
DepositInfo, Event, FinalizedOp, InscriptionInfo,
SequencerConfig as ZoneSdkSequencerConfig, WithdrawArg, WithdrawInfo, ZoneSequencer,
@ -63,12 +65,27 @@ pub trait BlockPublisherTrait: Clone {
async fn publish_block(&self, block: &Block, withdrawals: Vec<WithdrawArg>) -> Result<()>;
fn channel_id(&self) -> ChannelId;
/// Current channel frontier slot on the connected chain, or `None` if the
/// channel does not exist there. Drives the startup frontier check.
async fn channel_tip_slot(&self) -> Result<Option<Slot>>;
/// Finalized channel messages from `after_slot` (exclusive) up to LIB, used
/// for the startup consistency check and reconstruction. Pass `None` to read
/// from the channel's genesis.
async fn read_channel_after(
&self,
after_slot: Option<Slot>,
) -> Result<Vec<(ZoneMessage, Slot)>>;
}
/// Real block publisher backed by zone-sdk's `ZoneSequencer`.
#[derive(Clone)]
pub struct ZoneSdkPublisher {
channel_id: ChannelId,
/// Direct node handle retained for channel reads (startup consistency check
/// and reconstruction); the sequencer itself lives in the drive task.
node: NodeHttpClient,
publish_tx: mpsc::Sender<(Inscription, Vec<WithdrawArg>)>,
// Aborts the drive task when the last clone is dropped.
_drive_task: Arc<DriveTaskGuard>,
@ -104,7 +121,7 @@ impl BlockPublisherTrait for ZoneSdkPublisher {
let mut sequencer = ZoneSequencer::init_with_config(
config.channel_id,
bedrock_signing_key,
node,
node.clone(),
zone_sdk_config,
initial_checkpoint,
);
@ -196,6 +213,7 @@ impl BlockPublisherTrait for ZoneSdkPublisher {
Ok(Self {
channel_id: config.channel_id,
node,
publish_tx,
_drive_task: Arc::new(DriveTaskGuard(drive_task)),
})
@ -218,6 +236,27 @@ impl BlockPublisherTrait for ZoneSdkPublisher {
fn channel_id(&self) -> ChannelId {
self.channel_id
}
async fn channel_tip_slot(&self) -> Result<Option<Slot>> {
Ok(self
.node
.channel_state(self.channel_id)
.await
.context("Failed to read channel state")?
.map(|state| state.tip_slot))
}
async fn read_channel_after(
&self,
after_slot: Option<Slot>,
) -> Result<Vec<(ZoneMessage, Slot)>> {
let indexer = ZoneIndexer::new(self.channel_id, self.node.clone());
let stream = indexer
.next_messages(after_slot)
.await
.context("Failed to start channel read stream")?;
Ok(stream.collect().await)
}
}
/// Deserialize inscription payload as a `Block` and return it's`block_id`.

12
lez/sequencer/core/src/block_store.rs
View File

@ -12,7 +12,7 @@ use logos_blockchain_zone_sdk::sequencer::SequencerCheckpoint;
pub use storage::DbResult;
use storage::sequencer::{
RocksDBIO,
sequencer_cells::{PendingDepositEventRecord, WithdrawalReconciliationKey},
sequencer_cells::{PendingDepositEventRecord, WithdrawalReconciliationKey, ZoneAnchorRecord},
};
pub struct SequencerStore {
@ -165,6 +165,16 @@ impl SequencerStore {
Ok(())
}
/// The last channel block read back and verified from Bedrock (L1 slot +
/// `id`/`hash`), or `None` before any block has been read from the channel.
pub fn get_zone_anchor(&self) -> DbResult<Option<ZoneAnchorRecord>> {
self.dbio.get_zone_cursor()
}
pub fn set_zone_anchor(&self, anchor: &ZoneAnchorRecord) -> DbResult<()> {
self.dbio.put_zone_cursor(anchor)
}
pub fn get_unfulfilled_deposit_events(&self) -> DbResult<Vec<PendingDepositEventRecord>> {
self.dbio.get_pending_deposit_events()
}

192
lez/sequencer/core/src/lib.rs
View File

@ -2,17 +2,21 @@ use std::{path::Path, sync::Arc, time::Instant};
use anyhow::{Context as _, Result, anyhow};
use borsh::BorshDeserialize;
use chain_consistency::{Anchor, ChainConsistency, Tip};
use common::{
HashType,
block::{BedrockStatus, Block, HashableBlockData},
transaction::{LeeTransaction, clock_invocation},
};
use config::{GenesisAction, SequencerConfig};
use lee::{AccountId, PublicTransaction, public_transaction::Message};
use lee::{AccountId, PublicTransaction, V03State, public_transaction::Message};
use lee_core::GENESIS_BLOCK_ID;
use log::{error, info, warn};
use logos_blockchain_key_management_system_service::keys::{ED25519_SECRET_KEY_SIZE, Ed25519Key};
use logos_blockchain_zone_sdk::sequencer::{DepositInfo, WithdrawArg};
use logos_blockchain_zone_sdk::{
Slot, ZoneMessage,
sequencer::{DepositInfo, WithdrawArg},
};
use mempool::{MemPool, MemPoolHandle};
#[cfg(feature = "mock")]
pub use mock::SequencerCoreWithMockClients;
@ -20,7 +24,7 @@ use num_bigint::BigUint;
pub use storage::error::DbError;
use storage::sequencer::{
RocksDBIO,
sequencer_cells::{PendingDepositEventRecord, WithdrawalReconciliationKey},
sequencer_cells::{PendingDepositEventRecord, WithdrawalReconciliationKey, ZoneAnchorRecord},
};
use crate::{
@ -64,6 +68,13 @@ pub struct SequencerCore<BP: BlockPublisherTrait = ZoneSdkPublisher> {
block_publisher: BP,
}
/// Outcome of the startup verify-and-reconstruct pass.
struct ReconstructionSummary {
/// Whether the channel served any blocks. Drives the fresh-start decision
/// between publishing our own genesis and adopting the channel's history.
channel_had_blocks: bool,
}
impl<BP: BlockPublisherTrait> SequencerCore<BP> {
/// Starts the sequencer using the provided configuration.
/// If an existing database is found, the sequencer state is loaded from it and
@ -124,11 +135,7 @@ impl<BP: BlockPublisherTrait> SequencerCore<BP> {
load_or_create_signing_key(&config.home.join("bedrock_signing_key"))
.expect("Failed to load or create bedrock signing key");
let (store, state, genesis_block) = Self::open_or_create_store(&config);
let latest_block_meta = store
.latest_block_meta()
.expect("Failed to read latest block meta from store");
let (mut store, mut state, genesis_block) = Self::open_or_create_store(&config);
let initial_checkpoint = store
.get_zone_checkpoint()
@ -151,17 +158,27 @@ impl<BP: BlockPublisherTrait> SequencerCore<BP> {
.await
.expect("Failed to initialize Block Publisher");
// On a truly fresh start (no checkpoint persisted yet), publish the
// genesis block so the indexer can find the channel start. After the
// first publish, zone-sdk's checkpoint persistence covers further
// restarts.
if is_fresh_start {
// Before producing, verify our local state still belongs to the chain
// the channel serves and replay any channel blocks we are missing
// (e.g. from other sequencers).
let reconstruction = Self::verify_and_reconstruct(&block_publisher, &mut store, &mut state)
.await
.expect("Failed to verify/reconstruct sequencer state from Bedrock");
// Publish our genesis only when bootstrapping an empty channel. If the
// channel already has blocks (another sequencer bootstrapped it), we
// adopted them during reconstruction instead.
if is_fresh_start && !reconstruction.channel_had_blocks {
block_publisher
.publish_block(&genesis_block, vec![])
.await
.expect("Failed to publish genesis block");
}
let latest_block_meta = store
.latest_block_meta()
.expect("Failed to read latest block meta from store");
let sequencer_core = Self {
state,
store,
@ -174,6 +191,155 @@ impl<BP: BlockPublisherTrait> SequencerCore<BP> {
(sequencer_core, mempool_handle)
}
/// Verifies the local store still belongs to the chain the connected channel
/// serves and replays any finalized channel blocks missing locally into
/// `state`/`store`, recording each block's L1 inscription slot as the new
/// anchor. Fails (never parks) on any divergence.
async fn verify_and_reconstruct(
publisher: &BP,
store: &mut SequencerStore,
state: &mut V03State,
) -> Result<ReconstructionSummary> {
let anchor_record = store
.get_zone_anchor()
.context("Failed to read zone anchor")?;
let after_slot = anchor_record
.and_then(|record| record.slot.checked_sub(1))
.map(Slot::from);
let messages = publisher
.read_channel_after(after_slot)
.await
.context("Failed to read channel history for reconstruction")?;
// Consistency check against the recorded anchor: fail fast on positive
// evidence the channel is a different chain (missing/behind/reinscribed).
if let Some(record) = anchor_record {
let anchor = Anchor::new(
Slot::from(record.slot),
Some((record.block_id, record.hash)),
);
let channel_tip_slot = publisher
.channel_tip_slot()
.await
.context("Failed to read channel tip slot")?;
if let ChainConsistency::Inconsistent(mismatch) =
chain_consistency::classify_channel(&anchor, channel_tip_slot, &messages)
{
return Err(anyhow!(
"Sequencer store diverges from the Bedrock channel ({mismatch}). \
Delete the sequencer storage directory or point at the correct channel."
));
}
}
// Replay: apply channel blocks we are missing, verify the ones we have.
let mut channel_had_blocks = false;
for (message, slot) in &messages {
let ZoneMessage::Block(zone_block) = message else {
continue;
};
let block: Block = borsh::from_slice(&zone_block.data).map_err(|err| {
anyhow!(
"Failed to deserialize channel block at slot {}: {err}",
slot.into_inner()
)
})?;
channel_had_blocks = true;
Self::apply_reconstructed_block(store, state, &block, *slot)?;
}
Ok(ReconstructionSummary { channel_had_blocks })
}
/// Applies a single channel block during reconstruction: idempotent for
/// blocks we already hold (verifying their hash), a validated continuation
/// for new ones. Advances the persisted anchor to the block's slot.
fn apply_reconstructed_block(
store: &mut SequencerStore,
state: &mut V03State,
block: &Block,
slot: Slot,
) -> Result<()> {
let tip = store
.latest_block_meta()
.context("Failed to read latest block meta")?;
let block_id = block.header.block_id;
let block_hash = block.header.hash;
let record = ZoneAnchorRecord {
slot: slot.into_inner(),
block_id,
hash: block_hash,
};
// A block at/below the tip must match what we already stored, otherwise
// the channel is a different chain.
if block_id <= tip.id {
match store
.get_block_at_id(block_id)
.context("Failed to read stored block")?
{
Some(stored) if stored.header.hash == block_hash => {
store
.set_zone_anchor(&record)
.context("Failed to persist zone anchor")?;
return Ok(());
}
Some(stored) => {
return Err(anyhow!(
"Channel block {block_id} hash {block_hash} does not match stored hash {}",
stored.header.hash
));
}
None => {
return Err(anyhow!(
"Channel block {block_id} is at/below local tip {} but is missing locally",
tip.id
));
}
}
}
// New continuation: validate it chains onto the tip, then apply.
let cc_tip = Tip {
block_id: tip.id,
hash: tip.hash,
};
chain_consistency::validate_against_tip(Some(&cc_tip), block).map_err(|err| {
anyhow!(
"Channel block {block_id} does not extend local tip {}: {err}",
tip.id
)
})?;
let mut scratch = state.clone();
chain_consistency::apply_block(block, &mut scratch)
.map_err(|err| anyhow!("Failed to apply channel block {block_id}: {err}"))?;
// Derive bridge bookkeeping from the block's transactions, matching the
// production path so the reconciliation counters stay consistent.
let mut deposit_event_ids = Vec::new();
let mut withdrawals = Vec::new();
for tx in &block.body.transactions {
if let Some(deposit_id) = extract_bridge_deposit_id(tx) {
deposit_event_ids.push(deposit_id);
}
if let Some(withdraw) = extract_bridge_withdraw_data(tx) {
withdrawals.push(withdraw_event_reconciliation_key(&withdraw.outputs)?);
}
}
store
.update(block, &deposit_event_ids, withdrawals, &scratch)
.context("Failed to persist reconstructed block")?;
*state = scratch;
store
.set_zone_anchor(&record)
.context("Failed to persist zone anchor")?;
Ok(())
}
fn on_checkpoint(dbio: Arc<RocksDBIO>) -> block_publisher::CheckpointSink {
Box::new(move |cp| {
let bytes = match serde_json::to_vec(&cp) {

43
lez/sequencer/core/src/mock.rs
View File

@ -4,7 +4,7 @@ use anyhow::Result;
use common::block::Block;
use logos_blockchain_core::mantle::ops::channel::ChannelId;
use logos_blockchain_key_management_system_service::keys::Ed25519Key;
use logos_blockchain_zone_sdk::sequencer::WithdrawArg;
use logos_blockchain_zone_sdk::{Slot, ZoneMessage, sequencer::WithdrawArg};
use crate::{
block_publisher::{
@ -19,6 +19,28 @@ pub type SequencerCoreWithMockClients = crate::SequencerCore<MockBlockPublisher>
#[derive(Clone)]
pub struct MockBlockPublisher {
channel_id: ChannelId,
/// Canned channel frontier returned by [`Self::channel_tip_slot`].
tip_slot: Option<Slot>,
/// Canned finalized channel history returned by [`Self::read_channel_after`].
messages: Vec<(ZoneMessage, Slot)>,
}
impl MockBlockPublisher {
/// Builds a mock publisher backed by a canned channel, for reconstruction
/// and consistency tests. The default (via [`BlockPublisherTrait::new`])
/// serves an empty channel.
#[must_use]
pub const fn with_canned_channel(
channel_id: ChannelId,
tip_slot: Option<Slot>,
messages: Vec<(ZoneMessage, Slot)>,
) -> Self {
Self {
channel_id,
tip_slot,
messages,
}
}
}
impl BlockPublisherTrait for MockBlockPublisher {
@ -34,6 +56,8 @@ impl BlockPublisherTrait for MockBlockPublisher {
) -> Result<Self> {
Ok(Self {
channel_id: config.channel_id,
tip_slot: None,
messages: Vec::new(),
})
}
@ -48,4 +72,21 @@ impl BlockPublisherTrait for MockBlockPublisher {
fn channel_id(&self) -> ChannelId {
self.channel_id
}
async fn channel_tip_slot(&self) -> Result<Option<Slot>> {
Ok(self.tip_slot)
}
async fn read_channel_after(
&self,
after_slot: Option<Slot>,
) -> Result<Vec<(ZoneMessage, Slot)>> {
// Mirror `next_messages`: `after_slot` is exclusive.
Ok(self
.messages
.iter()
.filter(|(_, slot)| after_slot.is_none_or(|after| *slot > after))
.cloned()
.collect())
}
}

150
lez/sequencer/core/src/tests.rs
View File

@ -1231,3 +1231,153 @@ fn pda_mechanism_with_pinata_token_program() {
expected_winner_token_holding_post
);
}
mod reconstruction {
use common::block::Block;
use logos_blockchain_core::mantle::ops::channel::{MsgId, inscribe::Inscription};
use logos_blockchain_zone_sdk::{Slot, ZoneBlock, ZoneMessage};
use storage::sequencer::sequencer_cells::ZoneAnchorRecord;
use super::*;
use crate::{SequencerCore, block_store::SequencerStore, mock::MockBlockPublisher};
fn block_to_channel_message(block: &Block, slot: u64) -> (ZoneMessage, Slot) {
let bytes = borsh::to_vec(block).expect("serialize block");
let message = ZoneMessage::Block(ZoneBlock {
id: MsgId::from([0_u8; 32]),
data: Inscription::try_from(bytes.as_slice()).expect("inscription"),
});
(message, Slot::from(slot))
}
/// Collects a sequencer's whole chain (genesis..=tip) into a canned channel,
/// one block per slot at `slot_step` spacing.
fn channel_from_store(store: &SequencerStore, slot_step: u64) -> Vec<(ZoneMessage, Slot)> {
let genesis_id = store.genesis_id();
let tip_id = store.latest_block_meta().expect("tip").id;
(genesis_id..=tip_id)
.enumerate()
.map(|(index, id)| {
let block = store.get_block_at_id(id).expect("read").expect("present");
block_to_channel_message(&block, (index as u64 + 1) * slot_step)
})
.collect()
}
#[tokio::test]
async fn reconstructs_missing_channel_blocks_into_fresh_store() {
// Sequencer A produces a few blocks; treat its chain as the channel.
let config_a = setup_sequencer_config();
let (mut seq_a, _handle_a) =
SequencerCoreWithMockClients::start_from_config(config_a.clone()).await;
seq_a.produce_new_block().await.unwrap();
seq_a.produce_new_block().await.unwrap();
let tip_a = seq_a.block_store().latest_block_meta().unwrap();
let messages = channel_from_store(seq_a.block_store(), 10);
let tip_slot = messages.last().unwrap().1;
let channel_id = config_a.bedrock_config.channel_id;
// Sequencer B starts from a fresh store and reconstructs A's chain.
let config_b = setup_sequencer_config();
let (mut store_b, mut state_b, _genesis_b) =
SequencerCore::<MockBlockPublisher>::open_or_create_store(&config_b);
let mock_b = MockBlockPublisher::with_canned_channel(channel_id, Some(tip_slot), messages);
let summary = SequencerCore::<MockBlockPublisher>::verify_and_reconstruct(
&mock_b,
&mut store_b,
&mut state_b,
)
.await
.expect("reconstruct");
assert!(summary.channel_had_blocks);
let tip_b = store_b.latest_block_meta().unwrap();
assert_eq!(tip_b.id, tip_a.id);
assert_eq!(tip_b.hash, tip_a.hash);
// State matches: initial account balances agree with sequencer A.
for account in initial_public_user_accounts() {
assert_eq!(
state_b.get_account_by_id(account.account_id).balance,
seq_a.state().get_account_by_id(account.account_id).balance,
);
}
let anchor = store_b.get_zone_anchor().unwrap().expect("anchor recorded");
assert_eq!(anchor.block_id, tip_a.id);
assert_eq!(anchor.slot, tip_slot.into_inner());
// Re-running is idempotent: everything is already applied, no error.
let summary = SequencerCore::<MockBlockPublisher>::verify_and_reconstruct(
&mock_b,
&mut store_b,
&mut state_b,
)
.await
.expect("reconstruct idempotent");
assert!(summary.channel_had_blocks);
assert_eq!(store_b.latest_block_meta().unwrap().id, tip_a.id);
}
#[tokio::test]
async fn fails_when_channel_serves_a_divergent_block() {
let config = setup_sequencer_config();
let (mut store, mut state, _genesis) =
SequencerCore::<MockBlockPublisher>::open_or_create_store(&config);
// Anchor on the local genesis at some slot.
let genesis_id = store.genesis_id();
let genesis = store.get_block_at_id(genesis_id).unwrap().unwrap();
let anchor_slot = 100_u64;
store
.set_zone_anchor(&ZoneAnchorRecord {
slot: anchor_slot,
block_id: genesis_id,
hash: genesis.header.hash,
})
.unwrap();
// The channel serves a different block at the anchor id/slot.
let mut tampered = genesis.clone();
tampered.header.hash = HashType([9_u8; 32]);
let messages = vec![block_to_channel_message(&tampered, anchor_slot)];
let mock = MockBlockPublisher::with_canned_channel(
config.bedrock_config.channel_id,
Some(Slot::from(anchor_slot)),
messages,
);
let result = SequencerCore::<MockBlockPublisher>::verify_and_reconstruct(
&mock, &mut store, &mut state,
)
.await;
assert!(result.is_err(), "divergent channel must abort startup");
}
#[tokio::test]
async fn fails_when_channel_is_missing() {
let config = setup_sequencer_config();
let (mut store, mut state, _genesis) =
SequencerCore::<MockBlockPublisher>::open_or_create_store(&config);
let genesis_id = store.genesis_id();
let genesis = store.get_block_at_id(genesis_id).unwrap().unwrap();
store
.set_zone_anchor(&ZoneAnchorRecord {
slot: 100,
block_id: genesis_id,
hash: genesis.header.hash,
})
.unwrap();
// Anchor present, but the channel does not exist on the connected chain.
let mock =
MockBlockPublisher::with_canned_channel(config.bedrock_config.channel_id, None, vec![]);
let result = SequencerCore::<MockBlockPublisher>::verify_and_reconstruct(
&mock, &mut store, &mut state,
)
.await;
assert!(result.is_err(), "missing channel must abort startup");
}
}

14
lez/storage/src/sequencer/mod.rs
View File

@ -20,7 +20,7 @@ use crate::{
LEEStateCellOwned, LEEStateCellRef, LastFinalizedBlockIdCell, LatestBlockMetaCellOwned,
LatestBlockMetaCellRef, PendingDepositEventRecord, PendingDepositEventsCellOwned,
PendingDepositEventsCellRef, UnseenWithdrawCountCell, WithdrawalReconciliationKey,
ZoneSdkCheckpointCellOwned, ZoneSdkCheckpointCellRef,
ZoneAnchorRecord, ZoneCursorCell, ZoneSdkCheckpointCellOwned, ZoneSdkCheckpointCellRef,
},
};
@ -32,6 +32,10 @@ pub const DB_META_LAST_FINALIZED_BLOCK_ID: &str = "last_finalized_block_id";
pub const DB_META_LATEST_BLOCK_META_KEY: &str = "latest_block_meta";
/// Key base for storing the zone-sdk sequencer checkpoint (opaque bytes).
pub const DB_META_ZONE_SDK_CHECKPOINT_KEY: &str = "zone_sdk_checkpoint";
/// Key base for storing the last channel block read back and verified from
/// Bedrock (its L1 slot + `id`/`hash`) — the anchor for the startup
/// consistency check and the resume point for reconstruction.
pub const DB_META_ZONE_CURSOR_KEY: &str = "zone_cursor";
/// Key base for storing queued deposit events that were not yet
/// fulfilled on L2.
pub const DB_META_PENDING_DEPOSIT_EVENTS_KEY: &str = "pending_deposit_events";
@ -246,6 +250,14 @@ impl RocksDBIO {
self.put(&ZoneSdkCheckpointCellRef(bytes), ())
}
pub fn get_zone_cursor(&self) -> DbResult<Option<ZoneAnchorRecord>> {
Ok(self.get_opt::<ZoneCursorCell>(())?.map(|cell| cell.0))
}
pub fn put_zone_cursor(&self, anchor: &ZoneAnchorRecord) -> DbResult<()> {
self.put(&ZoneCursorCell(*anchor), ())
}
pub fn get_pending_deposit_events(&self) -> DbResult<Vec<PendingDepositEventRecord>> {
Ok(self
.get_opt::<PendingDepositEventsCellOwned>(())?

36
lez/storage/src/sequencer/sequencer_cells.rs
View File

@ -9,7 +9,8 @@ use crate::{
sequencer::{
CF_LEE_STATE_NAME, DB_LEE_STATE_KEY, DB_META_LAST_FINALIZED_BLOCK_ID,
DB_META_LATEST_BLOCK_META_KEY, DB_META_PENDING_DEPOSIT_EVENTS_KEY,
DB_META_UNSEEN_WITHDRAW_COUNT_KEY, DB_META_ZONE_SDK_CHECKPOINT_KEY,
DB_META_UNSEEN_WITHDRAW_COUNT_KEY, DB_META_ZONE_CURSOR_KEY,
DB_META_ZONE_SDK_CHECKPOINT_KEY,
},
};
@ -132,6 +133,39 @@ impl SimpleWritableCell for ZoneSdkCheckpointCellRef<'_> {
}
}
/// The last channel block read back and verified from Bedrock.
///
/// Holds its L1 inscription `slot` plus the block's `id`/`hash`, and serves as
/// both the anchor for the startup consistency check and the resume point for
/// reconstruction. `slot` is stored as a raw `u64` because the zone-sdk `Slot`
/// does not derive borsh; the caller converts to/from `Slot`.
#[derive(Debug, Clone, Copy, BorshSerialize, BorshDeserialize)]
pub struct ZoneAnchorRecord {
pub slot: u64,
pub block_id: u64,
pub hash: HashType,
}
#[derive(Debug, BorshSerialize, BorshDeserialize)]
pub struct ZoneCursorCell(pub ZoneAnchorRecord);
impl SimpleStorableCell for ZoneCursorCell {
type KeyParams = ();
const CELL_NAME: &'static str = DB_META_ZONE_CURSOR_KEY;
const CF_NAME: &'static str = CF_META_NAME;
}
impl SimpleReadableCell for ZoneCursorCell {}
impl SimpleWritableCell for ZoneCursorCell {
fn value_constructor(&self) -> DbResult<Vec<u8>> {
borsh::to_vec(&self).map_err(|err| {
DbError::borsh_cast_message(err, Some("Failed to serialize zone cursor".to_owned()))
})
}
}
#[derive(Debug, Clone, PartialEq, Eq, BorshSerialize, BorshDeserialize)]
pub struct PendingDepositEventRecord {
pub deposit_op_id: HashType,