From 0428dae84d2a3280674786f9c94c2e34f84d79ae Mon Sep 17 00:00:00 2001 From: Daniil Polyakov Date: Thu, 9 Jul 2026 00:22:33 +0300 Subject: [PATCH] feat(sequencer): bootstrap state from Bedrock --- Cargo.lock | 21 ++ Cargo.toml | 2 + lez/chain_consistency/Cargo.toml | 26 ++ lez/chain_consistency/src/apply.rs | 182 ++++++++++ .../src/consistency.rs} | 310 +++++++----------- lez/chain_consistency/src/lib.rs | 9 + lez/indexer/core/Cargo.toml | 1 + lez/indexer/core/src/block_store.rs | 121 +------ lez/indexer/core/src/ingest_error.rs | 67 ---- lez/indexer/core/src/lib.rs | 121 ++++++- lez/sequencer/core/Cargo.toml | 1 + lez/sequencer/core/src/block_publisher.rs | 45 ++- lez/sequencer/core/src/block_store.rs | 12 +- lez/sequencer/core/src/lib.rs | 192 ++++++++++- lez/sequencer/core/src/mock.rs | 43 ++- lez/sequencer/core/src/tests.rs | 150 +++++++++ lez/storage/src/sequencer/mod.rs | 14 +- lez/storage/src/sequencer/sequencer_cells.rs | 36 +- 18 files changed, 958 insertions(+), 395 deletions(-) create mode 100644 lez/chain_consistency/Cargo.toml create mode 100644 lez/chain_consistency/src/apply.rs rename lez/{indexer/core/src/chain_consistency.rs => chain_consistency/src/consistency.rs} (58%) create mode 100644 lez/chain_consistency/src/lib.rs delete mode 100644 lez/indexer/core/src/ingest_error.rs diff --git a/Cargo.lock b/Cargo.lock index 0bd194bb..a40599be 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -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", diff --git a/Cargo.toml b/Cargo.toml index 3e71f293..5c53b6f4 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -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" } diff --git a/lez/chain_consistency/Cargo.toml b/lez/chain_consistency/Cargo.toml new file mode 100644 index 00000000..02bf0ea0 --- /dev/null +++ b/lez/chain_consistency/Cargo.toml @@ -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 diff --git a/lez/chain_consistency/src/apply.rs b/lez/chain_consistency/src/apply.rs new file mode 100644 index 00000000..e90d21a3 --- /dev/null +++ b/lez/chain_consistency/src/apply.rs @@ -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 + } + )); + } +} diff --git a/lez/indexer/core/src/chain_consistency.rs b/lez/chain_consistency/src/consistency.rs similarity index 58% rename from lez/indexer/core/src/chain_consistency.rs rename to lez/chain_consistency/src/consistency.rs index 3c3ab1e3..dce720b1 100644 --- a/lez/indexer/core/src/chain_consistency.rs +++ b/lez/chain_consistency/src/consistency.rs @@ -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 { - 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 { - 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( + node: &N, + channel_id: ChannelId, + anchor: &Anchor, +) -> Result +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, + 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) -> 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"), diff --git a/lez/chain_consistency/src/lib.rs b/lez/chain_consistency/src/lib.rs new file mode 100644 index 00000000..7ef25725 --- /dev/null +++ b/lez/chain_consistency/src/lib.rs @@ -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; diff --git a/lez/indexer/core/Cargo.toml b/lez/indexer/core/Cargo.toml index b3877639..ec0a0830 100644 --- a/lez/indexer/core/Cargo.toml +++ b/lez/indexer/core/Cargo.toml @@ -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 diff --git a/lez/indexer/core/src/block_store.rs b/lez/indexer/core/src/block_store.rs index 0fd34c36..17407745 100644 --- a/lez/indexer/core/src/block_store.rs +++ b/lez/indexer/core/src/block_store.rs @@ -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") diff --git a/lez/indexer/core/src/ingest_error.rs b/lez/indexer/core/src/ingest_error.rs deleted file mode 100644 index bda53497..00000000 --- a/lez/indexer/core/src/ingest_error.rs +++ /dev/null @@ -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 - } - )); - } -} diff --git a/lez/indexer/core/src/lib.rs b/lez/indexer/core/src/lib.rs index 32f0547d..04247140 100644 --- a/lez/indexer/core/src/lib.rs +++ b/lez/indexer/core/src/lib.rs @@ -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 { + 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 + )); + } +} diff --git a/lez/sequencer/core/Cargo.toml b/lez/sequencer/core/Cargo.toml index 2931c833..e48f23f2 100644 --- a/lez/sequencer/core/Cargo.toml +++ b/lez/sequencer/core/Cargo.toml @@ -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 diff --git a/lez/sequencer/core/src/block_publisher.rs b/lez/sequencer/core/src/block_publisher.rs index 21551131..b3483336 100644 --- a/lez/sequencer/core/src/block_publisher.rs +++ b/lez/sequencer/core/src/block_publisher.rs @@ -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) -> 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>; + + /// 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, + ) -> Result>; } /// 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)>, // Aborts the drive task when the last clone is dropped. _drive_task: Arc, @@ -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> { + 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, + ) -> Result> { + 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`. diff --git a/lez/sequencer/core/src/block_store.rs b/lez/sequencer/core/src/block_store.rs index 0db05d74..8d96425d 100644 --- a/lez/sequencer/core/src/block_store.rs +++ b/lez/sequencer/core/src/block_store.rs @@ -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> { + 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> { self.dbio.get_pending_deposit_events() } diff --git a/lez/sequencer/core/src/lib.rs b/lez/sequencer/core/src/lib.rs index 75dcc1a9..04a257e5 100644 --- a/lez/sequencer/core/src/lib.rs +++ b/lez/sequencer/core/src/lib.rs @@ -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 { 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 SequencerCore { /// 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 SequencerCore { 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 SequencerCore { .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 SequencerCore { (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 { + 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) -> block_publisher::CheckpointSink { Box::new(move |cp| { let bytes = match serde_json::to_vec(&cp) { diff --git a/lez/sequencer/core/src/mock.rs b/lez/sequencer/core/src/mock.rs index 39f635f9..65232685 100644 --- a/lez/sequencer/core/src/mock.rs +++ b/lez/sequencer/core/src/mock.rs @@ -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 #[derive(Clone)] pub struct MockBlockPublisher { channel_id: ChannelId, + /// Canned channel frontier returned by [`Self::channel_tip_slot`]. + tip_slot: Option, + /// 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, + messages: Vec<(ZoneMessage, Slot)>, + ) -> Self { + Self { + channel_id, + tip_slot, + messages, + } + } } impl BlockPublisherTrait for MockBlockPublisher { @@ -34,6 +56,8 @@ impl BlockPublisherTrait for MockBlockPublisher { ) -> Result { 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> { + Ok(self.tip_slot) + } + + async fn read_channel_after( + &self, + after_slot: Option, + ) -> Result> { + // Mirror `next_messages`: `after_slot` is exclusive. + Ok(self + .messages + .iter() + .filter(|(_, slot)| after_slot.is_none_or(|after| *slot > after)) + .cloned() + .collect()) + } } diff --git a/lez/sequencer/core/src/tests.rs b/lez/sequencer/core/src/tests.rs index a5c535a0..9125baca 100644 --- a/lez/sequencer/core/src/tests.rs +++ b/lez/sequencer/core/src/tests.rs @@ -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::::open_or_create_store(&config_b); + let mock_b = MockBlockPublisher::with_canned_channel(channel_id, Some(tip_slot), messages); + + let summary = SequencerCore::::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::::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::::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::::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::::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::::verify_and_reconstruct( + &mock, &mut store, &mut state, + ) + .await; + assert!(result.is_err(), "missing channel must abort startup"); + } +} diff --git a/lez/storage/src/sequencer/mod.rs b/lez/storage/src/sequencer/mod.rs index 44068517..13fcebdd 100644 --- a/lez/storage/src/sequencer/mod.rs +++ b/lez/storage/src/sequencer/mod.rs @@ -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> { + Ok(self.get_opt::(())?.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> { Ok(self .get_opt::(())? diff --git a/lez/storage/src/sequencer/sequencer_cells.rs b/lez/storage/src/sequencer/sequencer_cells.rs index 7672e271..302a0b97 100644 --- a/lez/storage/src/sequencer/sequencer_cells.rs +++ b/lez/storage/src/sequencer/sequencer_cells.rs @@ -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> { + 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,