From 7fd4e57fcbb396f5d345d8f6e6c6f5fae148b648 Mon Sep 17 00:00:00 2001 From: Daniil Polyakov Date: Tue, 7 Jul 2026 23:03:46 +0300 Subject: [PATCH] feat(sequencer): bootstrap state from Bedrock --- Cargo.lock | 22 +- Cargo.toml | 2 + integration_tests/tests/cross_zone_bridge.rs | 23 +- .../tests/cross_zone_ingress_guard.rs | 7 +- integration_tests/tests/cross_zone_ping.rs | 18 +- .../tests/cross_zone_verified.rs | 23 +- .../tests/sequencer_bootstrap.rs | 545 ++++++++++++++++++ integration_tests/tests/two_zone.rs | 12 +- lez/chain_consistency/Cargo.toml | 26 + lez/chain_consistency/src/apply.rs | 195 +++++++ .../src/consistency.rs} | 350 ++++------- lez/chain_consistency/src/lib.rs | 9 + lez/indexer/core/Cargo.toml | 2 +- lez/indexer/core/src/block_store.rs | 120 +--- lez/indexer/core/src/ingest_error.rs | 80 --- lez/indexer/core/src/lib.rs | 181 +++++- lez/indexer/core/src/stall_reason.rs | 25 - lez/indexer/core/src/status.rs | 27 +- lez/sequencer/core/Cargo.toml | 1 + lez/sequencer/core/src/block_publisher.rs | 45 +- lez/sequencer/core/src/block_store.rs | 80 ++- lez/sequencer/core/src/lib.rs | 249 +++++++- lez/sequencer/core/src/mock.rs | 43 +- lez/sequencer/core/src/tests.rs | 2 + .../core/src/tests/reconstruction.rs | 401 +++++++++++++ lez/storage/src/sequencer/mod.rs | 14 +- lez/storage/src/sequencer/sequencer_cells.rs | 36 +- .../src/bin/regenerate_test_fixture.rs | 19 +- test_fixtures/src/config.rs | 19 + test_fixtures/src/lib.rs | 25 +- test_fixtures/src/setup.rs | 186 +++--- tools/cross_zone_chat/src/main.rs | 24 +- 32 files changed, 2130 insertions(+), 681 deletions(-) create mode 100644 integration_tests/tests/sequencer_bootstrap.rs 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} (53%) create mode 100644 lez/chain_consistency/src/lib.rs delete mode 100644 lez/indexer/core/src/ingest_error.rs delete mode 100644 lez/indexer/core/src/stall_reason.rs create mode 100644 lez/sequencer/core/src/tests/reconstruction.rs diff --git a/Cargo.lock b/Cargo.lock index d6afde36..01f7b1df 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1368,6 +1368,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" @@ -3939,6 +3958,7 @@ dependencies = [ "arc-swap", "async-stream", "borsh", + "chain_consistency", "common", "cross_zone", "cross_zone_inbox_core", @@ -3958,7 +3978,6 @@ dependencies = [ "storage", "tempfile", "testnet_initial_state", - "thiserror 2.0.18", "tokio", "url", ] @@ -9023,6 +9042,7 @@ dependencies = [ "borsh", "bridge_core", "bytesize", + "chain_consistency", "chrono", "common", "cross_zone", diff --git a/Cargo.toml b/Cargo.toml index 164b64cc..bac0ea4f 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", @@ -72,6 +73,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/integration_tests/tests/cross_zone_bridge.rs b/integration_tests/tests/cross_zone_bridge.rs index 63b14c95..f835602a 100644 --- a/integration_tests/tests/cross_zone_bridge.rs +++ b/integration_tests/tests/cross_zone_bridge.rs @@ -17,7 +17,7 @@ use cross_zone_outbox_core::outbox_pda; use integration_tests::{ config::{self, SequencerPartialConfig}, indexer_client::IndexerClient, - setup::{setup_bedrock_node, setup_indexer, setup_sequencer}, + setup::{SequencerSetup, setup_bedrock_node, setup_indexer}, }; use lee::{ AccountId, PrivateKey, PublicKey, PublicTransaction, @@ -62,18 +62,19 @@ async fn lock_on_zone_a_mints_wrapped_token_on_zone_b() -> Result<()> { holder: holder_id, amount: INITIAL_BALANCE, }]; - let (seq_a, _seq_a_home) = setup_sequencer(partial, bedrock_addr, genesis_a, channel_a, None) + let (seq_a, _seq_a_home) = SequencerSetup::new(partial, bedrock_addr) + .with_channel_id(channel_a) + .with_genesis(genesis_a) + .setup() .await .context("Failed to set up zone A sequencer")?; - let (_seq_b, _seq_b_home) = setup_sequencer( - partial, - bedrock_addr, - vec![], - channel_b, - Some(cross_zone.clone()), - ) - .await - .context("Failed to set up zone B sequencer")?; + let (_seq_b, _seq_b_home) = SequencerSetup::new(partial, bedrock_addr) + .with_channel_id(channel_b) + .with_genesis(vec![]) + .with_cross_zone(cross_zone.clone()) + .setup() + .await + .context("Failed to set up zone B sequencer")?; let (idx_b, _idx_b_home) = setup_indexer(bedrock_addr, channel_b, Some(cross_zone)) .await .context("Failed to set up zone B indexer")?; diff --git a/integration_tests/tests/cross_zone_ingress_guard.rs b/integration_tests/tests/cross_zone_ingress_guard.rs index 5f401869..fe5fe7fc 100644 --- a/integration_tests/tests/cross_zone_ingress_guard.rs +++ b/integration_tests/tests/cross_zone_ingress_guard.rs @@ -18,7 +18,7 @@ use cross_zone_inbox_core::{ }; use integration_tests::{ config::{self, SequencerPartialConfig}, - setup::{setup_bedrock_node, setup_sequencer}, + setup::{SequencerSetup, setup_bedrock_node}, }; use lee::{ PublicTransaction, @@ -34,7 +34,10 @@ async fn user_origin_inbox_call_rejected() -> Result<()> { .context("Failed to set up Bedrock node")?; let partial = SequencerPartialConfig::default(); let channel = config::bedrock_channel_id(); - let (seq, _seq_home) = setup_sequencer(partial, bedrock_addr, vec![], channel, None) + let (seq, _seq_home) = SequencerSetup::new(partial, bedrock_addr) + .with_channel_id(channel) + .with_genesis(vec![]) + .setup() .await .context("Failed to set up sequencer")?; diff --git a/integration_tests/tests/cross_zone_ping.rs b/integration_tests/tests/cross_zone_ping.rs index 19223b5f..a3ed0c0c 100644 --- a/integration_tests/tests/cross_zone_ping.rs +++ b/integration_tests/tests/cross_zone_ping.rs @@ -18,7 +18,7 @@ use common::transaction::LeeTransaction; use cross_zone_outbox_core::outbox_pda; use integration_tests::{ config::{self, SequencerPartialConfig}, - setup::{setup_bedrock_node, setup_sequencer}, + setup::{SequencerSetup, setup_bedrock_node}, }; use lee::{AccountId, PublicTransaction, public_transaction::Message}; use lee_core::program::ProgramId; @@ -54,13 +54,19 @@ async fn ping_crosses_from_zone_a_to_zone_b() -> Result<()> { }], }; - let (seq_a, _seq_a_home) = setup_sequencer(partial, bedrock_addr, vec![], channel_a, None) + let (seq_a, _seq_a_home) = SequencerSetup::new(partial, bedrock_addr) + .with_channel_id(channel_a) + .with_genesis(vec![]) + .setup() .await .context("Failed to set up zone A sequencer")?; - let (seq_b, _seq_b_home) = - setup_sequencer(partial, bedrock_addr, vec![], channel_b, Some(cross_zone)) - .await - .context("Failed to set up zone B sequencer")?; + let (seq_b, _seq_b_home) = SequencerSetup::new(partial, bedrock_addr) + .with_channel_id(channel_b) + .with_genesis(vec![]) + .with_cross_zone(cross_zone) + .setup() + .await + .context("Failed to set up zone B sequencer")?; // Submit the ping on zone A, addressed to ping_receiver on zone B. let ping = build_ping_tx(zone_b, receiver_id); diff --git a/integration_tests/tests/cross_zone_verified.rs b/integration_tests/tests/cross_zone_verified.rs index e2f016c0..aa2ac06a 100644 --- a/integration_tests/tests/cross_zone_verified.rs +++ b/integration_tests/tests/cross_zone_verified.rs @@ -17,7 +17,7 @@ use cross_zone_outbox_core::outbox_pda; use integration_tests::{ config::{self, SequencerPartialConfig}, indexer_client::IndexerClient, - setup::{setup_bedrock_node, setup_indexer, setup_sequencer}, + setup::{SequencerSetup, setup_bedrock_node, setup_indexer}, }; use lee::{AccountId, PublicTransaction, public_transaction::Message}; use lee_core::program::ProgramId; @@ -53,21 +53,22 @@ async fn indexer_verifies_and_delivers_cross_zone_ping() -> Result<()> { // Zone A: source. Zone B: destination, with the watcher on its sequencer and // the verifier on its indexer. - let (seq_a, _seq_a_home) = setup_sequencer(partial, bedrock_addr, vec![], channel_a, None) + let (seq_a, _seq_a_home) = SequencerSetup::new(partial, bedrock_addr) + .with_channel_id(channel_a) + .with_genesis(vec![]) + .setup() .await .context("Failed to set up zone A sequencer")?; let (_idx_a, _idx_a_home) = setup_indexer(bedrock_addr, channel_a, None) .await .context("Failed to set up zone A indexer")?; - let (_seq_b, _seq_b_home) = setup_sequencer( - partial, - bedrock_addr, - vec![], - channel_b, - Some(cross_zone.clone()), - ) - .await - .context("Failed to set up zone B sequencer")?; + let (_seq_b, _seq_b_home) = SequencerSetup::new(partial, bedrock_addr) + .with_channel_id(channel_b) + .with_genesis(vec![]) + .with_cross_zone(cross_zone.clone()) + .setup() + .await + .context("Failed to set up zone B sequencer")?; let (idx_b, _idx_b_home) = setup_indexer(bedrock_addr, channel_b, Some(cross_zone)) .await .context("Failed to set up zone B indexer")?; diff --git a/integration_tests/tests/sequencer_bootstrap.rs b/integration_tests/tests/sequencer_bootstrap.rs new file mode 100644 index 00000000..5261c412 --- /dev/null +++ b/integration_tests/tests/sequencer_bootstrap.rs @@ -0,0 +1,545 @@ +//! End-to-end tests for the sequencer's startup bootstrap/reconstruction from +//! Bedrock (verify-and-reconstruct). Each test drives a real Bedrock node and one +//! or more sequencer instances sharing the same channel, exercising how a +//! sequencer reconciles its local store against what the channel serves. + +#![expect( + clippy::tests_outside_test_module, + reason = "Integration tests live at crate root and don't care about these lints" +)] + +use std::{net::SocketAddr, path::Path, time::Duration}; + +use anyhow::{Context as _, Result, bail}; +use indexer_service_rpc::RpcClient as _; +use integration_tests::L2_TO_L1_TIMEOUT; +use lee::{AccountId, PrivateKey, PublicKey}; +use logos_blockchain_core::mantle::ops::channel::ChannelId; +use sequencer_core::config::GenesisAction; +use sequencer_service_rpc::{RpcClient as _, SequencerClient, SequencerClientBuilder}; +use test_fixtures::{ + config::{SequencerPartialConfig, UrlProtocol, addr_to_url}, + indexer_client::IndexerClient, + setup::{SequencerSetup, setup_bedrock_node, setup_indexer}, +}; +use tokio::test; + +/// Block cadence for the tests: short so we don't wait long for local production. +fn fast_blocks() -> SequencerPartialConfig { + SequencerPartialConfig { + block_create_timeout: Duration::from_secs(2), + ..SequencerPartialConfig::default() + } +} + +/// Block cadence for a sequencer we don't want producing during the inspection +/// window, so its post-reconstruction tip stays put while we read it. +fn slow_blocks() -> SequencerPartialConfig { + SequencerPartialConfig { + block_create_timeout: Duration::from_secs(30), + ..SequencerPartialConfig::default() + } +} + +fn sequencer_client(addr: SocketAddr) -> Result { + let url = addr_to_url(UrlProtocol::Http, addr).context("Failed to build sequencer URL")?; + SequencerClientBuilder::default() + .build(url) + .context("Failed to build sequencer client") +} + +/// Polls the indexer's last finalized block id until it reaches `target`. The +/// indexer reads finalized channel history, so this is our oracle for "block +/// `target` is finalized on Bedrock" — exactly what a reconstructing sequencer +/// can read back. +async fn wait_for_finalized( + indexer: &IndexerClient, + target: u64, + timeout: Duration, +) -> Result { + let poll = async { + loop { + let finalized = indexer + .get_last_finalized_block_id() + .await + .context("Failed to read indexer last finalized block id")? + .unwrap_or(0); + if finalized >= target { + return Ok::<_, anyhow::Error>(finalized); + } + tokio::time::sleep(Duration::from_secs(1)).await; + } + }; + tokio::time::timeout(timeout, poll) + .await + .with_context(|| format!("Timed out waiting for indexer to finalize block {target}"))? +} + +/// Polls the sequencer's last block id until it reaches `target` or `timeout` elapses. +async fn wait_for_block_id( + client: &SequencerClient, + target: u64, + timeout: Duration, +) -> Result { + let poll = async { + loop { + let id = client + .get_last_block_id() + .await + .context("Failed to read sequencer last block id")?; + if id >= target { + return Ok::<_, anyhow::Error>(id); + } + tokio::time::sleep(Duration::from_millis(500)).await; + } + }; + tokio::time::timeout(timeout, poll) + .await + .with_context(|| format!("Timed out waiting for block id {target}"))? +} + +/// Best-effort extraction of a panic payload's message (panics carry a `String` +/// or `&str`), for asserting a startup aborted for the *expected* reason. +fn panic_message(payload: &(dyn std::any::Any + Send)) -> String { + payload + .downcast_ref::() + .cloned() + .or_else(|| payload.downcast_ref::<&str>().map(|s| (*s).to_owned())) + .unwrap_or_else(|| "".to_owned()) +} + +/// A `SupplyAccount` genesis action for a fresh account, returning the vault +/// account id the funds land in (genesis supply goes into a claimable vault, not +/// the account directly), so tests can assert genesis state is present. +fn supplied_account(balance: u128) -> (AccountId, GenesisAction) { + let account_id = AccountId::from(&PublicKey::new_from_private_key( + &PrivateKey::new_os_random(), + )); + let vault_id = vault_core::compute_vault_account_id(programs::vault().id(), account_id); + ( + vault_id, + GenesisAction::SupplyAccount { + account_id, + balance, + }, + ) +} + +/// Recursively copies the contents of `src` into `dst` (used to snapshot/restore a +/// sequencer's rocksdb directory while it is stopped). +fn copy_dir_recursive(src: &Path, dst: &Path) -> Result<()> { + std::fs::create_dir_all(dst) + .with_context(|| format!("Failed to create dir {}", dst.display()))?; + for entry in + std::fs::read_dir(src).with_context(|| format!("Failed to read dir {}", src.display()))? + { + let entry = entry.context("Failed to read dir entry")?; + let target = dst.join(entry.file_name()); + if entry + .file_type() + .context("Failed to read file type")? + .is_dir() + { + copy_dir_recursive(&entry.path(), &target)?; + } else { + std::fs::copy(entry.path(), &target) + .with_context(|| format!("Failed to copy into {}", target.display()))?; + } + } + Ok(()) +} + +/// Case 1: local store is empty and the Bedrock channel is empty. +/// +/// The sequencer bootstraps genesis state, finds nothing to reconstruct, publishes +/// its own genesis to open the channel, and starts producing. +#[test] +async fn empty_local_and_empty_bedrock_bootstraps_from_genesis() -> Result<()> { + let (_bedrock, bedrock_addr) = setup_bedrock_node() + .await + .context("Failed to setup Bedrock")?; + let home = tempfile::tempdir().context("Failed to create sequencer home")?; + + let (vault_id, supply) = supplied_account(12_345); + let genesis = vec![ + supply, + GenesisAction::SupplyBridgeAccount { balance: 1_000_000 }, + ]; + + let handle = SequencerSetup::new(fast_blocks(), bedrock_addr) + .with_genesis(genesis) + .setup_at(home.path()) + .await + .context("Failed to start sequencer")?; + let client = sequencer_client(handle.addr())?; + + // Fresh store + empty channel: startup bootstrapped genesis state directly. + assert_eq!( + client.get_account_balance(vault_id).await?, + 12_345, + "genesis-supplied vault balance must be present after bootstrap" + ); + + // The sequencer is live and producing on the freshly opened channel. + let last = wait_for_block_id(&client, 3, Duration::from_secs(60)).await?; + assert!( + last >= 3, + "sequencer should keep producing blocks, last={last}" + ); + assert!(handle.is_healthy(), "sequencer must stay healthy"); + + Ok(()) +} + +/// Case 2: local store is empty, but the Bedrock channel already has blocks. +/// +/// A first sequencer opens the channel and produces blocks; a second sequencer +/// starts from an empty store on the same channel (reusing the first's bedrock +/// signing key) and reconstructs the finalized history into its state. +#[test] +async fn empty_local_reconstructs_from_populated_bedrock() -> Result<()> { + const PRODUCED_TARGET: u64 = 3; + + let (_bedrock, bedrock_addr) = setup_bedrock_node() + .await + .context("Failed to setup Bedrock")?; + let (indexer_handle, _indexer_dir) = setup_indexer( + bedrock_addr, + test_fixtures::config::bedrock_channel_id(), + None, + ) + .await + .context("Failed to setup indexer")?; + let indexer_url = addr_to_url(UrlProtocol::Ws, indexer_handle.addr()) + .context("Failed to build indexer URL")?; + let indexer = IndexerClient::new(&indexer_url) + .await + .context("Failed to build indexer client")?; + + let (vault_id, supply) = supplied_account(7_777); + let genesis = vec![ + supply, + GenesisAction::SupplyBridgeAccount { balance: 1_000_000 }, + ]; + + // Sequencer A opens the channel and produces a few blocks. + let home_a = tempfile::tempdir().context("Failed to create sequencer A home")?; + let handle_a = SequencerSetup::new(fast_blocks(), bedrock_addr) + .with_genesis(genesis.clone()) + .setup_at(home_a.path()) + .await + .context("Failed to start sequencer A")?; + let client_a = sequencer_client(handle_a.addr())?; + wait_for_block_id(&client_a, PRODUCED_TARGET, Duration::from_secs(60)).await?; + + // Wait until those blocks are finalized on Bedrock — reconstruction only + // reads finalized history. A stays alive so its publish task keeps flushing. + let finalized = wait_for_finalized(&indexer, PRODUCED_TARGET, L2_TO_L1_TIMEOUT).await?; + + // Stop A, then wipe just its L2 store (keeping the bedrock signing key) so it + // restarts from an empty store on the same channel/identity — a sequencer that + // lost its local DB. + drop(handle_a); + tokio::time::sleep(Duration::from_secs(2)).await; + std::fs::remove_dir_all(home_a.path().join("rocksdb")) + .context("Failed to wipe sequencer L2 store")?; + + // Sequencer B restarts on the same home from that empty store and reconstructs. + let handle_b = SequencerSetup::new(slow_blocks(), bedrock_addr) + .with_genesis(genesis) + .setup_at(home_a.path()) + .await + .context("Failed to start sequencer B")?; + let client_b = sequencer_client(handle_b.addr())?; + + // Reconstruction ran synchronously during B's startup: even though its local + // store was empty, its tip is past genesis, matching the finalized channel. + let tip_b = client_b.get_last_block_id().await?; + assert!( + tip_b >= finalized, + "B should reconstruct at least the finalized blocks; tip_b={tip_b}, finalized={finalized}" + ); + assert!( + tip_b > 1, + "B should have reconstructed blocks beyond genesis; tip_b={tip_b}" + ); + + // Genesis state was rebuilt as part of the reconstruction. + assert_eq!( + client_b.get_account_balance(vault_id).await?, + 7_777, + "reconstructed genesis vault balance must be present" + ); + assert!(handle_b.is_healthy(), "sequencer B must stay healthy"); + + Ok(()) +} + +/// Case 3: local store is not empty, but the Bedrock channel is empty. +/// +/// A sequencer produces blocks (committing to a channel), is stopped, and is +/// restarted against a fresh/empty channel — i.e. the channel it committed to +/// was wiped or the node points at a different chain. Startup must fail rather +/// than silently resume onto a foreign channel. Crucially this must hold even +/// though the sequencer only ever *produced* (so it never recorded a per-block +/// anchor): the committed-but-missing-channel invariant catches it. +#[test] +async fn nonempty_local_against_empty_channel_fails_startup() -> Result<()> { + const PRODUCED_TARGET: u64 = 3; + + let (_bedrock, bedrock_addr) = setup_bedrock_node() + .await + .context("Failed to setup Bedrock")?; + + let (_vault_id, supply) = supplied_account(1); + let genesis = vec![ + supply, + GenesisAction::SupplyBridgeAccount { balance: 1_000_000 }, + ]; + + // A opens the channel and produces blocks. They land in its local store + // immediately, so no need to wait for finalization. + let home_a = tempfile::tempdir().context("Failed to create sequencer A home")?; + let handle_a = SequencerSetup::new(fast_blocks(), bedrock_addr) + .with_genesis(genesis.clone()) + .setup_at(home_a.path()) + .await + .context("Failed to start sequencer A")?; + wait_for_block_id( + &sequencer_client(handle_a.addr())?, + PRODUCED_TARGET, + Duration::from_secs(60), + ) + .await?; + drop(handle_a); + tokio::time::sleep(Duration::from_secs(2)).await; + + // Restart on the SAME home (A's committed store: blocks + checkpoint) but + // pointed at a fresh, never-used channel — the channel it committed to is gone. + let empty_channel = ChannelId::from([0x5a_u8; 32]); + + // Startup aborts on the missing-channel invariant (a panic in + // `start_from_config`). Run it on a dedicated OS thread with its own runtime + // so the panic is isolated to `join()` instead of failing the test thread. + // The `timeout` future must be created *inside* `block_on` (it needs a running + // reactor), so build it in an `async` block rather than as an eager argument. + let home_a_path = home_a.path().to_owned(); + let outcome = std::thread::spawn(move || { + let runtime = tokio::runtime::Runtime::new().expect("Failed to build runtime"); + runtime.block_on(async { + tokio::time::timeout( + Duration::from_secs(90), + SequencerSetup::new(slow_blocks(), bedrock_addr) + .with_channel_id(empty_channel) + .with_genesis(genesis) + .setup_at(&home_a_path), + ) + .await + }) + }) + .join(); + + match outcome { + // Expected: `start_from_config` panicked on the missing-channel invariant. + // Assert the *reason*, so an unrelated panic fails the test rather than + // masquerading as success. + Err(panic) => { + let message = panic_message(&*panic); + assert!( + message.contains("Refusing to resume onto a foreign channel"), + "startup panicked for an unexpected reason: {message}" + ); + } + Ok(Err(_elapsed)) => { + bail!("Sequencer startup hung instead of failing against an empty channel") + } + Ok(Ok(Err(err))) => { + bail!("Sequencer expected to panic, but it failed with error: {err:#?}") + } + Ok(Ok(Ok(_handle))) => { + bail!("Sequencer startup unexpectedly succeeded against an empty channel") + } + } + + Ok(()) +} + +/// Case 4: both non-empty, but the local store is *ahead* of the finalized channel. +/// +/// A sequencer produces blocks faster than Bedrock finalizes them (the normal +/// steady state), so on restart its local tip leads the channel's finalized tip. +/// Startup must succeed: reconstruction re-verifies the finalized blocks it already +/// holds and leaves the extra, not-yet-finalized local blocks untouched. +#[test] +async fn local_ahead_of_channel_resumes() -> Result<()> { + const FINALIZED_TARGET: u64 = 2; + + let (_bedrock, bedrock_addr) = setup_bedrock_node() + .await + .context("Failed to setup Bedrock")?; + let (indexer_handle, _indexer_dir) = setup_indexer( + bedrock_addr, + test_fixtures::config::bedrock_channel_id(), + None, + ) + .await + .context("Failed to setup indexer")?; + let indexer_url = addr_to_url(UrlProtocol::Ws, indexer_handle.addr()) + .context("Failed to build indexer URL")?; + let indexer = IndexerClient::new(&indexer_url) + .await + .context("Failed to build indexer client")?; + + let (vault_id, supply) = supplied_account(4_242); + let genesis = vec![ + supply, + GenesisAction::SupplyBridgeAccount { balance: 1_000_000 }, + ]; + + // A produces continuously (fast cadence) while Bedrock finalizes slowly, so + // its local tip runs well ahead of the channel's finalized tip. + let home = tempfile::tempdir().context("Failed to create sequencer home")?; + let handle_a = SequencerSetup::new(fast_blocks(), bedrock_addr) + .with_genesis(genesis.clone()) + .setup_at(home.path()) + .await + .context("Failed to start sequencer A")?; + let client_a = sequencer_client(handle_a.addr())?; + let finalized = wait_for_finalized(&indexer, FINALIZED_TARGET, L2_TO_L1_TIMEOUT).await?; + let tip_before = client_a.get_last_block_id().await?; + assert!( + tip_before > finalized, + "local tip {tip_before} should lead the finalized tip {finalized}" + ); + + // Restart on the same home; slow cadence so its tip stays put while we inspect. + drop(handle_a); + tokio::time::sleep(Duration::from_secs(2)).await; + let handle_b = SequencerSetup::new(slow_blocks(), bedrock_addr) + .with_genesis(genesis) + .setup_at(home.path()) + .await + .context("Failed to restart sequencer")?; + let client_b = sequencer_client(handle_b.addr())?; + + // Reconstruction verified the finalized prefix and preserved the extra blocks. + let tip_b = client_b.get_last_block_id().await?; + assert!( + tip_b >= tip_before, + "restart must not lose locally-produced blocks; tip_b={tip_b}, before={tip_before}" + ); + assert_eq!( + client_b.get_account_balance(vault_id).await?, + 4_242, + "genesis state must survive the restart" + ); + assert!(handle_b.is_healthy(), "sequencer must stay healthy"); + + Ok(()) +} + +/// Case 5: both non-empty, but the local store is *behind* the finalized channel. +/// +/// We snapshot a sequencer's store at an early tip, let it keep extending and +/// finalizing the channel, then restore the early snapshot and restart. Startup +/// must reconstruct forward — replay the finalized blocks the local store is +/// missing — catching the local tip up to the channel. +/// +/// The snapshot/restore is essential here (not a gratuitous copy): a live +/// sequencer's local tip always leads finalization, so the only way to obtain a +/// local store that *lags* the finalized channel is to preserve an earlier state +/// while the same channel advances past it. +#[test] +async fn local_behind_channel_reconstructs_forward() -> Result<()> { + const SNAPSHOT_TIP: u64 = 2; + const FINALIZED_TARGET: u64 = 4; + + let (_bedrock, bedrock_addr) = setup_bedrock_node() + .await + .context("Failed to setup Bedrock")?; + let (indexer_handle, _indexer_dir) = setup_indexer( + bedrock_addr, + test_fixtures::config::bedrock_channel_id(), + None, + ) + .await + .context("Failed to setup indexer")?; + let indexer_url = addr_to_url(UrlProtocol::Ws, indexer_handle.addr()) + .context("Failed to build indexer URL")?; + let indexer = IndexerClient::new(&indexer_url) + .await + .context("Failed to build indexer client")?; + + let (vault_id, supply) = supplied_account(5_005); + let genesis = vec![ + supply, + GenesisAction::SupplyBridgeAccount { balance: 1_000_000 }, + ]; + + let home = tempfile::tempdir().context("Failed to create sequencer home")?; + let rocksdb = home.path().join("rocksdb"); + + // Bring the sequencer up to an early tip, then stop it so its store is at rest. + { + let handle = SequencerSetup::new(slow_blocks(), bedrock_addr) + .with_genesis(genesis.clone()) + .setup_at(home.path()) + .await + .context("Failed to start sequencer")?; + wait_for_block_id( + &sequencer_client(handle.addr())?, + SNAPSHOT_TIP, + Duration::from_secs(120), + ) + .await?; + drop(handle); + tokio::time::sleep(Duration::from_secs(2)).await; + } + + // Snapshot the early store (tip == SNAPSHOT_TIP), safe because it is at rest. + let snapshot = tempfile::tempdir().context("Failed to create snapshot dir")?; + copy_dir_recursive(&rocksdb, snapshot.path()).context("Failed to snapshot store")?; + + // Resume the sequencer (fast cadence) so it extends and finalizes the channel + // beyond the snapshot. + let finalized = { + let handle = SequencerSetup::new(fast_blocks(), bedrock_addr) + .with_genesis(genesis.clone()) + .setup_at(home.path()) + .await + .context("Failed to resume sequencer")?; + let finalized = wait_for_finalized(&indexer, FINALIZED_TARGET, L2_TO_L1_TIMEOUT).await?; + drop(handle); + tokio::time::sleep(Duration::from_secs(2)).await; + finalized + }; + + // Restore the early snapshot: the local store now lags the finalized channel. + std::fs::remove_dir_all(&rocksdb).context("Failed to remove store before restore")?; + copy_dir_recursive(snapshot.path(), &rocksdb).context("Failed to restore snapshot")?; + + // Restart: reconstruction must catch the lagging store up to the channel. + let handle = SequencerSetup::new(slow_blocks(), bedrock_addr) + .with_genesis(genesis) + .setup_at(home.path()) + .await + .context("Failed to restart sequencer from a lagging store")?; + let client = sequencer_client(handle.addr())?; + let tip = client.get_last_block_id().await?; + assert!( + tip >= finalized, + "lagging store must reconstruct forward to the finalized tip; tip={tip}, finalized={finalized}" + ); + assert!( + tip > SNAPSHOT_TIP, + "reconstruction must advance beyond the snapshot tip; tip={tip}" + ); + assert_eq!( + client.get_account_balance(vault_id).await?, + 5_005, + "genesis state must be intact after reconstruction" + ); + assert!(handle.is_healthy(), "sequencer must stay healthy"); + + Ok(()) +} diff --git a/integration_tests/tests/two_zone.rs b/integration_tests/tests/two_zone.rs index c895acd1..8f4b2697 100644 --- a/integration_tests/tests/two_zone.rs +++ b/integration_tests/tests/two_zone.rs @@ -13,7 +13,7 @@ use indexer_service_rpc::RpcClient as _; use integration_tests::{ config::{self, SequencerPartialConfig}, indexer_client::IndexerClient, - setup::{setup_bedrock_node, setup_indexer, setup_sequencer}, + setup::{SequencerSetup, setup_bedrock_node, setup_indexer}, }; use sequencer_service_rpc::{RpcClient as _, SequencerClientBuilder}; use tokio::test; @@ -35,13 +35,19 @@ async fn two_zones_share_one_bedrock_and_both_advance() -> Result<()> { let channel_b = config::bedrock_channel_id_b(); // Empty genesis is enough: the clock transaction drives block production. - let (seq_a, _seq_a_home) = setup_sequencer(partial, bedrock_addr, vec![], channel_a, None) + let (seq_a, _seq_a_home) = SequencerSetup::new(partial, bedrock_addr) + .with_channel_id(channel_a) + .with_genesis(vec![]) + .setup() .await .context("Failed to set up zone A sequencer")?; let (idx_a, _idx_a_home) = setup_indexer(bedrock_addr, channel_a, None) .await .context("Failed to set up zone A indexer")?; - let (seq_b, _seq_b_home) = setup_sequencer(partial, bedrock_addr, vec![], channel_b, None) + let (seq_b, _seq_b_home) = SequencerSetup::new(partial, bedrock_addr) + .with_channel_id(channel_b) + .with_genesis(vec![]) + .setup() .await .context("Failed to set up zone B sequencer")?; let (idx_b, _idx_b_home) = setup_indexer(bedrock_addr, channel_b, None) 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..ea97d58a --- /dev/null +++ b/lez/chain_consistency/src/apply.rs @@ -0,0 +1,195 @@ +//! 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, + }, +} + +impl BlockIngestError { + /// Whether the failure may be transient rather than a property of the block. + /// + /// FIXME: `StateTransition` is too coarse — its `reason` string mixes genuine + /// state-transition rejections with infra failures (risc0 executor teardown, + /// storage errors). Once it carries a structured cause, narrow this so only + /// infra failures retry. + #[must_use] + pub const fn is_retryable(&self) -> bool { + matches!(self, Self::StateTransition { .. }) + } +} + +/// 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 53% rename from lez/indexer/core/src/chain_consistency.rs rename to lez/chain_consistency/src/consistency.rs index df4b5ba6..90aed632 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,26 @@ 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 { +#[derive(Debug, PartialEq, Eq)] +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,118 +176,106 @@ enum AnchorProbe { KeepLooking, } -#[expect( - clippy::multiple_inherent_impl, - reason = "split for clarity & isolation of relevant code" -)] -impl IndexerCore { - /// Verifies whether the channel still serves the same chain the store was built from. - /// This may change frequently during development where we reset the chain from time to - /// time in devnet/testnet, but we do not expect [`ChainConsistency::Inconsistent`] in - /// production. - /// - /// To compare the chains, we use an [`Anchor`] block that is either the parked L2 block - /// while stalled, or the tip L2 block at its own inscription L1 slot. - pub(crate) async fn verify_chain_consistency(&self) -> Result { - let Some(anchor) = self.get_startup_anchor()? else { - // empty or cold store: nothing to compare - return Ok(ChainConsistency::Inconclusive); - }; - - self.verify_chain_at_anchor(&anchor).await - } - - /// Builds the anchor for the startup check. - /// - /// - If stalled, returns the recorded _parked_ block - /// - If not stalled, returns the validated tip at its _own_ inscription slot. - /// - If the store is empty, returns `None`. - fn get_startup_anchor(&self) -> Result> { - if let Some(stall) = self.store.get_stall_reason()? { - return Ok(Some(Anchor { - slot: stall.l1_slot, - block: stall.block_id.zip(stall.block_hash), - })); +/// 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)); + } + } + Err(err) => { + warn!("Failed to read channel state for the consistency check: {err:#}"); } - - // not stalled, so anchor on the tip at its own inscription slot - let Some(slot) = self.store.get_tip_slot()?.or(self.store.get_zone_cursor()?) else { - return Ok(None); - }; - let Some(tip_id) = self.store.get_last_block_id()? else { - return Ok(None); - }; - let Some(tip) = self.store.get_block_at_id(tip_id)? else { - return Ok(None); - }; - Ok(Some(Anchor { - slot, - block: Some((tip_id, tip.header.hash)), - })) } - /// Verifies the channel still carries the anchor block at its slot. - /// - /// The anchor was finalized at `anchor.slot`, so the same chain must still - /// serve it there, while a reset chain re-inscribes its content only at - /// later wall-clock slots. - /// - /// Only positive evidence of a different chain yields `Inconsistent`. - /// Absence of data stays `Inconclusive`. - async fn verify_chain_at_anchor(&self, anchor: &Anchor) -> Result { - 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)); + // `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))); } - } - Err(err) => { - warn!("Failed to read channel state for the consistency check: {err:#}"); + AnchorProbe::Bail => return Ok(Some(ChainConsistency::Inconclusive)), + AnchorProbe::KeepLooking => { /* dont do anything */ } } } + Ok::<_, anyhow::Error>(None) + }; - // `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) - } + 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. /// /// The anchor block was finalized at `anchor_slot`, so on the same chain the @@ -297,35 +293,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, - cross_zone: None, - bridge_lock_holdings: Vec::new(), - }; - 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]), @@ -344,98 +318,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 - )); - } - - #[tokio::test] - async fn startup_anchor_prefers_tip_slot_over_lagging_cursor() { - // Cursor persist failures are warn-only, so the read cursor can lag the - // tip by several blocks. The anchor must pair the tip with its own - // inscription slot; pairing it with the stale cursor would make the scan - // misread the chain's intermediate blocks as re-inscriptions. - let dir = tempfile::tempdir().expect("tempdir"); - let core = unreachable_core(dir.path()); - - let genesis = common::test_utils::produce_dummy_block(1, None, vec![]); - core.store - .accept_block(&genesis, Slot::from(1_000)) - .await - .expect("accept"); - let block2 = common::test_utils::produce_dummy_block(2, Some(genesis.header.hash), vec![]); - core.store - .accept_block(&block2, Slot::from(1_005)) - .await - .expect("accept"); - let block3 = common::test_utils::produce_dummy_block(3, Some(block2.header.hash), vec![]); - core.store - .accept_block(&block3, Slot::from(1_010)) - .await - .expect("accept"); - - // Cursor last persisted at the genesis slot: two blocks behind the tip. - core.store - .set_zone_cursor(&Slot::from(1_000)) - .expect("set cursor"); - - let anchor = core.get_startup_anchor().expect("anchor").expect("present"); - assert_eq!(anchor.slot, Slot::from(1_010)); - assert_eq!(anchor.block, Some((3, block3.header.hash))); + Anchor::new(slot, Some((block.header.block_id, block.header.hash))) } #[test] @@ -516,10 +399,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 14941773..cbe0b7e8 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 @@ -32,7 +33,6 @@ futures.workspace = true url.workspace = true logos-blockchain-core.workspace = true serde_json.workspace = true -thiserror.workspace = true async-stream.workspace = true tokio.workspace = true risc0-zkvm.workspace = true diff --git a/lez/indexer/core/src/block_store.rs b/lez/indexer/core/src/block_store.rs index 8c974399..774aa87d 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::status::StallReason; /// Outcome of feeding a parsed L2 block to the validated tip. pub enum AcceptOutcome { @@ -258,14 +253,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) { if err.is_retryable() { return Ok(AcceptOutcome::RetryableFailure(err)); } @@ -290,112 +285,11 @@ 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}; #[tokio::test] async fn stall_reason_roundtrips_and_clears() { @@ -541,10 +435,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 299013d7..00000000 --- a/lez/indexer/core/src/ingest_error.rs +++ /dev/null @@ -1,80 +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, - }, -} - -impl BlockIngestError { - /// Whether the failure may be transient rather than a property of the block. - /// - /// FIXME: `StateTransition` is too coarse — its `reason` string mixes genuine - /// state-transition rejections with infra failures (risc0 executor teardown, - /// storage errors). Once it carries a structured cause, narrow this so only - /// infra failures retry. - #[must_use] - pub const fn is_retryable(&self) -> bool { - matches!(self, Self::StateTransition { .. }) - } -} - -#[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 7bee6387..f024ccf4 100644 --- a/lez/indexer/core/src/lib.rs +++ b/lez/indexer/core/src/lib.rs @@ -2,31 +2,29 @@ use std::{path::Path, sync::Arc}; use anyhow::Result; use arc_swap::ArcSwap; +pub use chain_consistency::BlockIngestError; +use chain_consistency::{Anchor, 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, }; use retry::ApplyRetryGate; -pub use stall_reason::StallReason; +pub use status::StallReason; use crate::{ block_store::{AcceptOutcome, IndexerStore}, - chain_consistency::ChainConsistency, config::IndexerConfig, cross_zone_verifier::CrossZoneVerifier, status::{IndexerStatus, IndexerSyncStatus}, }; + pub mod block_store; -pub mod chain_consistency; pub mod config; pub mod cross_zone_verifier; -pub mod ingest_error; mod retry; -pub mod stall_reason; pub mod status; /// Consecutive failed apply attempts of the same block before parking. @@ -125,6 +123,49 @@ impl IndexerCore { }) } + /// Verifies whether the channel still serves the same chain the store was built from. + /// This may change frequently during development where we reset the chain from time to + /// time in devnet/testnet, but we do not expect [`ChainConsistency::Inconsistent`] in + /// production. + /// + /// To compare the chains, we use an [`Anchor`] block that is either the parked L2 block + /// while stalled, or the tip L2 block at its own inscription L1 slot. + pub(crate) async fn verify_chain_consistency(&self) -> Result { + let Some(anchor) = self.get_startup_anchor()? else { + // empty or cold store: nothing to compare + return Ok(ChainConsistency::Inconclusive); + }; + + chain_consistency::verify_chain_consistency(&self.node, self.config.channel_id, &anchor) + .await + } + + /// Builds the anchor for the startup check. + /// + /// - If stalled, returns the recorded _parked_ block + /// - If not stalled, returns the validated tip at its _own_ inscription slot. + /// - If the store is empty, returns `None`. + fn get_startup_anchor(&self) -> Result> { + if let Some(stall) = self.store.get_stall_reason()? { + return Ok(Some(Anchor::new( + stall.l1_slot, + stall.block_id.zip(stall.block_hash), + ))); + } + + // not stalled, so anchor on the tip at its own inscription slot + let Some(slot) = self.store.get_tip_slot()?.or(self.store.get_zone_cursor()?) else { + return Ok(None); + }; + let Some(tip_id) = self.store.get_last_block_id()? else { + return Ok(None); + }; + let Some(tip) = self.store.get_block_at_id(tip_id)? else { + return Ok(None); + }; + Ok(Some(Anchor::new(slot, Some((tip_id, tip.header.hash))))) + } + /// 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 @@ -207,8 +248,8 @@ impl IndexerCore { let mut cursor = initial_cursor; let mut retry_gate = ApplyRetryGate::new(); - if cursor.is_some() { - info!("Resuming indexer from cursor {cursor:?}"); + if let Some(slot) = &cursor { + info!("Resuming indexer from cursor {slot:?}"); } else { info!("Starting indexer from beginning of channel"); } @@ -370,3 +411,127 @@ 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, + cross_zone: None, + bridge_lock_holdings: Vec::new(), + }; + 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 + )); + } + + #[tokio::test] + async fn startup_anchor_prefers_tip_slot_over_lagging_cursor() { + // Cursor persist failures are warn-only, so the read cursor can lag the + // tip by several blocks. The anchor must pair the tip with its own + // inscription slot; pairing it with the stale cursor would make the scan + // misread the chain's intermediate blocks as re-inscriptions. + let dir = tempfile::tempdir().expect("tempdir"); + let core = unreachable_core(dir.path()); + + let genesis = common::test_utils::produce_dummy_block(1, None, vec![]); + core.store + .accept_block(&genesis, Slot::from(1_000)) + .await + .expect("accept"); + let block2 = common::test_utils::produce_dummy_block(2, Some(genesis.header.hash), vec![]); + core.store + .accept_block(&block2, Slot::from(1_005)) + .await + .expect("accept"); + let block3 = common::test_utils::produce_dummy_block(3, Some(block2.header.hash), vec![]); + core.store + .accept_block(&block3, Slot::from(1_010)) + .await + .expect("accept"); + + // Cursor last persisted at the genesis slot: two blocks behind the tip. + core.store + .set_zone_cursor(&Slot::from(1_000)) + .expect("set cursor"); + + let anchor = core.get_startup_anchor().expect("anchor").expect("present"); + let expected = Anchor::new(Slot::from(1_010), Some((3, block3.header.hash))); + assert_eq!(anchor, expected); + } +} diff --git a/lez/indexer/core/src/stall_reason.rs b/lez/indexer/core/src/stall_reason.rs deleted file mode 100644 index c5118fcc..00000000 --- a/lez/indexer/core/src/stall_reason.rs +++ /dev/null @@ -1,25 +0,0 @@ -use common::HashType; -use logos_blockchain_zone_sdk::Slot; -use serde::{Deserialize, Serialize}; - -use crate::ingest_error::BlockIngestError; - -/// Diagnostic record of the first block that broke the L2 chain. -/// -/// The block-derived fields are `None` for a deserialize break (no header was -/// ever parsed). `l1_slot` is the L1 slot the breaking inscription was read at. -/// `first_seen` is the breaking block's L2 timestamp (`None` for a deserialize break). -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct StallReason { - pub block_id: Option, - pub block_hash: Option, - pub prev_block_hash: Option, - pub l1_slot: Slot, - pub error: BlockIngestError, - pub first_seen: Option, - /// Number of later non-chaining blocks (orphans, since the tip is frozen). - /// - /// TODO: We could store a different "branch" of blocks following this break, but for now we - /// just count them. - pub orphans_since: u64, -} diff --git a/lez/indexer/core/src/status.rs b/lez/indexer/core/src/status.rs index a483fde6..1307238b 100644 --- a/lez/indexer/core/src/status.rs +++ b/lez/indexer/core/src/status.rs @@ -1,6 +1,27 @@ -use serde::Serialize; +use chain_consistency::BlockIngestError; +use common::HashType; +use logos_blockchain_zone_sdk::Slot; +use serde::{Deserialize, Serialize}; -use crate::stall_reason::StallReason; +/// Diagnostic record of the first block that broke the L2 chain. +/// +/// The block-derived fields are `None` for a deserialize break (no header was +/// ever parsed). `l1_slot` is the L1 slot the breaking inscription was read at. +/// `first_seen` is the breaking block's L2 timestamp (`None` for a deserialize break). +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct StallReason { + pub block_id: Option, + pub block_hash: Option, + pub prev_block_hash: Option, + pub l1_slot: Slot, + pub error: BlockIngestError, + pub first_seen: Option, + /// Number of later non-chaining blocks (orphans, since the tip is frozen). + /// + /// TODO: We could store a different "branch" of blocks following this break, but for now we + /// just count them. + pub orphans_since: u64, +} /// Coarse lifecycle state of the indexer's ingestion loop, so a client can tell /// "still catching up" apart from "something went wrong". @@ -119,8 +140,6 @@ mod tests { fn stalled_status_serializes_with_stall_reason() { use logos_blockchain_zone_sdk::Slot; - use crate::{ingest_error::BlockIngestError, stall_reason::StallReason}; - let status = IndexerStatus { sync: IndexerSyncStatus::stalled("broken chain link".to_owned()), indexed_block_id: Some(41), diff --git a/lez/sequencer/core/Cargo.toml b/lez/sequencer/core/Cargo.toml index cd89f223..1fa7494d 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 77b57661..955ae7c0 100644 --- a/lez/sequencer/core/src/block_store.rs +++ b/lez/sequencer/core/src/block_store.rs @@ -12,7 +12,7 @@ use log::info; use logos_blockchain_zone_sdk::sequencer::SequencerCheckpoint; use storage::sequencer::{ RocksDBIO, - sequencer_cells::{PendingDepositEventRecord, WithdrawalReconciliationKey}, + sequencer_cells::{PendingDepositEventRecord, WithdrawalReconciliationKey, ZoneAnchorRecord}, }; pub use storage::{DbResult, sequencer::DbDump}; @@ -28,6 +28,45 @@ impl SequencerStore { /// Open existing database at the given location. Fails if no database is found. pub fn open_db(location: &Path, signing_key: lee::PrivateKey) -> DbResult { let dbio = Arc::new(RocksDBIO::open(location)?); + Self::from_dbio_and_signing_key(dbio, signing_key) + } + + /// Create a fresh rocksdb at `location` from `dump`. + pub fn restore_db_from_dump( + location: &Path, + dump: &DbDump, + signing_key: lee::PrivateKey, + ) -> DbResult { + let dbio = Arc::new(RocksDBIO::restore_from_dump(location, dump)?); + Self::from_dbio_and_signing_key(dbio, signing_key) + } + + /// Starting database at the start of new chain. + /// Creates files if necessary. + /// + /// ATTENTION: Will overwrite genesis block. + pub fn create_db_with_genesis( + location: &Path, + genesis_block: &Block, + genesis_state: &V03State, + signing_key: lee::PrivateKey, + ) -> DbResult { + let dbio = Arc::new(RocksDBIO::create(location, genesis_block, genesis_state)?); + let genesis_id = dbio.get_meta_first_block_in_db()?; + let tx_hash_to_block_map = block_to_transactions_map(genesis_block); + + Ok(Self { + dbio, + tx_hash_to_block_map, + genesis_id, + signing_key, + }) + } + + fn from_dbio_and_signing_key( + dbio: Arc, + signing_key: lee::PrivateKey, + ) -> DbResult { let genesis_id = dbio.get_meta_first_block_in_db()?; let last_id = dbio.latest_block_meta()?.id; @@ -53,28 +92,6 @@ impl SequencerStore { }) } - /// Starting database at the start of new chain. - /// Creates files if necessary. - /// - /// ATTENTION: Will overwrite genesis block. - pub fn create_db_with_genesis( - location: &Path, - genesis_block: &Block, - genesis_state: &V03State, - signing_key: lee::PrivateKey, - ) -> DbResult { - let dbio = Arc::new(RocksDBIO::create(location, genesis_block, genesis_state)?); - let genesis_id = dbio.get_meta_first_block_in_db()?; - let tx_hash_to_block_map = block_to_transactions_map(genesis_block); - - Ok(Self { - dbio, - tx_hash_to_block_map, - genesis_id, - signing_key, - }) - } - /// Shared handle to the underlying rocksdb. Used to persist the zone-sdk /// checkpoint from the sequencer's drive task without needing &mut to the /// store. @@ -165,13 +182,6 @@ impl SequencerStore { self.dbio.dump_all() } - /// Create a fresh rocksdb at `location` from `dump`, closing it before returning so a sequencer - /// can open it normally afterwards. - pub fn restore_db_from_dump(location: &Path, dump: &DbDump) -> DbResult<()> { - RocksDBIO::restore_from_dump(location, dump)?; - Ok(()) - } - pub fn get_zone_checkpoint(&self) -> Result> { let Some(bytes) = self.dbio.get_zone_sdk_checkpoint_bytes()? else { return Ok(None); @@ -188,6 +198,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 3a96e6d5..0fbb012d 100644 --- a/lez/sequencer/core/src/lib.rs +++ b/lez/sequencer/core/src/lib.rs @@ -2,6 +2,7 @@ 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}, @@ -9,11 +10,14 @@ use common::{ }; use config::{GenesisAction, SequencerConfig}; use itertools::Itertools as _; -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; @@ -21,7 +25,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::{ @@ -66,13 +70,20 @@ 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 /// assumed to represent the correct latest state consistent with Bedrock-finalized data. /// If no database is found, the sequencer performs a fresh start from genesis, /// initializing its state with the accounts defined in the configuration file. - fn open_or_create_store(config: &SequencerConfig) -> (SequencerStore, lee::V03State, Block) { + fn open_or_create_store(config: &SequencerConfig) -> (SequencerStore, lee::V03State) { let signing_key = lee::PrivateKey::try_new(config.signing_key).unwrap(); let db_path = config.home.join("rocksdb"); @@ -86,11 +97,7 @@ impl SequencerCore { let state = store .get_lee_state() .expect("Failed to read state from store"); - let genesis_block = store - .get_block_at_id(store.genesis_id()) - .expect("Failed to read genesis block from store") - .expect("Genesis block not found in store"); - (store, state, genesis_block) + (store, state) } else { warn!( "Database not found at {}, starting from genesis", @@ -115,7 +122,7 @@ impl SequencerCore { ) .expect("Failed to create database with genesis block"); - (store, genesis_state, genesis_block) + (store, genesis_state) } } @@ -126,11 +133,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) = Self::open_or_create_store(&config); let initial_checkpoint = store .get_zone_checkpoint() @@ -153,8 +156,28 @@ impl SequencerCore { .await .expect("Failed to initialize Block Publisher"); - // Fresh start (no checkpoint): republish all pending blocks - if is_fresh_start { + // Cross-zone messaging: start a watcher per configured peer. The inbox + // config account is seeded into genesis state in `build_genesis_state`. + if let Some(cross_zone) = &config.cross_zone { + cross_zone_watcher::spawn_watchers( + &config.bedrock_config, + cross_zone, + config.block_create_timeout, + &mempool_handle, + ); + } + // 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, is_fresh_start) + .await + .expect("Failed to verify/reconstruct sequencer state from Bedrock"); + + // Publish our blocks 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 { let mut pending_blocks = store .get_all_blocks() .filter_ok(|block| matches!(block.bedrock_status, BedrockStatus::Pending)) @@ -182,16 +205,9 @@ impl SequencerCore { } } - // Cross-zone messaging: start a watcher per configured peer. The inbox - // config account is seeded into genesis state in `build_genesis_state`. - if let Some(cross_zone) = &config.cross_zone { - cross_zone_watcher::spawn_watchers( - &config.bedrock_config, - cross_zone, - config.block_create_timeout, - &mempool_handle, - ); - } + let latest_block_meta = store + .latest_block_meta() + .expect("Failed to read latest block meta from store"); let sequencer_core = Self { state, @@ -205,6 +221,185 @@ 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, + is_fresh_start: bool, + ) -> 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")?; + let channel_tip_slot = publisher + .channel_tip_slot() + .await + .context("Failed to read channel tip slot")?; + + // If this sequencer has already committed blocks to the channel, that + // channel must still exist. A missing channel then means a wiped/rewound + // Bedrock or a node pointing at a different chain, so refuse to resume + // onto a foreign channel. + // + // "Committed" requires *both* a non-genesis tip and a checkpoint that was + // persisted before this startup: the tip alone is set the moment we produce + // (before the channel confirms it), while a checkpoint alone is written by + // zone-sdk's cold-start backfill even on a brand-new empty channel before we + // publish genesis. We must read the checkpoint presence from before `BP::new` + // ran (`!is_fresh_start`), because its cold-start backfill re-persists a + // checkpoint by the time we reach here — reading the store now would always + // see one. Together they mean we produced blocks and zone-sdk processed + // channel activity in a prior run. + let local_tip = store + .latest_block_meta() + .context("Failed to read latest block meta")? + .id; + let had_checkpoint_before_start = !is_fresh_start; + let has_committed_to_channel = local_tip > GENESIS_BLOCK_ID && had_checkpoint_before_start; + if has_committed_to_channel && channel_tip_slot.is_none() { + return Err(anyhow!( + "Sequencer holds committed blocks (tip {local_tip}) but the Bedrock channel \ + no longer exists on the connected chain — the channel was wiped or the node \ + points at a different chain. Refusing to resume onto a foreign channel." + )); + } + + // With a recorded anchor, additionally run the full consistency probe + // (frontier + reinscription/hash) for positive evidence of a different + // chain even when the channel exists. + if let Some(record) = anchor_record { + let anchor = Anchor::new( + Slot::from(record.slot), + Some((record.block_id, record.hash)), + ); + 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 05e2ec4c..bcc40a0a 100644 --- a/lez/sequencer/core/src/tests.rs +++ b/lez/sequencer/core/src/tests.rs @@ -37,6 +37,8 @@ use crate::{ mock::SequencerCoreWithMockClients, }; +mod reconstruction; + #[derive(borsh::BorshSerialize)] struct DepositMetadataForEncoding { recipient_id: lee::AccountId, diff --git a/lez/sequencer/core/src/tests/reconstruction.rs b/lez/sequencer/core/src/tests/reconstruction.rs new file mode 100644 index 00000000..04728bb5 --- /dev/null +++ b/lez/sequencer/core/src/tests/reconstruction.rs @@ -0,0 +1,401 @@ +#![expect( + clippy::arithmetic_side_effects, + clippy::as_conversions, + reason = "We don't care about it in tests" +)] + +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, config::GenesisAction, 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) = + 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, + true, + ) + .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, + true, + ) + .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) = 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, true, + ) + .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) = 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, true, + ) + .await; + assert!(result.is_err(), "missing channel must abort startup"); +} + +/// A sequencer config whose genesis funds the bridge account, so replayed bridge +/// deposit transactions have a source balance to mint from. +fn bridge_funded_config() -> SequencerConfig { + let mut config = setup_sequencer_config(); + config.genesis = vec![GenesisAction::SupplyBridgeAccount { balance: 1_000_000 }]; + config +} + +/// Builds an unfulfilled pending deposit event for `recipient`, matching the +/// encoding `build_bridge_deposit_tx_from_event` expects. +fn deposit_event_record( + op_id: [u8; 32], + amount: u64, + recipient: lee::AccountId, +) -> PendingDepositEventRecord { + PendingDepositEventRecord { + deposit_op_id: HashType(op_id), + source_tx_hash: HashType([0_u8; 32]), + amount, + metadata: borsh::to_vec(&DepositMetadataForEncoding { + recipient_id: recipient, + }) + .unwrap(), + submitted_in_block_id: None, + } +} + +/// Builds a signed public bridge `Withdraw` transaction (the normal user path). +fn build_public_withdraw_tx( + sender: lee::AccountId, + nonce: u128, + amount: u64, + bedrock_account_pk: [u8; 32], + signing_key: &lee::PrivateKey, +) -> LeeTransaction { + let message = lee::public_transaction::Message::try_new( + programs::bridge().id(), + vec![sender, system_accounts::bridge_account_id()], + vec![nonce.into()], + bridge_core::Instruction::Withdraw { + amount, + bedrock_account_pk, + }, + ) + .unwrap(); + let witness_set = lee::public_transaction::WitnessSet::for_message(&message, &[signing_key]); + LeeTransaction::Public(lee::PublicTransaction::new(message, witness_set)) +} + +/// The channel is ahead of the local store and its extra blocks carry +/// bridge deposit and withdraw transactions. Reconstruction must replay them, +/// reproducing both the minted/burned balances and the withdraw reconciliation +/// bookkeeping the production path records. +#[tokio::test] +async fn reconstructs_channel_with_deposit_and_withdraw() { + let recipient = initial_public_user_accounts()[0].account_id; + let deposit_amount = 500_u64; + let withdraw_amount = 100_u64; + let bedrock_account_pk = [0x22_u8; 32]; + let deposit_op_id = [0x0d_u8; 32]; + + // Sequencer A: a deposit block (sequencer-origin) then a withdraw block (user). + let config_a = bridge_funded_config(); + let (mut seq_a, mempool_a) = + SequencerCoreWithMockClients::start_from_config(config_a.clone()).await; + + let deposit_record = deposit_event_record(deposit_op_id, deposit_amount, recipient); + let deposit_tx = + crate::build_bridge_deposit_tx_from_event(&deposit_record).expect("build deposit tx"); + mempool_a + .push((TransactionOrigin::Sequencer, deposit_tx)) + .await + .unwrap(); + seq_a.produce_new_block().await.unwrap(); + + let withdraw_tx = build_public_withdraw_tx( + recipient, + 0, + withdraw_amount, + bedrock_account_pk, + &create_signing_key_for_account1(), + ); + mempool_a + .push((TransactionOrigin::User, withdraw_tx.clone())) + .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 reconstructs A's chain (bridge blocks included) from empty. + let config_b = bridge_funded_config(); + let (mut store_b, mut state_b) = + SequencerCore::::open_or_create_store(&config_b); + let mock_b = MockBlockPublisher::with_canned_channel(channel_id, Some(tip_slot), messages); + SequencerCore::::verify_and_reconstruct( + &mock_b, + &mut store_b, + &mut state_b, + true, + ) + .await + .expect("reconstruct"); + + 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); + + // The deposit minted into the vault and the withdraw burned from the sender — + // reconstructed balances agree with sequencer A for every affected account. + let vault_id = vault_core::compute_vault_account_id(programs::vault().id(), recipient); + let bridge_id = system_accounts::bridge_account_id(); + for account in [vault_id, bridge_id, recipient] { + assert_eq!( + state_b.get_account_by_id(account).balance, + seq_a.state().get_account_by_id(account).balance, + "reconstructed balance mismatch for {account:?}", + ); + } + assert_eq!( + state_b.get_account_by_id(vault_id).balance, + u128::from(deposit_amount), + "deposit must mint into the recipient vault" + ); + + // Reconstruction recorded the withdraw so the later L1 withdraw event reconciles. + let withdraw_arg = crate::extract_bridge_withdraw_data(&withdraw_tx).expect("withdraw data"); + let key = crate::withdraw_event_reconciliation_key(&withdraw_arg.outputs).expect("recon key"); + assert!( + store_b.dbio().consume_unseen_withdraw_count(key).unwrap(), + "reconstruction must record an unseen withdraw for bridge reconciliation" + ); +} + +/// A deposit whose L1 event was observed (an unfulfilled pending record +/// exists) and whose L2 mint is already contained in a finalized channel block. +/// Reconstruction must reconcile the pending record against that block — marking +/// it submitted so the startup replay does not re-inject it — and apply the mint +/// exactly once. +#[tokio::test] +async fn reconstruction_reconciles_already_finished_deposit() { + let recipient = initial_public_user_accounts()[0].account_id; + let deposit_amount = 400_u64; + let deposit_op_id = [0x1a_u8; 32]; + + // Sequencer A: a single block that fully processes the bridge deposit. + let config_a = bridge_funded_config(); + let (mut seq_a, mempool_a) = + SequencerCoreWithMockClients::start_from_config(config_a.clone()).await; + let deposit_record = deposit_event_record(deposit_op_id, deposit_amount, recipient); + let deposit_tx = + crate::build_bridge_deposit_tx_from_event(&deposit_record).expect("build deposit tx"); + mempool_a + .push((TransactionOrigin::Sequencer, deposit_tx)) + .await + .unwrap(); + seq_a.produce_new_block().await.unwrap(); + let deposit_block_id = seq_a.block_store().latest_block_meta().unwrap().id; + + 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: fresh store, but with the *unfulfilled* pending deposit event + // pre-seeded, as the cold-start backfill would when it re-observes this + // already-finalized deposit. + let config_b = bridge_funded_config(); + let (mut store_b, mut state_b) = + SequencerCore::::open_or_create_store(&config_b); + assert!( + store_b + .dbio() + .add_pending_deposit_event(deposit_record.clone()) + .unwrap() + ); + + let mock_b = MockBlockPublisher::with_canned_channel(channel_id, Some(tip_slot), messages); + SequencerCore::::verify_and_reconstruct( + &mock_b, + &mut store_b, + &mut state_b, + true, + ) + .await + .expect("reconstruct"); + + // The mint was applied exactly once. + let vault_id = vault_core::compute_vault_account_id(programs::vault().id(), recipient); + assert_eq!( + state_b.get_account_by_id(vault_id).balance, + u128::from(deposit_amount), + "already-finished deposit must be applied exactly once" + ); + + // The pending event is now marked submitted in the reconstructed block, so the + // startup replay would not re-queue it — no double mint on restart. + let record = store_b + .get_unfulfilled_deposit_events() + .unwrap() + .into_iter() + .find(|event| event.deposit_op_id == HashType(deposit_op_id)) + .expect("pending deposit event should still be recorded"); + assert_eq!( + record.submitted_in_block_id, + Some(deposit_block_id), + "reconstruction must reconcile the already-finished deposit against its channel block" + ); +} + +#[tokio::test] +async fn committed_local_against_missing_channel_fails_without_anchor() { + // A sequencer that has committed blocks — a non-genesis tip plus a persisted + // checkpoint — but only ever produced (so it never recorded a per-block + // anchor). Restarting it against a wiped/missing channel must still fail, + // driven by the committed-blocks invariant rather than an anchor probe. + let config = setup_sequencer_config(); + { + let (mut seq, _handle) = + SequencerCoreWithMockClients::start_from_config(config.clone()).await; + seq.produce_new_block().await.unwrap(); + seq.produce_new_block().await.unwrap(); + assert!(seq.block_store().latest_block_meta().unwrap().id > 1); + } // drop releases the store so we can reopen it + + // Reopen: blocks beyond genesis, no anchor. `is_fresh_start = false` stands in + // for a checkpoint persisted by a prior sync (the mock never emits one). + let (mut store, mut state) = SequencerCore::::open_or_create_store(&config); + assert!(store.get_zone_anchor().unwrap().is_none()); + assert!(store.latest_block_meta().unwrap().id > 1); + + // The channel is gone: no tip, no messages. + let mock = + MockBlockPublisher::with_canned_channel(config.bedrock_config.channel_id, None, vec![]); + let result = SequencerCore::::verify_and_reconstruct( + &mock, &mut store, &mut state, false, + ) + .await; + assert!( + result.is_err(), + "committed blocks against a missing channel must abort startup" + ); +} diff --git a/lez/storage/src/sequencer/mod.rs b/lez/storage/src/sequencer/mod.rs index 1a3a1886..f8557771 100644 --- a/lez/storage/src/sequencer/mod.rs +++ b/lez/storage/src/sequencer/mod.rs @@ -22,7 +22,7 @@ use crate::{ LEEStateCellOwned, LEEStateCellRef, LastFinalizedBlockIdCell, LatestBlockMetaCellOwned, LatestBlockMetaCellRef, PendingDepositEventRecord, PendingDepositEventsCellOwned, PendingDepositEventsCellRef, UnseenWithdrawCountCell, WithdrawalReconciliationKey, - ZoneSdkCheckpointCellOwned, ZoneSdkCheckpointCellRef, + ZoneAnchorRecord, ZoneCursorCell, ZoneSdkCheckpointCellOwned, ZoneSdkCheckpointCellRef, }, }; @@ -34,6 +34,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"; @@ -359,6 +363,14 @@ impl RocksDBIO { self.del::(()) } + 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, diff --git a/test_fixtures/src/bin/regenerate_test_fixture.rs b/test_fixtures/src/bin/regenerate_test_fixture.rs index 80da8fdc..ec010b52 100644 --- a/test_fixtures/src/bin/regenerate_test_fixture.rs +++ b/test_fixtures/src/bin/regenerate_test_fixture.rs @@ -11,9 +11,9 @@ use sequencer_core::block_store::SequencerStore; use test_fixtures::{ config, setup::{ - prebuilt_sequencer_db_dump_path, setup_bedrock_node, + SequencerSetup, prebuilt_sequencer_db_dump_path, setup_bedrock_node, setup_private_accounts_with_initial_supply, setup_public_accounts_with_initial_supply, - setup_sequencer, setup_wallet, + setup_wallet, }, }; use wallet::config::WalletConfigOverrides; @@ -49,15 +49,12 @@ async fn generate_prebuilt_fixture(dest: &Path) -> Result<()> { let genesis = config::genesis_from_accounts(&initial_public_accounts, &initial_private_accounts); - let (sequencer_handle, temp_sequencer_dir) = setup_sequencer( - config::SequencerPartialConfig::default(), - bedrock_addr, - genesis, - config::bedrock_channel_id(), - None, - ) - .await - .context("Failed to setup Sequencer for fixture generation")?; + let (sequencer_handle, temp_sequencer_dir) = + SequencerSetup::new(config::SequencerPartialConfig::default(), bedrock_addr) + .with_genesis(genesis) + .setup() + .await + .context("Failed to setup Sequencer for fixture generation")?; let (mut wallet, _temp_wallet_dir, _wallet_password) = setup_wallet( sequencer_handle.addr(), diff --git a/test_fixtures/src/config.rs b/test_fixtures/src/config.rs index 960f349b..de670a2f 100644 --- a/test_fixtures/src/config.rs +++ b/test_fixtures/src/config.rs @@ -80,7 +80,26 @@ pub fn sequencer_config( home: PathBuf, bedrock_addr: SocketAddr, genesis_transactions: Vec, + cross_zone: Option, +) -> Result { + sequencer_config_with_channel( + partial, + home, + bedrock_addr, + bedrock_channel_id(), + genesis_transactions, + cross_zone, + ) +} + +/// Like [`sequencer_config`] but with an explicit Bedrock `channel_id`, so tests +/// can point a sequencer at a fresh/empty channel (e.g. to model a wiped Bedrock). +pub fn sequencer_config_with_channel( + partial: SequencerPartialConfig, + home: PathBuf, + bedrock_addr: SocketAddr, channel_id: ChannelId, + genesis_transactions: Vec, cross_zone: Option, ) -> Result { let SequencerPartialConfig { diff --git a/test_fixtures/src/lib.rs b/test_fixtures/src/lib.rs index ff87c8e9..97d93bd4 100644 --- a/test_fixtures/src/lib.rs +++ b/test_fixtures/src/lib.rs @@ -24,8 +24,8 @@ use wallet::{ use crate::{ indexer_client::IndexerClient, setup::{ - setup_bedrock_node, setup_indexer, setup_private_accounts_with_initial_supply, - setup_public_accounts_with_initial_supply, setup_sequencer, setup_sequencer_from_prebuilt, + SequencerSetup, setup_bedrock_node, setup_indexer, + setup_private_accounts_with_initial_supply, setup_public_accounts_with_initial_supply, setup_wallet, sync_wallet_from_prebuilt, }, }; @@ -359,11 +359,8 @@ impl TestContextBuilder { let partial_config = sequencer_partial_config.unwrap_or_default(); - let (sequencer_handle, temp_sequencer_dir) = if use_prebuilt { - setup_sequencer_from_prebuilt(partial_config, bedrock_addr) - .await - .context("Failed to setup Sequencer from prebuilt database")? - } else { + let mut sequencer_setup = SequencerSetup::new(partial_config, bedrock_addr); + if !use_prebuilt { // Wallet genesis must always be present so that // setup_public/private_accounts_with_initial_supply can claim from the vault PDAs. // When a test supplies custom genesis, merge rather than replace. @@ -376,16 +373,12 @@ impl TestContextBuilder { } None => wallet_genesis, }; - setup_sequencer( - partial_config, - bedrock_addr, - genesis, - config::bedrock_channel_id(), - None, - ) + sequencer_setup = sequencer_setup.with_genesis(genesis); + } + let (sequencer_handle, temp_sequencer_dir) = sequencer_setup + .setup() .await - .context("Failed to setup Sequencer")? - }; + .context("Failed to setup Sequencer")?; let (mut wallet, temp_wallet_dir, wallet_password) = setup_wallet( sequencer_handle.addr(), diff --git a/test_fixtures/src/setup.rs b/test_fixtures/src/setup.rs index 325e1628..5d397856 100644 --- a/test_fixtures/src/setup.rs +++ b/test_fixtures/src/setup.rs @@ -1,4 +1,7 @@ -use std::{net::SocketAddr, path::PathBuf}; +use std::{ + net::SocketAddr, + path::{Path, PathBuf}, +}; use anyhow::{Context as _, Result, bail}; use indexer_service::{ChannelId, IndexerHandle}; @@ -20,12 +23,106 @@ use crate::{ private_mention, public_mention, }; -/// How to initialize the sequencer's database. -pub enum SequencerInit<'dump> { - /// Apply these genesis actions from scratch. - Genesis(Vec), - /// Restore an existing chain from a prebuilt dump (genesis actions are then irrelevant). - Prebuilt(&'dump DbDump), +pub struct SequencerSetup { + partial: config::SequencerPartialConfig, + bedrock_addr: SocketAddr, + channel_id: ChannelId, + genesis_transactions: Option>, + cross_zone: Option, +} + +impl SequencerSetup { + #[must_use] + pub fn new(partial: config::SequencerPartialConfig, bedrock_addr: SocketAddr) -> Self { + Self { + partial, + bedrock_addr, + channel_id: config::bedrock_channel_id(), + genesis_transactions: None, + cross_zone: None, + } + } + + /// Set the Bedrock channel ID to use for the sequencer. + /// If not set, the default channel ID from the Bedrock config will be used. + #[must_use] + pub const fn with_channel_id(mut self, channel_id: ChannelId) -> Self { + self.channel_id = channel_id; + self + } + + /// Set the cross-zone configuration to use for the sequencer. + /// If not set, the sequencer will be configured to run in single-zone mode. + #[must_use] + pub fn with_cross_zone(mut self, cross_zone: sequencer_core::config::CrossZoneConfig) -> Self { + self.cross_zone = Some(cross_zone); + self + } + + /// Set the genesis transactions to apply when initializing the sequencer. + /// If not set, the sequencer will be initialized from a prebuilt database dump. + #[must_use] + pub fn with_genesis(mut self, genesis_transactions: Vec) -> Self { + self.genesis_transactions = Some(genesis_transactions); + self + } + + /// Set up the sequencer in a fresh temporary home directory, returning the + /// owning [`TempDir`] alongside the handle. + pub async fn setup(self) -> Result<(SequencerHandle, TempDir)> { + let temp_sequencer_dir = + tempfile::tempdir().context("Failed to create temp dir for sequencer home")?; + + let sequencer_handle = self.setup_at(temp_sequencer_dir.path()).await?; + + Ok((sequencer_handle, temp_sequencer_dir)) + } + + /// Set up the sequencer in an explicit `home` directory owned by the caller. + /// + /// Useful for tests that restart the sequencer against the same on-disk store. + pub async fn setup_at(self, home: &Path) -> Result { + let Self { + partial, + bedrock_addr, + channel_id, + genesis_transactions, + cross_zone, + } = self; + + debug!("Using sequencer home at {}", home.display()); + + let genesis_transactions = if let Some(genesis) = genesis_transactions { + genesis + } else { + let dump = load_prebuilt_dump()?; + // `SequencerCore::open_or_create_store` looks for `/rocksdb`. + let dst = home.join("rocksdb"); + let _store = SequencerStore::restore_db_from_dump( + &dst, + &dump, + lee::PrivateKey::try_new(config::SEQUENCER_SIGNING_KEY)?, + ) + .context("Failed to restore prebuilt sequencer database from dump")?; + // TODO: Technically not correct, we should reconstruct the genesis transactions + // from the dump, but this crutch doesn't affect anything for now + Vec::new() + }; + + let config = config::sequencer_config_with_channel( + partial, + home.to_owned(), + bedrock_addr, + channel_id, + genesis_transactions, + cross_zone, + ) + .context("Failed to create Sequencer config")?; + + sequencer_service::run(config, 0) + .await + .context("Failed to run Sequencer Service") + } } /// Committed single-file dump of the prebuilt sequencer database (`just regenerate-test-fixture`). @@ -139,81 +236,6 @@ pub async fn setup_indexer( .map(|handle| (handle, temp_indexer_dir)) } -pub async fn setup_sequencer( - partial: config::SequencerPartialConfig, - bedrock_addr: SocketAddr, - genesis_transactions: Vec, - channel_id: ChannelId, - cross_zone: Option, -) -> Result<(SequencerHandle, TempDir)> { - setup_sequencer_inner( - partial, - bedrock_addr, - SequencerInit::Genesis(genesis_transactions), - channel_id, - cross_zone, - ) - .await -} - -/// Like [`setup_sequencer`], but rebuilds the sequencer database from the committed prebuilt dump -/// so it starts from an existing chain instead of applying genesis from scratch. -pub async fn setup_sequencer_from_prebuilt( - partial: config::SequencerPartialConfig, - bedrock_addr: SocketAddr, -) -> Result<(SequencerHandle, TempDir)> { - let dump = load_prebuilt_dump()?; - setup_sequencer_inner( - partial, - bedrock_addr, - SequencerInit::Prebuilt(&dump), - config::bedrock_channel_id(), - None, - ) - .await -} - -async fn setup_sequencer_inner( - partial: config::SequencerPartialConfig, - bedrock_addr: SocketAddr, - init: SequencerInit<'_>, - channel_id: ChannelId, - cross_zone: Option, -) -> Result<(SequencerHandle, TempDir)> { - let temp_sequencer_dir = - tempfile::tempdir().context("Failed to create temp dir for sequencer home")?; - - debug!( - "Using temp sequencer home at {}", - temp_sequencer_dir.path().display() - ); - - let genesis_transactions = match init { - SequencerInit::Genesis(genesis) => genesis, - SequencerInit::Prebuilt(dump) => { - // `SequencerCore::open_or_create_store` looks for `/rocksdb`. - let dst = temp_sequencer_dir.path().join("rocksdb"); - SequencerStore::restore_db_from_dump(&dst, dump) - .context("Failed to restore prebuilt sequencer database from dump")?; - Vec::new() - } - }; - - let config = config::sequencer_config( - partial, - temp_sequencer_dir.path().to_owned(), - bedrock_addr, - genesis_transactions, - channel_id, - cross_zone, - ) - .context("Failed to create Sequencer config")?; - - let sequencer_handle = sequencer_service::run(config, 0).await?; - - Ok((sequencer_handle, temp_sequencer_dir)) -} - pub fn setup_wallet( sequencer_addr: SocketAddr, initial_public_accounts: &[(PrivateKey, u128)], diff --git a/tools/cross_zone_chat/src/main.rs b/tools/cross_zone_chat/src/main.rs index 5a066e9e..1606c201 100644 --- a/tools/cross_zone_chat/src/main.rs +++ b/tools/cross_zone_chat/src/main.rs @@ -66,7 +66,7 @@ use sequencer_service_rpc::{RpcClient as _, SequencerClient, SequencerClientBuil use serde::{Deserialize, Serialize}; use test_fixtures::{ config::{self, SequencerPartialConfig, UrlProtocol, bedrock_channel_id, bedrock_channel_id_b}, - setup::{setup_bedrock_node, setup_sequencer}, + setup::{SequencerSetup, setup_bedrock_node}, }; const HTTP_PORT: u16 = 8088; @@ -289,14 +289,20 @@ async fn main() -> Result<()> { let cross_zone_b = watch_peer(zone_a, receiver_id); let partial = SequencerPartialConfig::default(); - let (seq_a, _home_a) = - setup_sequencer(partial, bedrock_addr, vec![], channel_a, Some(cross_zone_a)) - .await - .context("Failed to set up zone A sequencer")?; - let (seq_b, _home_b) = - setup_sequencer(partial, bedrock_addr, vec![], channel_b, Some(cross_zone_b)) - .await - .context("Failed to set up zone B sequencer")?; + let (seq_a, _home_a) = SequencerSetup::new(partial, bedrock_addr) + .with_genesis(vec![]) + .with_channel_id(channel_a) + .with_cross_zone(cross_zone_a) + .setup() + .await + .context("Failed to set up zone A sequencer")?; + let (seq_b, _home_b) = SequencerSetup::new(partial, bedrock_addr) + .with_genesis(vec![]) + .with_channel_id(channel_b) + .with_cross_zone(cross_zone_b) + .setup() + .await + .context("Failed to set up zone B sequencer")?; let state = Arc::new(AppState { zone_a: ZoneRuntime {