logos-blockchain/logos-execution-zone
1
0
Fork 0
mirror of https://github.com/logos-blockchain/logos-execution-zone.git synced 2026-07-15 10:20:01 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(sequencer): bootstrap state from Bedrock

Browse Source
This commit is contained in:
Daniil Polyakov 2026-07-07 23:03:46 +03:00
parent 40595604a4
commit 7fd4e57fcb
32 changed files with 2130 additions and 681 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

22
Cargo.lock generated
View File

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

2
Cargo.toml
View File

@ -32,6 +32,7 @@ members = [
"lez/wallet",
"lez/wallet-ffi",
"lez/common",
"lez/chain_consistency",
"lez/programs",
"lez/programs/amm",
"lez/programs/associated_token_account",
@ -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" }

23
integration_tests/tests/cross_zone_bridge.rs
View File

@ -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")?;

7
integration_tests/tests/cross_zone_ingress_guard.rs
View File

@ -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")?;

18
integration_tests/tests/cross_zone_ping.rs
View File

@ -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);

23
integration_tests/tests/cross_zone_verified.rs
View File

@ -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")?;

545
integration_tests/tests/sequencer_bootstrap.rs Normal file
View File

@ -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<SequencerClient> {
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<u64> {
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<u64> {
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::<String>()
.cloned()
.or_else(|| payload.downcast_ref::<&str>().map(|s| (*s).to_owned()))
.unwrap_or_else(|| "<non-string panic payload>".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(())
}

12
integration_tests/tests/two_zone.rs
View File

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

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

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

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

@ -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
}
));
}
}

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

