feat(sequencer): bootstrap state from Bedrock

Cargo.lock

@@ -1368,6 +1368,25 @@ dependencies = [
  "rand_core 0.10.1",
 ]
 
+[[package]]
+name = "chain_consistency"
+version = "0.1.0"
+dependencies = [
+ "anyhow",
+ "borsh",
+ "common",
+ "futures",
+ "lee",
+ "lee_core",
+ "log",
+ "logos-blockchain-core",
+ "logos-blockchain-zone-sdk",
+ "serde",
+ "serde_json",
+ "thiserror 2.0.18",
+ "tokio",
+]
+
 [[package]]
 name = "chkstk_stub"
 version = "0.1.0"
@@ -3939,6 +3958,7 @@ dependencies = [
  "arc-swap",
  "async-stream",
  "borsh",
+ "chain_consistency",
  "common",
  "cross_zone",
  "cross_zone_inbox_core",
@@ -3958,7 +3978,6 @@ dependencies = [
  "storage",
  "tempfile",
  "testnet_initial_state",
- "thiserror 2.0.18",
  "tokio",
  "url",
 ]
@@ -9023,6 +9042,7 @@ dependencies = [
  "borsh",
  "bridge_core",
  "bytesize",
+ "chain_consistency",
  "chrono",
  "common",
  "cross_zone",

Cargo.toml

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

integration_tests/tests/cross_zone_bridge.rs

@@ -17,7 +17,7 @@ use cross_zone_outbox_core::outbox_pda;
 use integration_tests::{
     config::{self, SequencerPartialConfig},
     indexer_client::IndexerClient,
-    setup::{setup_bedrock_node, setup_indexer, setup_sequencer},
+    setup::{SequencerSetup, setup_bedrock_node, setup_indexer},
 };
 use lee::{
     AccountId, PrivateKey, PublicKey, PublicTransaction,
@@ -62,18 +62,19 @@ async fn lock_on_zone_a_mints_wrapped_token_on_zone_b() -> Result<()> {
         holder: holder_id,
         amount: INITIAL_BALANCE,
     }];
-    let (seq_a, _seq_a_home) = setup_sequencer(partial, bedrock_addr, genesis_a, channel_a, None)
+    let (seq_a, _seq_a_home) = SequencerSetup::new(partial, bedrock_addr)
+        .with_channel_id(channel_a)
+        .with_genesis(genesis_a)
+        .setup()
         .await
         .context("Failed to set up zone A sequencer")?;
-    let (_seq_b, _seq_b_home) = setup_sequencer(
+    let (_seq_b, _seq_b_home) = SequencerSetup::new(partial, bedrock_addr)
-        partial,
+        .with_channel_id(channel_b)
-        bedrock_addr,
+        .with_genesis(vec![])
-        vec![],
+        .with_cross_zone(cross_zone.clone())
-        channel_b,
+        .setup()
-        Some(cross_zone.clone()),
+        .await
-    )
+        .context("Failed to set up zone B sequencer")?;
-    .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")?;

integration_tests/tests/cross_zone_ingress_guard.rs

@@ -18,7 +18,7 @@ use cross_zone_inbox_core::{
 };
 use integration_tests::{
     config::{self, SequencerPartialConfig},
-    setup::{setup_bedrock_node, setup_sequencer},
+    setup::{SequencerSetup, setup_bedrock_node},
 };
 use lee::{
     PublicTransaction,
@@ -34,7 +34,10 @@ async fn user_origin_inbox_call_rejected() -> Result<()> {
         .context("Failed to set up Bedrock node")?;
     let partial = SequencerPartialConfig::default();
     let channel = config::bedrock_channel_id();
-    let (seq, _seq_home) = setup_sequencer(partial, bedrock_addr, vec![], channel, None)
+    let (seq, _seq_home) = SequencerSetup::new(partial, bedrock_addr)
+        .with_channel_id(channel)
+        .with_genesis(vec![])
+        .setup()
         .await
         .context("Failed to set up sequencer")?;
 

integration_tests/tests/cross_zone_ping.rs

@@ -18,7 +18,7 @@ use common::transaction::LeeTransaction;
 use cross_zone_outbox_core::outbox_pda;
 use integration_tests::{
     config::{self, SequencerPartialConfig},
-    setup::{setup_bedrock_node, setup_sequencer},
+    setup::{SequencerSetup, setup_bedrock_node},
 };
 use lee::{AccountId, PublicTransaction, public_transaction::Message};
 use lee_core::program::ProgramId;
@@ -54,13 +54,19 @@ async fn ping_crosses_from_zone_a_to_zone_b() -> Result<()> {
         }],
     };
 
-    let (seq_a, _seq_a_home) = setup_sequencer(partial, bedrock_addr, vec![], channel_a, None)
+    let (seq_a, _seq_a_home) = SequencerSetup::new(partial, bedrock_addr)
+        .with_channel_id(channel_a)
+        .with_genesis(vec![])
+        .setup()
         .await
         .context("Failed to set up zone A sequencer")?;
-    let (seq_b, _seq_b_home) =
+    let (seq_b, _seq_b_home) = SequencerSetup::new(partial, bedrock_addr)
-        setup_sequencer(partial, bedrock_addr, vec![], channel_b, Some(cross_zone))
+        .with_channel_id(channel_b)
-            .await
+        .with_genesis(vec![])
-            .context("Failed to set up zone B sequencer")?;
+        .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);

integration_tests/tests/cross_zone_verified.rs

@@ -17,7 +17,7 @@ use cross_zone_outbox_core::outbox_pda;
 use integration_tests::{
     config::{self, SequencerPartialConfig},
     indexer_client::IndexerClient,
-    setup::{setup_bedrock_node, setup_indexer, setup_sequencer},
+    setup::{SequencerSetup, setup_bedrock_node, setup_indexer},
 };
 use lee::{AccountId, PublicTransaction, public_transaction::Message};
 use lee_core::program::ProgramId;
@@ -53,21 +53,22 @@ async fn indexer_verifies_and_delivers_cross_zone_ping() -> Result<()> {
 
     // Zone A: source. Zone B: destination, with the watcher on its sequencer and
     // the verifier on its indexer.
-    let (seq_a, _seq_a_home) = setup_sequencer(partial, bedrock_addr, vec![], channel_a, None)
+    let (seq_a, _seq_a_home) = SequencerSetup::new(partial, bedrock_addr)
+        .with_channel_id(channel_a)
+        .with_genesis(vec![])
+        .setup()
         .await
         .context("Failed to set up zone A sequencer")?;
     let (_idx_a, _idx_a_home) = setup_indexer(bedrock_addr, channel_a, None)
         .await
         .context("Failed to set up zone A indexer")?;
-    let (_seq_b, _seq_b_home) = setup_sequencer(
+    let (_seq_b, _seq_b_home) = SequencerSetup::new(partial, bedrock_addr)
-        partial,
+        .with_channel_id(channel_b)
-        bedrock_addr,
+        .with_genesis(vec![])
-        vec![],
+        .with_cross_zone(cross_zone.clone())
-        channel_b,
+        .setup()
-        Some(cross_zone.clone()),
+        .await
-    )
+        .context("Failed to set up zone B sequencer")?;
-    .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")?;

integration_tests/tests/sequencer_bootstrap.rs

@@ -0,0 +1,545 @@
+//! End-to-end tests for the sequencer's startup bootstrap/reconstruction from
+//! Bedrock (verify-and-reconstruct). Each test drives a real Bedrock node and one
+//! or more sequencer instances sharing the same channel, exercising how a
+//! sequencer reconciles its local store against what the channel serves.
+
+#![expect(
+    clippy::tests_outside_test_module,
+    reason = "Integration tests live at crate root and don't care about these lints"
+)]
+
+use std::{net::SocketAddr, path::Path, time::Duration};
+
+use anyhow::{Context as _, Result, bail};
+use indexer_service_rpc::RpcClient as _;
+use integration_tests::L2_TO_L1_TIMEOUT;
+use lee::{AccountId, PrivateKey, PublicKey};
+use logos_blockchain_core::mantle::ops::channel::ChannelId;
+use sequencer_core::config::GenesisAction;
+use sequencer_service_rpc::{RpcClient as _, SequencerClient, SequencerClientBuilder};
+use test_fixtures::{
+    config::{SequencerPartialConfig, UrlProtocol, addr_to_url},
+    indexer_client::IndexerClient,
+    setup::{SequencerSetup, setup_bedrock_node, setup_indexer},
+};
+use tokio::test;
+
+/// Block cadence for the tests: short so we don't wait long for local production.
+fn fast_blocks() -> SequencerPartialConfig {
+    SequencerPartialConfig {
+        block_create_timeout: Duration::from_secs(2),
+        ..SequencerPartialConfig::default()
+    }
+}
+
+/// Block cadence for a sequencer we don't want producing during the inspection
+/// window, so its post-reconstruction tip stays put while we read it.
+fn slow_blocks() -> SequencerPartialConfig {
+    SequencerPartialConfig {
+        block_create_timeout: Duration::from_secs(30),
+        ..SequencerPartialConfig::default()
+    }
+}
+
+fn sequencer_client(addr: SocketAddr) -> Result<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(())
+}