@ -1,18 +1,18 @@
//! Startup check that the local store still belongs to the chain the
//! connected channel serves.
//! Startup check that a local store still belongs to the chain the connected
//! channel serves.
use anyhow::Result;
use common::{HashType, block::Block};
use futures::StreamExt as _;
use lee_core::BlockId;
use log::warn;
use logos_blockchain_zone_sdk::{Slot, ZoneMessage, adapter::Node as _};
use crate::IndexerCore;
use logos_blockchain_core::mantle::ops::channel::ChannelId;
use logos_blockchain_zone_sdk::{Slot, ZoneMessage, adapter, indexer::ZoneIndexer};
/// Upper bound on the channel reads of the startup consistency check.
const CHANNEL_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(60);
/// Result of comparing the indexer's stored chain against the channel.
/// Result of comparing a caller's stored chain against the channel.
pub enum ChainConsistency {
/// Channel still serves our anchor block (the stored tip position, or the
/// parked block while stalled).
@ -24,7 +24,7 @@ pub enum ChainConsistency {
/// - or the channel read was inconclusive (timeout / error / empty stream)
///
/// NOTE: None of these prove a reset, so the caller proceeds.
/// A genuine divergence is still caught later when the ingest loop tries to apply and parks.
/// A genuine divergence is still caught later when applying the channel history.
Inconclusive,
/// Positive evidence that the channel is a different chain than the store.
///
@ -36,13 +36,13 @@ pub enum ChainConsistency {
pub enum ChainMismatch {
/// The channel serves a different block at the anchor's id.
Block {
ours: (u64, HashType),
channel: (u64, HashType),
ours: (BlockId, HashType),
channel: (BlockId, HashType),
},
/// The channel serves a block at/below the anchor's id past the anchor
/// slot; on the same chain those ids live at earlier slots.
ReinscribedBlock {
channel: (u64, HashType),
channel: (BlockId, HashType),
slot: Slot,
anchor_slot: Slot,
},
@ -97,18 +97,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<ChainConsistency> {
let Some(anchor) = self.get_startup_anchor()? else {
// empty or cold store: nothing to compare
return Ok(ChainConsistency::Inconclusive);
};
self.verify_chain_at_anchor(&anchor).await
}
/// Builds the anchor for the startup check.
///
/// - If stalled, returns the recorded _parked_ block
/// - If not stalled, returns the validated tip at its _own_ inscription slot.
/// - If the store is empty, returns `None`.
fn get_startup_anchor(&self) -> Result<Option<Anchor>> {
if let Some(stall) = self.store.get_stall_reason()? {
return Ok(Some(Anchor {
slot: stall.l1_slot,
block: stall.block_id.zip(stall.block_hash),
}));
/// Detects when a local store belongs to a different chain than the connected
/// L1 (e.g. a wiped/restarted Bedrock) so startup can react instead of silently
/// diverging.
///
/// Verifies the channel still carries the anchor block at its slot. The anchor
/// was finalized at `anchor.slot`, so the same chain must still serve it there,
/// while a reset chain re-inscribes its content only at later wall-clock slots.
/// Only positive evidence of a different chain yields
/// [`ChainConsistency::Inconsistent`]; absence of data stays
/// [`ChainConsistency::Inconclusive`].
///
/// `node` need only implement the zone-sdk [`adapter::Node`] trait; a throwaway
/// [`ZoneIndexer`] is built internally for the channel read.
pub async fn verify_chain_consistency<N>(
node: &N,
channel_id: ChannelId,
anchor: &Anchor,
) -> Result<ChainConsistency>
where
N: adapter::Node + Clone + Sync,
{
match node.channel_state(channel_id).await {
Ok(state) => {
if let Some(mismatch) = frontier_verdict(anchor.slot, state.map(|s| s.tip_slot)) {
return Ok(ChainConsistency::Inconsistent(mismatch));
}
}
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<ChainConsistency> {
match self.node.channel_state(self.config.channel_id).await {
Ok(state) => {
if let Some(mismatch) = frontier_verdict(anchor.slot, state.map(|s| s.tip_slot)) {
return Ok(ChainConsistency::Inconsistent(mismatch));
// `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<Slot>,
messages: &[(ZoneMessage, Slot)],
) -> ChainConsistency {
if let Some(mismatch) = frontier_verdict(anchor.slot, channel_tip_slot) {
return ChainConsistency::Inconsistent(mismatch);
}
for (msg, slot) in messages {
match anchor.probe_anchor_slot(msg, *slot) {
AnchorProbe::SameChain => return ChainConsistency::Consistent,
AnchorProbe::Mismatch(mismatch) => return ChainConsistency::Inconsistent(mismatch),
AnchorProbe::Bail => return ChainConsistency::Inconclusive,
AnchorProbe::KeepLooking => {}
}
}
ChainConsistency::Inconclusive
}
/// Checks the channel frontier against the anchor slot.
///
/// 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<Slot>) -> Option
#[cfg(test)]
mod tests {
use std::time::Duration;
use common::block::HashableBlockData;
use logos_blockchain_core::mantle::ops::channel::{MsgId, inscribe::Inscription};
use logos_blockchain_zone_sdk::ZoneBlock;
use super::*;
use crate::{
BlockIngestError,
block_store::AcceptOutcome,
config::{ChannelId, ClientConfig, IndexerConfig},
};
fn unreachable_core(dir: &std::path::Path) -> IndexerCore {
let config = IndexerConfig {
consensus_info_polling_interval: Duration::from_secs(1),
bedrock_config: ClientConfig {
addr: "http://localhost:1".parse().expect("url"),
auth: None,
},
channel_id: ChannelId::from([1; 32]),
allow_chain_reset: false,
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"),

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

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

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

@ -13,6 +13,7 @@ testnet = []
[dependencies]
common.workspace = true
chain_consistency.workspace = true
logos-blockchain-zone-sdk.workspace = true
lee.workspace = true
lee_core.workspace = true
@ -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

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

@ -1,12 +1,12 @@
use std::{path::Path, sync::Arc};
use anyhow::{Context as _, Result};
use chain_consistency::{BlockIngestError, Tip};
use common::{
HashType,
block::{BedrockStatus, Block, BlockHeader},
transaction::{LeeTransaction, clock_invocation},
transaction::LeeTransaction,
};
use lee::{Account, AccountId, GENESIS_BLOCK_ID, V03State};
use lee::{Account, AccountId, V03State};
use lee_core::BlockId;
use log::warn;
use logos_blockchain_core::header::HeaderId;
@ -14,12 +14,7 @@ use logos_blockchain_zone_sdk::Slot;
use storage::indexer::RocksDBIO;
use tokio::sync::RwLock;
use crate::{ingest_error::BlockIngestError, stall_reason::StallReason};
struct Tip {
block_id: u64,
hash: HashType,
}
use crate::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")

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

@ -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
}
));
}
}

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

@ -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<ChainConsistency> {
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<Option<Anchor>> {
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);
}
}

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

@ -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<u64>,
pub block_hash: Option<HashType>,
pub prev_block_hash: Option<HashType>,
pub l1_slot: Slot,
pub error: BlockIngestError,
pub first_seen: Option<u64>,
/// Number of later non-chaining blocks (orphans, since the tip is frozen).
///
/// TODO: We could store a different "branch" of blocks following this break, but for now we
/// just count them.
pub orphans_since: u64,
}

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

@ -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<u64>,
pub block_hash: Option<HashType>,
pub prev_block_hash: Option<HashType>,
pub l1_slot: Slot,
pub error: BlockIngestError,
pub first_seen: Option<u64>,
/// Number of later non-chaining blocks (orphans, since the tip is frozen).
///
/// TODO: We could store a different "branch" of blocks following this break, but for now we
/// just count them.
pub orphans_since: u64,
}
/// 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),

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

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

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

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

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

@ -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<Self> {
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<Self> {
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<Self> {
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<RocksDBIO>,
signing_key: lee::PrivateKey,
) -> DbResult<Self> {
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<Self> {
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<Option<SequencerCheckpoint>> {
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<Option<ZoneAnchorRecord>> {
self.dbio.get_zone_cursor()
}
pub fn set_zone_anchor(&self, anchor: &ZoneAnchorRecord) -> DbResult<()> {
self.dbio.put_zone_cursor(anchor)
}
pub fn get_unfulfilled_deposit_events(&self) -> DbResult<Vec<PendingDepositEventRecord>> {
self.dbio.get_pending_deposit_events()
}

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

@ -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<BP: BlockPublisherTrait = ZoneSdkPublisher> {
block_publisher: BP,
}
/// Outcome of the startup verify-and-reconstruct pass.
struct ReconstructionSummary {
/// Whether the channel served any blocks. Drives the fresh-start decision
/// between publishing our own genesis and adopting the channel's history.
channel_had_blocks: bool,
}
impl<BP: BlockPublisherTrait> SequencerCore<BP> {
/// Starts the sequencer using the provided configuration.
/// If an existing database is found, the sequencer state is loaded from it and
/// 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<BP: BlockPublisherTrait> SequencerCore<BP> {
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<BP: BlockPublisherTrait> SequencerCore<BP> {
)
.expect("Failed to create database with genesis block");
(store, genesis_state, genesis_block)
(store, genesis_state)
}
}
@ -126,11 +133,7 @@ impl<BP: BlockPublisherTrait> SequencerCore<BP> {
load_or_create_signing_key(&config.home.join("bedrock_signing_key"))
.expect("Failed to load or create bedrock signing key");
let (store, state, _genesis_block) = Self::open_or_create_store(&config);
let latest_block_meta = store
.latest_block_meta()
.expect("Failed to read latest block meta from store");
let (mut store, mut state) = Self::open_or_create_store(&config);
let initial_checkpoint = store
.get_zone_checkpoint()
@ -153,8 +156,28 @@ impl<BP: BlockPublisherTrait> SequencerCore<BP> {
.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<BP: BlockPublisherTrait> SequencerCore<BP> {
}
}
// 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<BP: BlockPublisherTrait> SequencerCore<BP> {
(sequencer_core, mempool_handle)
}
/// Verifies the local store still belongs to the chain the connected channel
/// serves and replays any finalized channel blocks missing locally into
/// `state`/`store`, recording each block's L1 inscription slot as the new
/// anchor. Fails (never parks) on any divergence.
async fn verify_and_reconstruct(
publisher: &BP,
store: &mut SequencerStore,
state: &mut V03State,
is_fresh_start: bool,
) -> Result<ReconstructionSummary> {
let anchor_record = store
.get_zone_anchor()
.context("Failed to read zone anchor")?;
let after_slot = anchor_record
.and_then(|record| record.slot.checked_sub(1))
.map(Slot::from);
let messages = publisher
.read_channel_after(after_slot)
.await
.context("Failed to read channel history for reconstruction")?;
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<RocksDBIO>) -> block_publisher::CheckpointSink {
Box::new(move |cp| {
let bytes = match serde_json::to_vec(&cp) {

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

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

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

@ -37,6 +37,8 @@ use crate::{
mock::SequencerCoreWithMockClients,
};
mod reconstruction;
#[derive(borsh::BorshSerialize)]
struct DepositMetadataForEncoding {
recipient_id: lee::AccountId,

401
lez/sequencer/core/src/tests/reconstruction.rs Normal file
View File

@ -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::<MockBlockPublisher>::open_or_create_store(&config_b);
let mock_b = MockBlockPublisher::with_canned_channel(channel_id, Some(tip_slot), messages);
let summary = SequencerCore::<MockBlockPublisher>::verify_and_reconstruct(
&mock_b,
&mut store_b,
&mut state_b,
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::<MockBlockPublisher>::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::<MockBlockPublisher>::open_or_create_store(&config);
// Anchor on the local genesis at some slot.
let genesis_id = store.genesis_id();
let genesis = store.get_block_at_id(genesis_id).unwrap().unwrap();
let anchor_slot = 100_u64;
store
.set_zone_anchor(&ZoneAnchorRecord {
slot: anchor_slot,
block_id: genesis_id,
hash: genesis.header.hash,
})
.unwrap();
// The channel serves a different block at the anchor id/slot.
let mut tampered = genesis.clone();
tampered.header.hash = HashType([9_u8; 32]);
let messages = vec![block_to_channel_message(&tampered, anchor_slot)];
let mock = MockBlockPublisher::with_canned_channel(
config.bedrock_config.channel_id,
Some(Slot::from(anchor_slot)),
messages,
);
let result = SequencerCore::<MockBlockPublisher>::verify_and_reconstruct(
&mock, &mut store, &mut state, 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::<MockBlockPublisher>::open_or_create_store(&config);
let genesis_id = store.genesis_id();
let genesis = store.get_block_at_id(genesis_id).unwrap().unwrap();
store
.set_zone_anchor(&ZoneAnchorRecord {
slot: 100,
block_id: genesis_id,
hash: genesis.header.hash,
})
.unwrap();
// Anchor present, but the channel does not exist on the connected chain.
let mock =
MockBlockPublisher::with_canned_channel(config.bedrock_config.channel_id, None, vec![]);
let result = SequencerCore::<MockBlockPublisher>::verify_and_reconstruct(
&mock, &mut store, &mut state, 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::<MockBlockPublisher>::open_or_create_store(&config_b);
let mock_b = MockBlockPublisher::with_canned_channel(channel_id, Some(tip_slot), messages);
SequencerCore::<MockBlockPublisher>::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::<MockBlockPublisher>::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::<MockBlockPublisher>::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::<MockBlockPublisher>::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::<MockBlockPublisher>::verify_and_reconstruct(
&mock, &mut store, &mut state, false,
)
.await;
assert!(
result.is_err(),
"committed blocks against a missing channel must abort startup"
);
}

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

@ -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::<ZoneSdkCheckpointCellOwned>(())
}
pub fn get_zone_cursor(&self) -> DbResult<Option<ZoneAnchorRecord>> {
Ok(self.get_opt::<ZoneCursorCell>(())?.map(|cell| cell.0))
}
pub fn put_zone_cursor(&self, anchor: &ZoneAnchorRecord) -> DbResult<()> {
self.put(&ZoneCursorCell(*anchor), ())
}
pub fn get_pending_deposit_events(&self) -> DbResult<Vec<PendingDepositEventRecord>> {
Ok(self
.get_opt::<PendingDepositEventsCellOwned>(())?

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

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

19
test_fixtures/src/bin/regenerate_test_fixture.rs
View File

@ -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(),

19
test_fixtures/src/config.rs
View File

@ -80,7 +80,26 @@ pub fn sequencer_config(
home: PathBuf,
bedrock_addr: SocketAddr,
genesis_transactions: Vec<GenesisAction>,
cross_zone: Option<CrossZoneConfig>,
) -> Result<SequencerConfig> {
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<GenesisAction>,
cross_zone: Option<CrossZoneConfig>,
) -> Result<SequencerConfig> {
let SequencerPartialConfig {

25
test_fixtures/src/lib.rs
View File

@ -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(),

186
test_fixtures/src/setup.rs
View File

@ -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<GenesisAction>),
/// 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<Vec<GenesisAction>>,
cross_zone: Option<sequencer_core::config::CrossZoneConfig>,
}
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<GenesisAction>) -> 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<SequencerHandle> {
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 `<home>/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<GenesisAction>,
channel_id: ChannelId,
cross_zone: Option<sequencer_core::config::CrossZoneConfig>,
) -> 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<sequencer_core::config::CrossZoneConfig>,
) -> 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 `<home>/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)],

24
tools/cross_zone_chat/src/main.rs
View File

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