integration_tests/tests/two_zone.rs

@@ -13,7 +13,7 @@ use indexer_service_rpc::RpcClient as _;
 use integration_tests::{
     config::{self, SequencerPartialConfig},
     indexer_client::IndexerClient,
-    setup::{setup_bedrock_node, setup_indexer, setup_sequencer},
+    setup::{SequencerSetup, setup_bedrock_node, setup_indexer},
 };
 use sequencer_service_rpc::{RpcClient as _, SequencerClientBuilder};
 use tokio::test;
@@ -35,13 +35,19 @@ async fn two_zones_share_one_bedrock_and_both_advance() -> Result<()> {
     let channel_b = config::bedrock_channel_id_b();
 
     // Empty genesis is enough: the clock transaction drives block production.
-    let (seq_a, _seq_a_home) = setup_sequencer(partial, bedrock_addr, vec![], channel_a, None)
+    let (seq_a, _seq_a_home) = SequencerSetup::new(partial, bedrock_addr)
+        .with_channel_id(channel_a)
+        .with_genesis(vec![])
+        .setup()
         .await
         .context("Failed to set up zone A sequencer")?;
     let (idx_a, _idx_a_home) = setup_indexer(bedrock_addr, channel_a, None)
         .await
         .context("Failed to set up zone A indexer")?;
-    let (seq_b, _seq_b_home) = setup_sequencer(partial, bedrock_addr, vec![], channel_b, None)
+    let (seq_b, _seq_b_home) = SequencerSetup::new(partial, bedrock_addr)
+        .with_channel_id(channel_b)
+        .with_genesis(vec![])
+        .setup()
         .await
         .context("Failed to set up zone B sequencer")?;
     let (idx_b, _idx_b_home) = setup_indexer(bedrock_addr, channel_b, None)

lez/chain_consistency/Cargo.toml

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

lez/chain_consistency/src/apply.rs

@@ -0,0 +1,195 @@
+//! Validating and applying channel L2 blocks to a `V03State`.
+//!
+//! These primitives are shared by the consumers that reconstruct L2 state from
+//! the Bedrock channel.
+
+use common::{
+    HashType,
+    block::Block,
+    transaction::{LeeTransaction, clock_invocation},
+};
+use lee::{GENESIS_BLOCK_ID, V03State};
+use lee_core::BlockId;
+use serde::{Deserialize, Serialize};
+
+/// Why a channel L2 block could not be validated or applied.
+///
+/// Persisted in `RocksDB` (via the Indexer's stall reason), so every variant
+/// must be `Clone + Serialize + Deserialize`.
+#[derive(Debug, Clone, Serialize, Deserialize, thiserror::Error)]
+pub enum BlockIngestError {
+    #[error("Failed to deserialize L2 block: {0}")]
+    /// Here we store the error string that is derived from [`borsh::from_slice`]'s [`Err`].
+    Deserialize(String),
+    #[error("Unexpected block id: expected {expected}, got {got}")]
+    UnexpectedBlockId { expected: BlockId, got: BlockId },
+    #[error("Broken chain link: expected prev {expected_prev}, got {got_prev}")]
+    BrokenChainLink {
+        expected_prev: HashType,
+        got_prev: HashType,
+    },
+    #[error("Block hash mismatch: computed {computed}, header {header}")]
+    HashMismatch {
+        computed: HashType,
+        header: HashType,
+    },
+    #[error("Block has no transactions")]
+    EmptyBlock,
+    #[error("Last transaction must be the public clock invocation for the block timestamp")]
+    InvalidClockTransaction,
+    #[error("Genesis block must contain only public transactions")]
+    NonPublicGenesisTransaction,
+    #[error("State transition failed at transaction {tx_index}: {reason}")]
+    StateTransition {
+        /// Index of the failing transaction within the block body.
+        tx_index: u64,
+        /// Reason string from `lee::Error` to `anyhow::Error` to `{:#}`.
+        ///
+        /// This is required because `lee::Error` is not `Clone + Serialize + Deserialize`, so we
+        /// cannot store it directly.
+        reason: String,
+    },
+}
+
+impl BlockIngestError {
+    /// Whether the failure may be transient rather than a property of the block.
+    ///
+    /// FIXME: `StateTransition` is too coarse — its `reason` string mixes genuine
+    /// state-transition rejections with infra failures (risc0 executor teardown,
+    /// storage errors). Once it carries a structured cause, narrow this so only
+    /// infra failures retry.
+    #[must_use]
+    pub const fn is_retryable(&self) -> bool {
+        matches!(self, Self::StateTransition { .. })
+    }
+}
+
+/// The last successfully applied block a candidate must extend.
+pub struct Tip {
+    pub block_id: BlockId,
+    pub hash: HashType,
+}
+
+/// Checks that `block` is the valid continuation of `tip`: hash integrity,
+/// then block-id continuity, then `prev_block_hash` linkage. A `None` tip
+/// (cold store) expects the genesis block.
+pub fn validate_against_tip(tip: Option<&Tip>, block: &Block) -> Result<(), BlockIngestError> {
+    let computed = block.recompute_hash();
+    if computed != block.header.hash {
+        return Err(BlockIngestError::HashMismatch {
+            computed,
+            header: block.header.hash,
+        });
+    }
+
+    match tip {
+        None => {
+            if block.header.block_id != GENESIS_BLOCK_ID {
+                return Err(BlockIngestError::UnexpectedBlockId {
+                    expected: GENESIS_BLOCK_ID,
+                    got: block.header.block_id,
+                });
+            }
+        }
+        Some(tip) => {
+            let expected = tip
+                .block_id
+                .checked_add(1)
+                .expect("block id should not overflow");
+            if block.header.block_id != expected {
+                return Err(BlockIngestError::UnexpectedBlockId {
+                    expected,
+                    got: block.header.block_id,
+                });
+            }
+            if block.header.prev_block_hash != tip.hash {
+                return Err(BlockIngestError::BrokenChainLink {
+                    expected_prev: tip.hash,
+                    got_prev: block.header.prev_block_hash,
+                });
+            }
+        }
+    }
+    Ok(())
+}
+
+/// Applies a block's transactions to `state`.
+pub fn apply_block(block: &Block, state: &mut V03State) -> Result<(), BlockIngestError> {
+    let (clock_tx, user_txs) = block
+        .body
+        .transactions
+        .split_last()
+        .ok_or(BlockIngestError::EmptyBlock)?;
+
+    let expected_clock = LeeTransaction::Public(clock_invocation(block.header.timestamp));
+    if *clock_tx != expected_clock {
+        return Err(BlockIngestError::InvalidClockTransaction);
+    }
+
+    let is_genesis = block.header.block_id == GENESIS_BLOCK_ID;
+    for (tx_index, transaction) in user_txs.iter().enumerate() {
+        let state_transition = |err: anyhow::Error| BlockIngestError::StateTransition {
+            tx_index: tx_index.try_into().expect("tx index fits in u64"),
+            reason: format!("{err:#}"),
+        };
+        if is_genesis {
+            let LeeTransaction::Public(public_tx) = transaction else {
+                return Err(BlockIngestError::NonPublicGenesisTransaction);
+            };
+            state
+                .transition_from_public_transaction(
+                    public_tx,
+                    block.header.block_id,
+                    block.header.timestamp,
+                )
+                .map_err(|err| state_transition(err.into()))?;
+        } else {
+            transaction
+                .clone()
+                .execute_on_state(state, block.header.block_id, block.header.timestamp)
+                .map_err(|err| state_transition(err.into()))?;
+        }
+    }
+
+    let LeeTransaction::Public(clock_public_tx) = clock_tx else {
+        return Err(BlockIngestError::InvalidClockTransaction);
+    };
+    state
+        .transition_from_public_transaction(
+            clock_public_tx,
+            block.header.block_id,
+            block.header.timestamp,
+        )
+        .map_err(|err| BlockIngestError::StateTransition {
+            tx_index: user_txs.len().try_into().expect("tx index fits in u64"),
+            reason: format!("{:#}", anyhow::Error::from(err)),
+        })?;
+
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn serializes_and_round_trips_externally_tagged() {
+        let err = BlockIngestError::UnexpectedBlockId {
+            expected: 5,
+            got: 7,
+        };
+        let value = serde_json::to_value(&err).expect("serialize");
+        assert_eq!(
+            value,
+            serde_json::json!({ "UnexpectedBlockId": { "expected": 5, "got": 7 } })
+        );
+        let back: BlockIngestError = serde_json::from_value(value).expect("deserialize");
+        assert!(matches!(
+            back,
+            BlockIngestError::UnexpectedBlockId {
+                expected: 5,
+                got: 7
+            }
+        ));
+    }
+}

lez/chain_consistency/src/consistency.rs

@@ -1,18 +1,18 @@
-//! Startup check that the local store still belongs to the chain the
+//! Startup check that a local store still belongs to the chain the connected
-//! connected channel serves.
+//! 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 logos_blockchain_core::mantle::ops::channel::ChannelId;
-
+use logos_blockchain_zone_sdk::{Slot, ZoneMessage, adapter, indexer::ZoneIndexer};
-use crate::IndexerCore;
 
 /// Upper bound on the channel reads of the startup consistency check.
 const CHANNEL_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(60);
 
-/// Result of comparing the indexer's stored chain against the channel.
+/// 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),
+        ours: (BlockId, HashType),
-        channel: (u64, 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, Clone, PartialEq, Eq)]
+pub struct Anchor {
     slot: Slot,
     /// The anchor block's `(id, hash)`.
     ///
-    /// `None` when parked on an undeserializable inscription (no header was recorded).
+    /// `None` when anchored on an undeserializable inscription (no header was recorded).
-    block: Option<(u64, HashType)>,
+    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`].
+    /// See [`verify_chain_consistency`].
-    pub fn probe_anchor_slot(&self, msg: &ZoneMessage, slot: Slot) -> AnchorProbe {
+    fn probe_anchor_slot(&self, msg: &ZoneMessage, slot: Slot) -> AnchorProbe {
         if slot < self.slot {
             return AnchorProbe::KeepLooking;
         }
@@ -157,6 +165,66 @@ impl Anchor {
     }
 }
 
+/// Classifies a stored chain against the channel one message at a time.
+///
+/// The shared driver behind both consistency consumers:
+/// [`verify_chain_consistency`] runs it over a channel it reads itself, while a
+/// caller that already streams the channel history (e.g. a reconstructing
+/// sequencer) feeds it the same messages it replays. Run [`Self::check_frontier`]
+/// once the channel tip is known, then feed messages in slot order via
+/// [`Self::observe`] until a verdict is reached.
+pub struct AnchorConsistencyCheck {
+    anchor: Anchor,
+    verdict: Option<ChainConsistency>,
+}
+
+impl AnchorConsistencyCheck {
+    /// New checker for `anchor` with an undetermined verdict.
+    #[must_use]
+    pub const fn new(anchor: Anchor) -> Self {
+        Self {
+            anchor,
+            verdict: None,
+        }
+    }
+
+    /// Applies the frontier check against a known channel tip. Skip it when the
+    /// tip could not be read, leaving the verdict to the message scan.
+    pub fn check_frontier(&mut self, channel_tip_slot: Option<Slot>) {
+        if self.verdict.is_none() {
+            self.verdict = frontier_verdict(self.anchor.slot, channel_tip_slot)
+                .map(ChainConsistency::Inconsistent);
+        }
+    }
+
+    /// Feeds the next channel message in slot order. Returns the verdict once it
+    /// is known so the caller can stop, or `None` while still undetermined.
+    pub fn observe(&mut self, msg: &ZoneMessage, slot: Slot) -> Option<&ChainConsistency> {
+        if self.verdict.is_none() {
+            self.verdict = match self.anchor.probe_anchor_slot(msg, slot) {
+                AnchorProbe::SameChain => Some(ChainConsistency::Consistent),
+                AnchorProbe::Mismatch(mismatch) => Some(ChainConsistency::Inconsistent(mismatch)),
+                AnchorProbe::Bail => Some(ChainConsistency::Inconclusive),
+                AnchorProbe::KeepLooking => None,
+            };
+        }
+        self.verdict.as_ref()
+    }
+
+    /// The verdict reached so far, or `None` while undetermined.
+    #[must_use]
+    pub const fn verdict(&self) -> Option<&ChainConsistency> {
+        self.verdict.as_ref()
+    }
+
+    /// Consumes the checker, defaulting an undetermined verdict (the anchor slot
+    /// was never observed) to [`ChainConsistency::Inconclusive`].
+    #[must_use]
+    pub fn finish(self) -> ChainConsistency {
+        self.verdict.unwrap_or(ChainConsistency::Inconclusive)
+    }
+}
+
 /// What a single channel message tells the anchored consistency check.
 enum AnchorProbe {
     /// The anchor is still in place: same chain.
@@ -168,114 +236,65 @@ enum AnchorProbe {
     KeepLooking,
 }
 
-#[expect(
+/// Detects when a local store belongs to a different chain than the connected
-    clippy::multiple_inherent_impl,
+/// L1 (e.g. a wiped/restarted Bedrock) so startup can react instead of silently
-    reason = "split for clarity & isolation of relevant code"
+/// diverging.
-)]
+///
-impl IndexerCore {
+/// Verifies the channel still carries the anchor block at its slot. The anchor
-    /// Verifies whether the channel still serves the same chain the store was built from.
+/// was finalized at `anchor.slot`, so the same chain must still serve it there,
-    /// This may change frequently during development where we reset the chain from time to
+/// while a reset chain re-inscribes its content only at later wall-clock slots.
-    /// time in devnet/testnet, but we do not expect [`ChainConsistency::Inconsistent`] in
+/// Only positive evidence of a different chain yields
-    /// production.
+/// [`ChainConsistency::Inconsistent`]; absence of data stays
-    ///
+/// [`ChainConsistency::Inconclusive`].
-    /// 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.
+/// `node` need only implement the zone-sdk [`adapter::Node`] trait; a throwaway
-    pub(crate) async fn verify_chain_consistency(&self) -> Result<ChainConsistency> {
+/// [`ZoneIndexer`] is built internally for the channel read.
-        let Some(anchor) = self.get_startup_anchor()? else {
+pub async fn verify_chain_consistency<N>(
-            // empty or cold store: nothing to compare
+    node: &N,
-            return Ok(ChainConsistency::Inconclusive);
+    channel_id: ChannelId,
-        };
+    anchor: &Anchor,
-
+) -> Result<ChainConsistency>
-        self.verify_chain_at_anchor(&anchor).await
+where
+    N: adapter::Node + Clone + Sync,
+{
+    let mut check = AnchorConsistencyCheck::new(anchor.clone());
+    match node.channel_state(channel_id).await {
+        Ok(state) => check.check_frontier(state.map(|s| s.tip_slot)),
+        Err(err) => warn!("Failed to read channel state for the consistency check: {err:#}"),
+    }
+    if check.verdict().is_some() {
+        return Ok(check.finish());
     }
 
-    /// Builds the anchor for the startup check.
+    // `next_messages` is exclusive, so `slot - 1` includes the anchor slot.
-    ///
+    let Some(from_slot) = anchor.slot.into_inner().checked_sub(1) else {
-    /// - If stalled, returns the recorded _parked_ block
+        return Ok(ChainConsistency::Inconclusive);
-    /// - If not stalled, returns the validated tip at its _own_ inscription slot.
+    };
-    /// - If the store is empty, returns `None`.
-    fn get_startup_anchor(&self) -> Result<Option<Anchor>> {
-        if let Some(stall) = self.store.get_stall_reason()? {
-            return Ok(Some(Anchor {
-                slot: stall.l1_slot,
-                block: stall.block_id.zip(stall.block_hash),
-            }));
-        }
 
-        // not stalled, so anchor on the tip at its own inscription slot
+    let zone_indexer = ZoneIndexer::new(channel_id, node.clone());
-        let Some(slot) = self.store.get_tip_slot()?.or(self.store.get_zone_cursor()?) else {
+    let scan = async {
-            return Ok(None);
+        let stream = zone_indexer
-        };
+            .next_messages(Some(Slot::from(from_slot)))
-        let Some(tip_id) = self.store.get_last_block_id()? else {
+            .await?;
-            return Ok(None);
+        let mut stream = std::pin::pin!(stream);
-        };
-        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.
+        while let Some((msg, slot)) = stream.next().await {
-    ///
+            if check.observe(&msg, slot).is_some() {
-    /// The anchor was finalized at `anchor.slot`, so the same chain must still
+                break;
-    /// serve it there, while a reset chain re-inscribes its content only at
-    /// later wall-clock slots.
-    ///
-    /// Only positive evidence of a different chain yields `Inconsistent`.
-    /// Absence of data stays `Inconclusive`.
-    async fn verify_chain_at_anchor(&self, anchor: &Anchor) -> Result<ChainConsistency> {
-        match self.node.channel_state(self.config.channel_id).await {
-            Ok(state) => {
-                if let Some(mismatch) = frontier_verdict(anchor.slot, state.map(|s| s.tip_slot)) {
-                    return Ok(ChainConsistency::Inconsistent(mismatch));
-                }
-            }
-            Err(err) => {
-                warn!("Failed to read channel state for the consistency check: {err:#}");
             }
         }
+        Ok::<_, anyhow::Error>(())
+    };
 
-        // `next_messages` is exclusive, so `slot - 1` includes the anchor slot.
+    match tokio::time::timeout(CHANNEL_READ_TIMEOUT, scan).await {
-        let Some(from_slot) = anchor.slot.into_inner().checked_sub(1) else {
+        Ok(Ok(())) => Ok(check.finish()),
-            return Ok(ChainConsistency::Inconclusive);
+        Ok(Err(err)) => {
-        };
+            warn!("Failed to read the anchor slot for the consistency check; proceeding: {err:#}");
-
+            Ok(ChainConsistency::Inconclusive)
-        let scan = async {
+        }
-            let stream = self
+        Err(_elapsed) => {
-                .zone_indexer
+            warn!("Timed out reading the anchor slot for the consistency check; proceeding");
-                .next_messages(Some(Slot::from(from_slot)))
+            Ok(ChainConsistency::Inconclusive)
-                .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)
-            }
         }
     }
 }
@@ -297,35 +316,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 {
+    fn test_block(block_id: BlockId, timestamp: u64) -> Block {
-        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]),
@@ -344,98 +341,7 @@ mod tests {
     }
 
     fn anchor_for(block: &Block, slot: Slot) -> Anchor {
-        Anchor {
+        Anchor::new(slot, Some((block.header.block_id, block.header.hash)))
-            slot,
-            block: Some((block.header.block_id, block.header.hash)),
-        }
-    }
-
-    #[tokio::test]
-    async fn cold_store_is_inconclusive() {
-        // An empty store has no cursor, so there is nothing to compare: the check
-        // must be Inconclusive (not Consistent), and it returns before any L1 read.
-        let dir = tempfile::tempdir().expect("tempdir");
-        let core = unreachable_core(dir.path());
-        assert!(matches!(
-            core.verify_chain_consistency().await.expect("verify"),
-            ChainConsistency::Inconclusive
-        ));
-    }
-
-    #[tokio::test]
-    async fn parked_store_with_unreachable_node_is_inconclusive() {
-        // Network failure is not evidence of a reset: a parked store must stay
-        // parked (Inconclusive), not error out or trip the wipe path.
-        let dir = tempfile::tempdir().expect("tempdir");
-        let core = unreachable_core(dir.path());
-        let parked = test_block(5, 42);
-        core.store
-            .record_stall(
-                Some(&parked.header),
-                Slot::from(1_000),
-                BlockIngestError::EmptyBlock,
-            )
-            .expect("record stall");
-        assert!(matches!(
-            core.verify_chain_consistency().await.expect("verify"),
-            ChainConsistency::Inconclusive
-        ));
-    }
-
-    #[tokio::test]
-    async fn caught_up_store_with_unreachable_node_is_inconclusive() {
-        let dir = tempfile::tempdir().expect("tempdir");
-        let core = unreachable_core(dir.path());
-        let genesis = common::test_utils::produce_dummy_block(1, None, vec![]);
-        assert!(matches!(
-            core.store
-                .accept_block(&genesis, Slot::from(1_000))
-                .await
-                .expect("accept"),
-            AcceptOutcome::Applied
-        ));
-        core.store
-            .set_zone_cursor(&Slot::from(1_000))
-            .expect("set cursor");
-        assert!(matches!(
-            core.verify_chain_consistency().await.expect("verify"),
-            ChainConsistency::Inconclusive
-        ));
-    }
-
-    #[tokio::test]
-    async fn startup_anchor_prefers_tip_slot_over_lagging_cursor() {
-        // Cursor persist failures are warn-only, so the read cursor can lag the
-        // tip by several blocks. The anchor must pair the tip with its own
-        // inscription slot; pairing it with the stale cursor would make the scan
-        // misread the chain's intermediate blocks as re-inscriptions.
-        let dir = tempfile::tempdir().expect("tempdir");
-        let core = unreachable_core(dir.path());
-
-        let genesis = common::test_utils::produce_dummy_block(1, None, vec![]);
-        core.store
-            .accept_block(&genesis, Slot::from(1_000))
-            .await
-            .expect("accept");
-        let block2 = common::test_utils::produce_dummy_block(2, Some(genesis.header.hash), vec![]);
-        core.store
-            .accept_block(&block2, Slot::from(1_005))
-            .await
-            .expect("accept");
-        let block3 = common::test_utils::produce_dummy_block(3, Some(block2.header.hash), vec![]);
-        core.store
-            .accept_block(&block3, Slot::from(1_010))
-            .await
-            .expect("accept");
-
-        // Cursor last persisted at the genesis slot: two blocks behind the tip.
-        core.store
-            .set_zone_cursor(&Slot::from(1_000))
-            .expect("set cursor");
-
-        let anchor = core.get_startup_anchor().expect("anchor").expect("present");
-        assert_eq!(anchor.slot, Slot::from(1_010));
-        assert_eq!(anchor.block, Some((3, block3.header.hash)));
     }
 
     #[test]
@@ -516,10 +422,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 {
+        let anchor = Anchor::new(Slot::from(1_000), None);
-            slot: Slot::from(1_000),
-            block: None,
-        };
         let garbage = ZoneMessage::Block(ZoneBlock {
             id: MsgId::from([0_u8; 32]),
             data: Inscription::try_from(&[1_u8, 2, 3][..]).expect("inscription"),

lez/chain_consistency/src/lib.rs

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

lez/indexer/core/Cargo.toml

@@ -13,6 +13,7 @@ testnet = []
 
 [dependencies]
 common.workspace = true
+chain_consistency.workspace = true
 logos-blockchain-zone-sdk.workspace = true
 lee.workspace = true
 lee_core.workspace = true
@@ -32,7 +33,6 @@ futures.workspace = true
 url.workspace = true
 logos-blockchain-core.workspace = true
 serde_json.workspace = true
-thiserror.workspace = true
 async-stream.workspace = true
 tokio.workspace = true
 risc0-zkvm.workspace = true

lez/indexer/core/src/block_store.rs

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

lez/indexer/core/src/ingest_error.rs

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

lez/indexer/core/src/lib.rs

@@ -2,31 +2,29 @@ use std::{path::Path, sync::Arc};
 
 use anyhow::Result;
 use arc_swap::ArcSwap;
+pub use chain_consistency::BlockIngestError;
+use chain_consistency::{Anchor, ChainConsistency};
 use common::block::Block;
 // TODO: Remove after testnet
 use futures::StreamExt as _;
-pub use ingest_error::BlockIngestError;
 use log::{error, info, warn};
 use logos_blockchain_zone_sdk::{
     CommonHttpClient, Slot, ZoneMessage, adapter::NodeHttpClient, indexer::ZoneIndexer,
 };
 use retry::ApplyRetryGate;
-pub use stall_reason::StallReason;
+pub use status::StallReason;
 
 use crate::{
     block_store::{AcceptOutcome, IndexerStore},
-    chain_consistency::ChainConsistency,
     config::IndexerConfig,
     cross_zone_verifier::CrossZoneVerifier,
     status::{IndexerStatus, IndexerSyncStatus},
 };
+
 pub mod block_store;
-pub mod chain_consistency;
 pub mod config;
 pub mod cross_zone_verifier;
-pub mod ingest_error;
 mod retry;
-pub mod stall_reason;
 pub mod status;
 
 /// Consecutive failed apply attempts of the same block before parking.
@@ -125,6 +123,53 @@ 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()?
+            .map_or_else(|| self.store.get_zone_cursor(), |slot| Ok(Some(slot)))?
+        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 +252,8 @@ impl IndexerCore {
             let mut cursor = initial_cursor;
             let mut retry_gate = ApplyRetryGate::new();
 
-            if cursor.is_some() {
+            if let Some(slot) = &cursor {
-                info!("Resuming indexer from cursor {cursor:?}");
+                info!("Resuming indexer from cursor {slot:?}");
             } else {
                 info!("Starting indexer from beginning of channel");
             }
@@ -370,3 +415,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);
+    }
+}

lez/indexer/core/src/stall_reason.rs

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

lez/indexer/core/src/status.rs

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

lez/sequencer/core/Cargo.toml

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

lez/sequencer/core/src/block_publisher.rs

@@ -2,14 +2,16 @@ use std::{pin::Pin, sync::Arc, time::Duration};
 
 use anyhow::{Context as _, Result, anyhow};
 use common::block::Block;
+use futures::Stream;
 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,
+    CommonHttpClient, Slot, ZoneMessage,
-    adapter::NodeHttpClient,
+    adapter::{Node as _, NodeHttpClient},
+    indexer::ZoneIndexer,
     sequencer::{
         DepositInfo, Event, FinalizedOp, InscriptionInfo,
         SequencerConfig as ZoneSdkSequencerConfig, WithdrawArg, WithdrawInfo, ZoneSequencer,
@@ -42,7 +44,7 @@ pub type OnWithdrawEventSink =
     Box<dyn Fn(WithdrawInfo) -> Pin<Box<dyn Future<Output = ()> + Send>> + Send + 'static>;
 
 #[expect(async_fn_in_trait, reason = "We don't care about Send/Sync here")]
-pub trait BlockPublisherTrait: Clone {
+pub trait BlockPublisherTrait: Sized {
     #[expect(
         clippy::too_many_arguments,
         reason = "Looks better than bundling all those callbacks into a struct"
@@ -63,15 +65,30 @@ 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<impl Stream<Item = (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>,
+    indexer: ZoneIndexer<NodeHttpClient>,
 }
 
 struct DriveTaskGuard(JoinHandle<()>);
@@ -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,8 @@ impl BlockPublisherTrait for ZoneSdkPublisher {
 
         Ok(Self {
             channel_id: config.channel_id,
+            indexer: ZoneIndexer::new(config.channel_id, node.clone()),
+            node,
             publish_tx,
             _drive_task: Arc::new(DriveTaskGuard(drive_task)),
         })
@@ -218,6 +237,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<impl Stream<Item = (ZoneMessage, Slot)> + '_> {
+        let stream = self
+            .indexer
+            .next_messages(after_slot)
+            .await
+            .context("Failed to start channel read stream")?;
+        Ok(stream)
+    }
 }
 
 /// Deserialize inscription payload as a `Block` and return it's`block_id`.

lez/sequencer/core/src/block_store.rs

@@ -12,7 +12,7 @@ use log::info;
 use logos_blockchain_zone_sdk::sequencer::SequencerCheckpoint;
 use storage::sequencer::{
     RocksDBIO,
-    sequencer_cells::{PendingDepositEventRecord, WithdrawalReconciliationKey},
+    sequencer_cells::{PendingDepositEventRecord, WithdrawalReconciliationKey, ZoneAnchorRecord},
 };
 pub use storage::{DbResult, sequencer::DbDump};
 
@@ -28,6 +28,45 @@ impl SequencerStore {
     /// Open existing database at the given location. Fails if no database is found.
     pub fn open_db(location: &Path, signing_key: lee::PrivateKey) -> DbResult<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()
     }

lez/sequencer/core/src/lib.rs

@@ -2,18 +2,23 @@ use std::{path::Path, sync::Arc, time::Instant};
 
 use anyhow::{Context as _, Result, anyhow};
 use borsh::BorshDeserialize;
+use chain_consistency::{Anchor, AnchorConsistencyCheck, ChainConsistency, ChainMismatch, Tip};
 use common::{
     HashType,
     block::{BedrockStatus, Block, HashableBlockData},
     transaction::{LeeTransaction, clock_invocation},
 };
 use config::{GenesisAction, SequencerConfig};
+use futures::StreamExt as _;
 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 +26,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::{
@@ -72,7 +77,7 @@ impl<BP: BlockPublisherTrait> SequencerCore<BP> {
     /// 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 +91,7 @@ impl<BP: BlockPublisherTrait> SequencerCore<BP> {
             let state = store
                 .get_lee_state()
                 .expect("Failed to read state from store");
-            let genesis_block = store
+            (store, state)
-                .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)
         } else {
             warn!(
                 "Database not found at {}, starting from genesis",
@@ -115,7 +116,7 @@ impl<BP: BlockPublisherTrait> SequencerCore<BP> {
             )
             .expect("Failed to create database with genesis block");
 
-            (store, genesis_state, genesis_block)
+            (store, genesis_state)
         }
     }
 
@@ -126,11 +127,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 (mut store, mut state) = Self::open_or_create_store(&config);
-
-        let latest_block_meta = store
-            .latest_block_meta()
-            .expect("Failed to read latest block meta from store");
 
         let initial_checkpoint = store
             .get_zone_checkpoint()
@@ -153,8 +150,28 @@ impl<BP: BlockPublisherTrait> SequencerCore<BP> {
         .await
         .expect("Failed to initialize Block Publisher");
 
-        // Fresh start (no checkpoint): republish all pending blocks
+        // Cross-zone messaging: start a watcher per configured peer. The inbox
-        if is_fresh_start {
+        // 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 chanel_was_empty =
+            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 && chanel_was_empty {
             let mut pending_blocks = store
                 .get_all_blocks()
                 .filter_ok(|block| matches!(block.bedrock_status, BedrockStatus::Pending))
@@ -182,16 +199,9 @@ impl<BP: BlockPublisherTrait> SequencerCore<BP> {
             }
         }
 
-        // Cross-zone messaging: start a watcher per configured peer. The inbox
+        let latest_block_meta = store
-        // config account is seeded into genesis state in `build_genesis_state`.
+            .latest_block_meta()
-        if let Some(cross_zone) = &config.cross_zone {
+            .expect("Failed to read latest block meta from store");
-            cross_zone_watcher::spawn_watchers(
-                &config.bedrock_config,
-                cross_zone,
-                config.block_create_timeout,
-                &mempool_handle,
-            );
-        }
 
         let sequencer_core = Self {
             state,
@@ -205,6 +215,204 @@ 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.
+    ///
+    /// Returns whatever channel was empty or not.
+    async fn verify_and_reconstruct(
+        publisher: &BP,
+        store: &mut SequencerStore,
+        state: &mut V03State,
+        is_fresh_start: bool,
+    ) -> Result<bool> {
+        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 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."
+            ));
+        }
+
+        let divergence_error = |mismatch: &ChainMismatch| {
+            anyhow!(
+                "Sequencer store diverges from the Bedrock channel ({mismatch}). \
+                 Delete the sequencer storage directory or point at the correct channel."
+            )
+        };
+
+        // With a recorded anchor, probe the channel for positive evidence of a
+        // different chain: the frontier upfront (a missing/behind channel serves
+        // no messages to scan), then the anchor block as messages stream in.
+        let mut consistency_check = anchor_record.map(|record| {
+            let anchor = Anchor::new(
+                Slot::from(record.slot),
+                Some((record.block_id, record.hash)),
+            );
+            let mut check = AnchorConsistencyCheck::new(anchor);
+            check.check_frontier(channel_tip_slot);
+            check
+        });
+        if let Some(ChainConsistency::Inconsistent(mismatch)) = consistency_check
+            .as_ref()
+            .and_then(AnchorConsistencyCheck::verdict)
+        {
+            return Err(divergence_error(mismatch));
+        }
+
+        // Verify each message against the anchor and replay the
+        // blocks (applying the ones we miss, checking the ones we hold).
+        let messages = publisher
+            .read_channel_after(after_slot)
+            .await
+            .context("Failed to read channel history for reconstruction")?;
+        let mut messages = std::pin::pin!(messages);
+        let mut channel_is_empty = true;
+        while let Some((message, slot)) = messages.next().await {
+            if let Some(check) = &mut consistency_check
+                && let Some(ChainConsistency::Inconsistent(mismatch)) =
+                    check.observe(&message, slot)
+            {
+                return Err(divergence_error(mismatch));
+            }
+
+            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_is_empty = false;
+            Self::apply_reconstructed_block(store, state, &block, slot)?;
+        }
+
+        Ok(channel_is_empty)
+    }
+
+    /// 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) {
@@ -561,8 +769,8 @@ impl<BP: BlockPublisherTrait> SequencerCore<BP> {
             .collect())
     }
 
-    pub fn block_publisher(&self) -> BP {
+    pub const fn block_publisher(&self) -> &BP {
-        self.block_publisher.clone()
+        &self.block_publisher
     }
 
     fn next_block_id(&self) -> u64 {

lez/sequencer/core/src/mock.rs

@@ -2,9 +2,10 @@ use std::time::Duration;
 
 use anyhow::Result;
 use common::block::Block;
+use futures::Stream;
 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 +20,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 +57,8 @@ impl BlockPublisherTrait for MockBlockPublisher {
     ) -> Result<Self> {
         Ok(Self {
             channel_id: config.channel_id,
+            tip_slot: None,
+            messages: Vec::new(),
         })
     }
 
@@ -48,4 +73,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<impl Stream<Item = (ZoneMessage, Slot)> + '_> {
+        // Mirror `next_messages`: `after_slot` is exclusive.
+        let messages = self
+            .messages
+            .iter()
+            .filter(move |(_, slot)| after_slot.is_none_or(|after| *slot > after))
+            .cloned();
+        Ok(futures::stream::iter(messages))
+    }
 }

lez/sequencer/core/src/tests.rs

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

lez/sequencer/core/src/tests/reconstruction.rs

@@ -0,0 +1,401 @@
+#![expect(
+    clippy::arithmetic_side_effects,
+    clippy::as_conversions,
+    reason = "We don't care about it in tests"
+)]
+
+use common::block::Block;
+use logos_blockchain_core::mantle::ops::channel::{MsgId, inscribe::Inscription};
+use logos_blockchain_zone_sdk::{Slot, ZoneBlock, ZoneMessage};
+use storage::sequencer::sequencer_cells::ZoneAnchorRecord;
+
+use super::*;
+use crate::{
+    SequencerCore, block_store::SequencerStore, config::GenesisAction, mock::MockBlockPublisher,
+};
+
+fn block_to_channel_message(block: &Block, slot: u64) -> (ZoneMessage, Slot) {
+    let bytes = borsh::to_vec(block).expect("serialize block");
+    let message = ZoneMessage::Block(ZoneBlock {
+        id: MsgId::from([0_u8; 32]),
+        data: Inscription::try_from(bytes.as_slice()).expect("inscription"),
+    });
+    (message, Slot::from(slot))
+}
+
+/// Collects a sequencer's whole chain (genesis..=tip) into a canned channel,
+/// one block per slot at `slot_step` spacing.
+fn channel_from_store(store: &SequencerStore, slot_step: u64) -> Vec<(ZoneMessage, Slot)> {
+    let genesis_id = store.genesis_id();
+    let tip_id = store.latest_block_meta().expect("tip").id;
+    (genesis_id..=tip_id)
+        .enumerate()
+        .map(|(index, id)| {
+            let block = store.get_block_at_id(id).expect("read").expect("present");
+            block_to_channel_message(&block, (index as u64 + 1) * slot_step)
+        })
+        .collect()
+}
+
+#[tokio::test]
+async fn reconstructs_missing_channel_blocks_into_fresh_store() {
+    // Sequencer A produces a few blocks; treat its chain as the channel.
+    let config_a = setup_sequencer_config();
+    let (mut seq_a, _handle_a) =
+        SequencerCoreWithMockClients::start_from_config(config_a.clone()).await;
+    seq_a.produce_new_block().await.unwrap();
+    seq_a.produce_new_block().await.unwrap();
+    let tip_a = seq_a.block_store().latest_block_meta().unwrap();
+
+    let messages = channel_from_store(seq_a.block_store(), 10);
+    let tip_slot = messages.last().unwrap().1;
+    let channel_id = config_a.bedrock_config.channel_id;
+
+    // Sequencer B starts from a fresh store and reconstructs A's chain.
+    let config_b = setup_sequencer_config();
+    let (mut store_b, mut state_b) =
+        SequencerCore::<MockBlockPublisher>::open_or_create_store(&config_b);
+    let mock_b = MockBlockPublisher::with_canned_channel(channel_id, Some(tip_slot), messages);
+
+    let channel_was_empty = SequencerCore::<MockBlockPublisher>::verify_and_reconstruct(
+        &mock_b,
+        &mut store_b,
+        &mut state_b,
+        true,
+    )
+    .await
+    .expect("reconstruct");
+    assert!(!channel_was_empty);
+
+    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 channel_was_empty = SequencerCore::<MockBlockPublisher>::verify_and_reconstruct(
+        &mock_b,
+        &mut store_b,
+        &mut state_b,
+        true,
+    )
+    .await
+    .expect("reconstruct idempotent");
+    assert!(!channel_was_empty);
+    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"
+    );
+}

lez/storage/src/sequencer/mod.rs

@@ -22,7 +22,7 @@ use crate::{
         LEEStateCellOwned, LEEStateCellRef, LastFinalizedBlockIdCell, LatestBlockMetaCellOwned,
         LatestBlockMetaCellRef, PendingDepositEventRecord, PendingDepositEventsCellOwned,
         PendingDepositEventsCellRef, UnseenWithdrawCountCell, WithdrawalReconciliationKey,
-        ZoneSdkCheckpointCellOwned, ZoneSdkCheckpointCellRef,
+        ZoneAnchorRecord, ZoneCursorCell, ZoneSdkCheckpointCellOwned, ZoneSdkCheckpointCellRef,
     },
 };
 
@@ -34,6 +34,10 @@ pub const DB_META_LAST_FINALIZED_BLOCK_ID: &str = "last_finalized_block_id";
 pub const DB_META_LATEST_BLOCK_META_KEY: &str = "latest_block_meta";
 /// Key base for storing the zone-sdk sequencer checkpoint (opaque bytes).
 pub const DB_META_ZONE_SDK_CHECKPOINT_KEY: &str = "zone_sdk_checkpoint";
+/// Key base for storing the last channel block read back and verified from
+/// Bedrock (its L1 slot + `id`/`hash`) — the anchor for the startup
+/// consistency check and the resume point for reconstruction.
+pub const DB_META_ZONE_CURSOR_KEY: &str = "zone_cursor";
 /// Key base for storing queued deposit events that were not yet
 /// fulfilled on L2.
 pub const DB_META_PENDING_DEPOSIT_EVENTS_KEY: &str = "pending_deposit_events";
@@ -359,6 +363,14 @@ impl RocksDBIO {
         self.del::<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>(())?

lez/storage/src/sequencer/sequencer_cells.rs

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

test_fixtures/src/bin/regenerate_test_fixture.rs

@@ -11,9 +11,9 @@ use sequencer_core::block_store::SequencerStore;
 use test_fixtures::{
     config,
     setup::{
-        prebuilt_sequencer_db_dump_path, setup_bedrock_node,
+        SequencerSetup, prebuilt_sequencer_db_dump_path, setup_bedrock_node,
         setup_private_accounts_with_initial_supply, setup_public_accounts_with_initial_supply,
-        setup_sequencer, setup_wallet,
+        setup_wallet,
     },
 };
 use wallet::config::WalletConfigOverrides;
@@ -49,15 +49,12 @@ async fn generate_prebuilt_fixture(dest: &Path) -> Result<()> {
     let genesis =
         config::genesis_from_accounts(&initial_public_accounts, &initial_private_accounts);
 
-    let (sequencer_handle, temp_sequencer_dir) = setup_sequencer(
+    let (sequencer_handle, temp_sequencer_dir) =
-        config::SequencerPartialConfig::default(),
+        SequencerSetup::new(config::SequencerPartialConfig::default(), bedrock_addr)
-        bedrock_addr,
+            .with_genesis(genesis)
-        genesis,
+            .setup()
-        config::bedrock_channel_id(),
+            .await
-        None,
+            .context("Failed to setup Sequencer for fixture generation")?;
-    )
-    .await
-    .context("Failed to setup Sequencer for fixture generation")?;
 
     let (mut wallet, _temp_wallet_dir, _wallet_password) = setup_wallet(
         sequencer_handle.addr(),

test_fixtures/src/config.rs

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

test_fixtures/src/lib.rs

@@ -24,8 +24,8 @@ use wallet::{
 use crate::{
     indexer_client::IndexerClient,
     setup::{
-        setup_bedrock_node, setup_indexer, setup_private_accounts_with_initial_supply,
+        SequencerSetup, setup_bedrock_node, setup_indexer,
-        setup_public_accounts_with_initial_supply, setup_sequencer, setup_sequencer_from_prebuilt,
+        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 {
+        let mut sequencer_setup = SequencerSetup::new(partial_config, bedrock_addr);
-            setup_sequencer_from_prebuilt(partial_config, bedrock_addr)
+        if !use_prebuilt {
-                .await
-                .context("Failed to setup Sequencer from prebuilt database")?
-        } else {
             // 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(
+            sequencer_setup = sequencer_setup.with_genesis(genesis);
-                partial_config,
+        }
-                bedrock_addr,
+        let (sequencer_handle, temp_sequencer_dir) = sequencer_setup
-                genesis,
+            .setup()
-                config::bedrock_channel_id(),
-                None,
-            )
             .await
-            .context("Failed to setup Sequencer")?
+            .context("Failed to setup Sequencer")?;
-        };
 
         let (mut wallet, temp_wallet_dir, wallet_password) = setup_wallet(
             sequencer_handle.addr(),

test_fixtures/src/setup.rs

@@ -1,4 +1,7 @@
-use std::{net::SocketAddr, path::PathBuf};
+use std::{
+    net::SocketAddr,
+    path::{Path, PathBuf},
+};
 
 use anyhow::{Context as _, Result, bail};
 use indexer_service::{ChannelId, IndexerHandle};
@@ -20,12 +23,106 @@ use crate::{
     private_mention, public_mention,
 };
 
-/// How to initialize the sequencer's database.
+pub struct SequencerSetup {
-pub enum SequencerInit<'dump> {
+    partial: config::SequencerPartialConfig,
-    /// Apply these genesis actions from scratch.
+    bedrock_addr: SocketAddr,
-    Genesis(Vec<GenesisAction>),
+    channel_id: ChannelId,
-    /// Restore an existing chain from a prebuilt dump (genesis actions are then irrelevant).
+    genesis_transactions: Option<Vec<GenesisAction>>,
-    Prebuilt(&'dump DbDump),
+    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 async fn setup_wallet(
     sequencer_addr: SocketAddr,
     initial_public_accounts: &[(PrivateKey, u128)],

tools/cross_zone_chat/src/main.rs

@@ -66,7 +66,7 @@ use sequencer_service_rpc::{RpcClient as _, SequencerClient, SequencerClientBuil
 use serde::{Deserialize, Serialize};
 use test_fixtures::{
     config::{self, SequencerPartialConfig, UrlProtocol, bedrock_channel_id, bedrock_channel_id_b},
-    setup::{setup_bedrock_node, setup_sequencer},
+    setup::{SequencerSetup, setup_bedrock_node},
 };
 
 const HTTP_PORT: u16 = 8088;
@@ -289,14 +289,20 @@ async fn main() -> Result<()> {
     let cross_zone_b = watch_peer(zone_a, receiver_id);
 
     let partial = SequencerPartialConfig::default();
-    let (seq_a, _home_a) =
+    let (seq_a, _home_a) = SequencerSetup::new(partial, bedrock_addr)
-        setup_sequencer(partial, bedrock_addr, vec![], channel_a, Some(cross_zone_a))
+        .with_genesis(vec![])
-            .await
+        .with_channel_id(channel_a)
-            .context("Failed to set up zone A sequencer")?;
+        .with_cross_zone(cross_zone_a)
-    let (seq_b, _home_b) =
+        .setup()
-        setup_sequencer(partial, bedrock_addr, vec![], channel_b, Some(cross_zone_b))
+        .await
-            .await
+        .context("Failed to set up zone A sequencer")?;
-            .context("Failed to set up zone B 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 {