Merge pull request #693 from logos-blockchain/moudy/cross-zone-per-block-seen-shard

Cargo.lock

@@ -4388,7 +4388,6 @@ dependencies = [
  "lee",
  "lee_core",
  "log",
- "logos-blockchain-core",
  "logos-blockchain-key-management-system-service",
  "ping_core",
  "programs",
@@ -9573,6 +9572,7 @@ dependencies = [
  "hex",
  "lee",
  "lee_core",
+ "serde",
  "serde_with",
 ]
 

Justfile

@@ -155,7 +155,7 @@ cross-zone-chat:
 clean:
     @echo "🧹 Cleaning run artifacts"
     rm -rf lez/sequencer/service/bedrock_signing_key
-    rm -rf lez/sequencer/service/rocksdb
+    rm -rf lez/sequencer/service/rocksdb*
     rm -rf lez/indexer/service/rocksdb*
     rm -rf lez/wallet/configs/debug/storage.json
     rm -rf lez/wallet/configs/debug/statistics.json

README.md

@@ -169,9 +169,9 @@ The sequencer and logos blockchain node can be run locally:
 
 After stopping services above you need to remove 3 folders to start cleanly:
  1. In the `logos-blockchain/logos-blockchain` folder `state` (not needed in case of docker setup)
- 2. In the `logos-execution-zone` folder `lez/sequencer/service/rocksdb`
+ 2. In the `logos-execution-zone` folder `lez/sequencer/service/rocksdb-<channel id>`
  3. In the `logos-execution-zone` file `lez/sequencer/service/bedrock_signing_key`
- 4. In the `logos-execution-zone` folder `lez/indexer/service/rocksdb`
+ 4. In the `logos-execution-zone` folder `lez/indexer/service/rocksdb-<channel id>`
 
 ### Normal mode (`just` commands)
 We provide a `Justfile` for developer and user needs, you can run the whole setup with it. The only difference will be that logos-blockchain (bedrock) will be started from docker.

build_utils/src/lib.rs

@@ -17,11 +17,19 @@ use anyhow::{Context as _, Result, bail};
 /// }
 /// ```
 pub fn include_artifacts(artifacts_sub_dir: &str) -> Result<()> {
-    let manifest_dir = PathBuf::from(std::env!("CARGO_MANIFEST_DIR"));
+    // Resolved at build-script runtime from the invoking crate, not at compile
+    // time: `env!` would bake in the path of whichever checkout compiled this
+    // rlib first, and with a shared cargo target dir every other worktree then
+    // embeds that checkout's artifacts instead of its own.
+    let invoking_manifest_dir = PathBuf::from(env::var("CARGO_MANIFEST_DIR")?);
+    let workspace_root = invoking_manifest_dir
+        .ancestors()
+        .find(|dir| dir.join("artifacts").is_dir())
+        .context("no artifacts/ directory above the invoking crate")?;
     let out_dir = PathBuf::from(env::var("OUT_DIR")?);
     let mod_dir = out_dir.join(artifacts_sub_dir);
     let mod_file = mod_dir.join("mod.rs");
-    let artifacts_dir = manifest_dir.join(format!("../artifacts/{artifacts_sub_dir}/"));
+    let artifacts_dir = workspace_root.join(format!("artifacts/{artifacts_sub_dir}/"));
 
     println!("cargo:rerun-if-changed={}", artifacts_dir.display());
 

integration_tests/Cargo.toml

@@ -36,7 +36,6 @@ programs.workspace = true
 test_programs.workspace = true
 testnet_initial_state.workspace = true
 
-logos-blockchain-core.workspace = true
 logos-blockchain-key-management-system-service.workspace = true
 anyhow.workspace = true
 log.workspace = true

integration_tests/tests/cross_zone_ingress_guard.rs

@@ -44,6 +44,7 @@ async fn user_origin_inbox_call_rejected() -> Result<()> {
     let msg = CrossZoneMessage {
         src_zone: [2; 32],
         src_block_id: 1,
+        src_block_hash: [7; 32],
         src_tx_index: 0,
         src_program_id: [9; 8],
         target_program_id: programs::ping_receiver().id(),

integration_tests/tests/cross_zone_state_machine.rs

@@ -14,7 +14,7 @@ use std::collections::BTreeMap;
 
 use cross_zone_inbox_core::{
     CrossZoneMessage, CrossZoneRoute, InboxConfig, Instruction as InboxInstruction, SeenShard,
-    inbox_config_account_id, inbox_seen_shard_account_id, message_key,
+    inbox_config_account_id, inbox_seen_shard_account_id,
 };
 use cross_zone_outbox_core::{OutboxRecord, outbox_pda};
 use lee::{
@@ -27,6 +27,8 @@ use ping_core::{ReceiverInstruction, ping_record_pda};
 const INITIAL_BALANCE: u128 = 100;
 const LOCK_AMOUNT: u128 = 30;
 const RECIPIENT: [u8; 32] = [9; 32];
+/// These tests drive the guest directly, so any fixed source-block hash does.
+const SRC_BLOCK_HASH: [u8; 32] = [7; 32];
 
 /// State registering the cross-zone builtins these tests exercise.
 fn base_state() -> V03State {
@@ -94,14 +96,86 @@ fn seed_wrapped_config(state: &mut V03State) {
 /// The wrapped-token `Mint` the bridge forwards, serialized as the cross-zone
 /// payload (risc0 words, little-endian bytes).
 fn mint_payload() -> Vec<u8> {
+    mint_payload_of(LOCK_AMOUNT)
+}
+
+fn mint_payload_of(amount: u128) -> Vec<u8> {
     let mint = wrapped_token_core::Instruction::Mint {
         recipient: RECIPIENT,
-        amount: LOCK_AMOUNT,
+        amount,
     };
     let words = risc0_zkvm::serde::to_vec(&mint).expect("serialize mint");
     words.iter().flat_map(|word| word.to_le_bytes()).collect()
 }
 
+/// Runs a bridge mint of `amount` through the inbox, as the watcher would.
+fn dispatch_mint(amount: u128) -> Result<ValidatedStateDiff, lee::error::LeeError> {
+    let inbox_id = programs::cross_zone_inbox().id();
+    let wrapped_token_id = programs::wrapped_token().id();
+    let self_zone = [1_u8; 32];
+    let src_zone = [2_u8; 32];
+    let src_block_id = 5;
+
+    let mut state = base_state();
+    seed_inbox_config(
+        &mut state,
+        self_zone,
+        src_zone,
+        [9_u32; 8],
+        wrapped_token_id,
+    );
+    seed_wrapped_config(&mut state);
+
+    let msg = CrossZoneMessage {
+        src_zone,
+        src_block_id,
+        src_block_hash: SRC_BLOCK_HASH,
+        src_tx_index: 0,
+        src_program_id: [9_u32; 8],
+        target_program_id: wrapped_token_id,
+        payload: mint_payload_of(amount),
+        l1_inclusion_witness: None,
+    };
+
+    let message = Message::try_new(
+        inbox_id,
+        vec![
+            inbox_config_account_id(inbox_id),
+            inbox_seen_shard_account_id(inbox_id, &src_zone, src_block_id),
+            wrapped_token_core::config_account_id(wrapped_token_id),
+            wrapped_token_core::holding_account_id(wrapped_token_id, &RECIPIENT),
+        ],
+        vec![],
+        InboxInstruction::Dispatch(msg),
+    )
+    .expect("build dispatch message");
+    let tx = PublicTransaction::new(message, WitnessSet::from_raw_parts(vec![]));
+
+    ValidatedStateDiff::from_public_transaction(&tx, &state, 1, 0)
+}
+
+/// One message must not be able to pin a holding near `u128::MAX`, which would
+/// make every later honest mint to that recipient overflow and fail for good.
+#[test]
+fn a_mint_above_the_cap_is_rejected() {
+    assert!(
+        dispatch_mint(wrapped_token_core::MAX_MINT_AMOUNT + 1).is_err(),
+        "an amount over the per-mint cap must not execute"
+    );
+}
+
+#[test]
+fn a_mint_at_the_cap_is_accepted() {
+    let diff = dispatch_mint(wrapped_token_core::MAX_MINT_AMOUNT)
+        .expect("the cap itself is a legitimate amount");
+    let holding_id =
+        wrapped_token_core::holding_account_id(programs::wrapped_token().id(), &RECIPIENT);
+    let minted = wrapped_token_core::read_balance(
+        &diff.public_diff()[&holding_id].data.clone().into_inner(),
+    );
+    assert_eq!(minted, wrapped_token_core::MAX_MINT_AMOUNT);
+}
+
 /// Drives `cross_zone_inbox::Dispatch` directly through the state machine
 /// (no watcher) and asserts the message is delivered to `ping_receiver`, which
 /// records the payload into its own PDA.
@@ -129,6 +203,7 @@ fn inbox_dispatch_delivers_payload_to_ping_receiver() {
     let msg = CrossZoneMessage {
         src_zone,
         src_block_id,
+        src_block_hash: SRC_BLOCK_HASH,
         src_tx_index: 0,
         src_program_id: [9_u32; 8],
         target_program_id: receiver_id,
@@ -241,53 +316,9 @@ fn lock_escrows_balance_and_emits_to_outbox() {
 /// and asserts it chains into `wrapped_token::Mint`, crediting the recipient.
 #[test]
 fn inbox_dispatch_mints_wrapped_token() {
-    let inbox_id = programs::cross_zone_inbox().id();
-    let wrapped_token_id = programs::wrapped_token().id();
-
-    let self_zone = [1_u8; 32];
-    let src_zone = [2_u8; 32];
-    let src_block_id = 5;
-
-    let mut state = base_state();
-    seed_inbox_config(
-        &mut state,
-        self_zone,
-        src_zone,
-        [9_u32; 8],
-        wrapped_token_id,
-    );
-    seed_wrapped_config(&mut state);
-
-    let msg = CrossZoneMessage {
-        src_zone,
-        src_block_id,
-        src_tx_index: 0,
-        src_program_id: [9_u32; 8],
-        target_program_id: wrapped_token_id,
-        payload: mint_payload(),
-        l1_inclusion_witness: None,
-    };
-
-    let seen_id = inbox_seen_shard_account_id(inbox_id, &src_zone, src_block_id);
-    let wrapped_config_id = wrapped_token_core::config_account_id(wrapped_token_id);
-    let holding_id = wrapped_token_core::holding_account_id(wrapped_token_id, &RECIPIENT);
-
-    let message = Message::try_new(
-        inbox_id,
-        vec![
-            inbox_config_account_id(inbox_id),
-            seen_id,
-            wrapped_config_id,
-            holding_id,
-        ],
-        vec![],
-        InboxInstruction::Dispatch(msg),
-    )
-    .expect("build dispatch message");
-    let tx = PublicTransaction::new(message, WitnessSet::from_raw_parts(vec![]));
-
-    let diff = ValidatedStateDiff::from_public_transaction(&tx, &state, 1, 0)
-        .expect("dispatch must validate and execute");
+    let diff = dispatch_mint(LOCK_AMOUNT).expect("dispatch must validate and execute");
+    let holding_id =
+        wrapped_token_core::holding_account_id(programs::wrapped_token().id(), &RECIPIENT);
     let minted = wrapped_token_core::read_balance(
         &diff.public_diff()[&holding_id].data.clone().into_inner(),
     );
@@ -326,6 +357,7 @@ fn a_mint_from_an_unrouted_emitter_is_rejected() {
     let msg = CrossZoneMessage {
         src_zone,
         src_block_id,
+        src_block_hash: SRC_BLOCK_HASH,
         src_tx_index: 0,
         // The emitter a user can drive directly, aimed at the bridge's target.
         src_program_id: programs::ping_sender().id(),
@@ -384,6 +416,7 @@ fn a_mint_from_the_routed_emitter_is_accepted() {
     let msg = CrossZoneMessage {
         src_zone,
         src_block_id,
+        src_block_hash: SRC_BLOCK_HASH,
         src_tx_index: 0,
         src_program_id: bridge_lock_id,
         target_program_id: wrapped_token_id,
@@ -440,12 +473,13 @@ fn mint_replay_rejected() {
     );
     seed_wrapped_config(&mut state);
 
-    // Seed the seen-shard as already containing this message's key, so the inbox
-    // takes the replay no-op branch. The shard is inbox-owned (claimed on a prior
-    // delivery), so the guest leaves it untouched.
+    // Seed the seen-shard as already holding this delivery, so the inbox takes
+    // the replay no-op branch. The shard is inbox-owned (claimed on a prior
+    // delivery) and bound to the same source block, so the guest leaves it
+    // untouched.
     let seen_id = inbox_seen_shard_account_id(inbox_id, &src_zone, src_block_id);
     let mut shard = SeenShard::default();
-    shard.insert(message_key(&src_zone, src_block_id, src_tx_index));
+    shard.insert(SRC_BLOCK_HASH, src_tx_index);
     state = state.with_public_accounts([(
         seen_id,
         Account {
@@ -462,6 +496,7 @@ fn mint_replay_rejected() {
     let msg = CrossZoneMessage {
         src_zone,
         src_block_id,
+        src_block_hash: SRC_BLOCK_HASH,
         src_tx_index,
         src_program_id: [9_u32; 8],
         target_program_id: wrapped_token_id,
@@ -503,3 +538,116 @@ fn mint_replay_rejected() {
         assert_eq!(shard_after, shard, "replay must not modify the seen-shard");
     }
 }
+
+/// A peer publishing two blocks at one block id gets at most one delivered from.
+///
+/// Both resolve to the same shard account; the first binds it. Failing rather
+/// than no-opping is the point: a replay no-op would let a peer choose which of
+/// two messages at one coordinate the target program ever sees.
+#[test]
+fn a_delivery_from_a_second_block_at_the_same_id_is_refused() {
+    let inbox_id = programs::cross_zone_inbox().id();
+    let receiver_id = programs::ping_receiver().id();
+
+    let self_zone = [1_u8; 32];
+    let src_zone = [2_u8; 32];
+    let src_block_id = 5;
+    let other_block_hash = [8_u8; 32];
+
+    let mut state = base_state();
+    seed_inbox_config(&mut state, self_zone, src_zone, [9_u32; 8], receiver_id);
+
+    // The shard as the first delivery left it: bound, holding transaction 0.
+    let seen_id = inbox_seen_shard_account_id(inbox_id, &src_zone, src_block_id);
+    let mut shard = SeenShard::default();
+    shard.insert(SRC_BLOCK_HASH, 0);
+    state = state.with_public_accounts([(
+        seen_id,
+        Account {
+            program_owner: inbox_id,
+            balance: 0,
+            data: shard
+                .to_bytes()
+                .try_into()
+                .expect("shard fits in account data"),
+            nonce: 0_u128.into(),
+        },
+    )]);
+
+    let words = risc0_zkvm::serde::to_vec(&ReceiverInstruction::Record {
+        payload: b"from-the-other-block".to_vec(),
+    })
+    .expect("serialize ping instruction");
+    let payload: Vec<u8> = words.iter().flat_map(|word| word.to_le_bytes()).collect();
+
+    // A different transaction index, so this is not a replay: only the source
+    // block differs from what the shard is bound to.
+    let msg = CrossZoneMessage {
+        src_zone,
+        src_block_id,
+        src_block_hash: other_block_hash,
+        src_tx_index: 1,
+        src_program_id: [9_u32; 8],
+        target_program_id: receiver_id,
+        payload,
+        l1_inclusion_witness: None,
+    };
+
+    let record_id = ping_record_pda(receiver_id);
+    let message = Message::try_new(
+        inbox_id,
+        vec![inbox_config_account_id(inbox_id), seen_id, record_id],
+        vec![],
+        InboxInstruction::Dispatch(msg),
+    )
+    .expect("build dispatch message");
+    let tx = PublicTransaction::new(message, WitnessSet::from_raw_parts(vec![]));
+
+    assert!(
+        ValidatedStateDiff::from_public_transaction(&tx, &state, 1, 0).is_err(),
+        "a delivery from a block the shard is not bound to must not execute"
+    );
+
+    // Control: the same delivery naming the bound block executes, so the refusal
+    // above is the binding and not the transaction's shape.
+    let control_words = risc0_zkvm::serde::to_vec(&ReceiverInstruction::Record {
+        payload: b"from-the-bound-block".to_vec(),
+    })
+    .expect("serialize ping instruction");
+    let control_msg = CrossZoneMessage {
+        src_zone,
+        src_block_id,
+        src_block_hash: SRC_BLOCK_HASH,
+        src_tx_index: 1,
+        src_program_id: [9_u32; 8],
+        target_program_id: receiver_id,
+        payload: control_words
+            .iter()
+            .flat_map(|word| word.to_le_bytes())
+            .collect(),
+        l1_inclusion_witness: None,
+    };
+    let control_message = Message::try_new(
+        inbox_id,
+        vec![inbox_config_account_id(inbox_id), seen_id, record_id],
+        vec![],
+        InboxInstruction::Dispatch(control_msg),
+    )
+    .expect("build dispatch message");
+    let control_tx = PublicTransaction::new(control_message, WitnessSet::from_raw_parts(vec![]));
+
+    let diff = ValidatedStateDiff::from_public_transaction(&control_tx, &state, 1, 0)
+        .expect("a second delivery from the bound block executes");
+    let public_diff = diff.public_diff();
+    let seen_after = public_diff
+        .get(&seen_id)
+        .expect("the shard records the new delivery");
+    let shard_after =
+        SeenShard::from_bytes(&seen_after.data.clone().into_inner()).expect("seen shard decodes");
+    assert!(shard_after.contains(0), "the first delivery is still there");
+    assert!(shard_after.contains(1), "and the second is recorded");
+    assert_eq!(
+        shard_after.src_block_hash, SRC_BLOCK_HASH,
+        "a shard stays bound to the block that claimed it"
+    );
+}

integration_tests/tests/sequencer_bootstrap.rs

@@ -13,7 +13,6 @@ use std::{path::Path, time::Duration};
 use anyhow::{Context as _, Result, bail};
 use indexer_service_rpc::RpcClient as _;
 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};
 use test_fixtures::{
@@ -237,8 +236,11 @@ async fn empty_local_reconstructs_from_populated_bedrock() -> Result<()> {
     // 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")?;
+    std::fs::remove_dir_all(home_a.path().join(format!(
+        "rocksdb-{}",
+        test_fixtures::config::bedrock_channel_id()
+    )))
+    .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)
@@ -274,11 +276,13 @@ async fn empty_local_reconstructs_from_populated_bedrock() -> Result<()> {
 /// 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.
+/// restarted with the same channel id against a Bedrock node where that channel
+/// is empty — i.e. the channel it committed to was wiped. 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.
+/// A *different* channel id no longer exercises this, because the db path is
+/// per-channel and a new id simply fresh-starts beside the old store.
 #[test]
 async fn nonempty_local_against_empty_channel_fails_startup() -> Result<()> {
     const PRODUCED_TARGET: u64 = 3;
@@ -310,9 +314,12 @@ async fn nonempty_local_against_empty_channel_fails_startup() -> Result<()> {
     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]);
+    // Restart on the SAME home (A's committed store: blocks + checkpoint) and the
+    // SAME channel id, but against a fresh Bedrock node where that channel does
+    // not exist — the channel it committed to is gone.
+    let (_bedrock_b, bedrock_addr_b) = setup_bedrock_node()
+        .await
+        .context("Failed to setup second Bedrock")?;
 
     // Startup aborts on the missing-channel invariant (a panic in
     // `start_from_config`). Run it on a dedicated OS thread with its own runtime
@@ -325,8 +332,7 @@ async fn nonempty_local_against_empty_channel_fails_startup() -> Result<()> {
         runtime.block_on(async {
             tokio::time::timeout(
                 Duration::from_secs(90),
-                SequencerSetup::new(slow_blocks(), bedrock_addr)
-                    .with_channel_id(empty_channel)
+                SequencerSetup::new(slow_blocks(), bedrock_addr_b)
                     .with_genesis(genesis)
                     .setup_at(&home_a_path),
             )
@@ -473,7 +479,10 @@ async fn local_behind_channel_reconstructs_forward() -> Result<()> {
     ];
 
     let home = tempfile::tempdir().context("Failed to create sequencer home")?;
-    let rocksdb = home.path().join("rocksdb");
+    let rocksdb = home.path().join(format!(
+        "rocksdb-{}",
+        test_fixtures::config::bedrock_channel_id()
+    ));
 
     // Bring the sequencer up to an early tip, then stop it so its store is at rest.
     {

lez/cross_zone/src/lib.rs

@@ -31,6 +31,21 @@ pub struct Emission {
     pub payload: Vec<u8>,
 }
 
+/// Where a delivery came from on the peer chain.
+///
+/// One struct so the watcher and the verifier fill the same field list: their
+/// dispatch transactions for one emission must be byte-identical.
+///
+/// `src_block_hash` is the recomputed hash on both sides, never the declared
+/// `header.hash`, which the signature does not cover.
+pub struct EmissionSource {
+    pub src_zone: ZoneId,
+    pub src_block_id: u64,
+    pub src_block_hash: [u8; 32],
+    pub src_tx_index: u32,
+    pub src_program_id: ProgramId,
+}
+
 /// Whether a program may only be invoked by sequencer-origin transactions.
 ///
 /// The cross-zone inbox is injected solely by the watcher; a user-submitted call
@@ -118,19 +133,17 @@ fn build_inbox_dispatch_tx(
 /// Option B check).
 #[must_use]
 pub fn build_dispatch_from_emission(
-    src_zone: ZoneId,
-    src_block_id: u64,
-    src_tx_index: u32,
-    src_program_id: ProgramId,
+    source: &EmissionSource,
     target_program_id: ProgramId,
     target_accounts: &[[u8; 32]],
     payload: Vec<u8>,
 ) -> lee::PublicTransaction {
     let msg = CrossZoneMessage {
-        src_zone,
-        src_block_id,
-        src_tx_index,
-        src_program_id,
+        src_zone: source.src_zone,
+        src_block_id: source.src_block_id,
+        src_block_hash: source.src_block_hash,
+        src_tx_index: source.src_tx_index,
+        src_program_id: source.src_program_id,
         target_program_id,
         payload,
         l1_inclusion_witness: None,

lez/indexer/core/src/cross_zone_verifier.rs

@@ -6,7 +6,7 @@ use std::{
 
 use anyhow::anyhow;
 use common::{block::Block, transaction::LeeTransaction};
-use cross_zone::{build_dispatch_from_emission, extract_emission};
+use cross_zone::{EmissionSource, build_dispatch_from_emission, extract_emission};
 use cross_zone_inbox_core::{
     CrossZoneMessage, Instruction as InboxInstruction, MessageKey, ZoneId, message_key,
 };
@@ -63,6 +63,11 @@ pub enum CrossZoneVerifyError {
     },
 }
 
+/// The replay key plus the source block, which is what the inbox treats as one
+/// delivery. Skipping re-derivation on the key alone would wave through a
+/// dispatch the guest refuses, parking the block and holding ingestion.
+type SeenKey = (MessageKey, [u8; 32]);
+
 /// One peer zone's cached blocks, plus how far this reader has read them as an
 /// unbroken hash-linked run from the peer's genesis.
 #[derive(Default)]
@@ -278,7 +283,7 @@ pub struct CrossZoneVerifier {
     /// optional: a peer with no configured key is not signature-checked.
     peer_pubkeys: HashMap<ZoneId, PublicKey>,
     peers: PeerBlocks,
-    seen: Arc<RwLock<HashSet<MessageKey>>>,
+    seen: Arc<RwLock<HashSet<SeenKey>>>,
 }
 
 impl CrossZoneVerifier {
@@ -330,17 +335,14 @@ impl CrossZoneVerifier {
     /// forged dispatch reuse it to skip re-derivation while the inbox delivers the
     /// forgery. A key already seen is a replay the inbox no-ops, so it is accepted
     /// without re-derivation rather than halting on a legitimate re-delivery.
-    pub async fn verify_block(
-        &self,
-        block: &Block,
-    ) -> Result<Vec<MessageKey>, CrossZoneVerifyError> {
+    pub async fn verify_block(&self, block: &Block) -> Result<Vec<SeenKey>, CrossZoneVerifyError> {
         let mut verified = Vec::new();
         for tx in &block.body.transactions {
             let Some(msg) = Self::decode_dispatch(tx) else {
                 continue;
             };
 
-            let key = message_key(&msg.src_zone, msg.src_block_id, msg.src_tx_index);
+            let key = seen_key(&msg);
             if self.seen.read().await.contains(&key) {
                 debug!(
                     "Skipping already-seen cross-zone dispatch from zone {} block {} tx {} (replay no-op)",
@@ -375,7 +377,7 @@ impl CrossZoneVerifier {
     /// Marks the given dispatch keys seen, so a later replay of them is accepted
     /// without re-derivation. Call only after the block that carried them has been
     /// applied on chain (see [`Self::verify_block`]).
-    pub async fn record_seen(&self, keys: Vec<MessageKey>) {
+    pub async fn record_seen(&self, keys: Vec<SeenKey>) {
         if keys.is_empty() {
             return;
         }
@@ -455,11 +457,16 @@ impl CrossZoneVerifier {
             )));
         }
 
+        // Recomputed rather than read from `msg`, which would make the field
+        // attest to itself.
         Ok(build_dispatch_from_emission(
-            msg.src_zone,
-            msg.src_block_id,
-            msg.src_tx_index,
-            message.program_id,
+            &EmissionSource {
+                src_zone: msg.src_zone,
+                src_block_id: msg.src_block_id,
+                src_block_hash: peer_block.recompute_hash().0,
+                src_tx_index: msg.src_tx_index,
+                src_program_id: message.program_id,
+            },
             emission.target_program_id,
             &emission.target_accounts,
             emission.payload,
@@ -528,6 +535,13 @@ struct PeerPass {
     stalled_at: Option<Slot>,
 }
 
+fn seen_key(msg: &CrossZoneMessage) -> SeenKey {
+    (
+        message_key(&msg.src_zone, msg.src_block_id, msg.src_tx_index),
+        msg.src_block_hash,
+    )
+}
+
 /// Whether a block read off a peer's channel may enter the cache. The channel
 /// authorizes who may write, not what they may claim.
 ///
@@ -808,12 +822,29 @@ mod tests {
 
     /// The dispatch a watcher would inject for a `PEER_BLOCK_ID` emission of `payload`.
     fn dispatch(payload: &[u8]) -> LeeTransaction {
+        dispatch_naming_block_hash(payload, source_block_hash(payload))
+    }
+
+    /// The recomputed hash of the `PEER_BLOCK_ID` block carrying `payload`,
+    /// which is what an honest watcher puts in the dispatch.
+    fn source_block_hash(payload: &[u8]) -> [u8; 32] {
+        peer_chain(payload)
+            .last()
+            .expect("chain reaches PEER_BLOCK_ID")
+            .recompute_hash()
+            .0
+    }
+
+    fn dispatch_naming_block_hash(payload: &[u8], src_block_hash: [u8; 32]) -> LeeTransaction {
         let receiver_id = programs::ping_receiver().id();
         LeeTransaction::Public(build_dispatch_from_emission(
-            PEER_ZONE,
-            PEER_BLOCK_ID,
-            0,
-            programs::ping_sender().id(),
+            &EmissionSource {
+                src_zone: PEER_ZONE,
+                src_block_id: PEER_BLOCK_ID,
+                src_block_hash,
+                src_tx_index: 0,
+                src_program_id: programs::ping_sender().id(),
+            },
             receiver_id,
             &[ping_record_pda(receiver_id).into_value()],
             payload.to_vec(),
@@ -832,6 +863,24 @@ mod tests {
             .expect("dispatch matching the peer emission verifies");
     }
 
+    #[tokio::test]
+    async fn rejects_dispatch_naming_the_wrong_source_block_hash() {
+        let verifier = verifier();
+        cache_chain(&verifier, peer_chain(b"hi")).await;
+
+        // Only the claimed source hash is wrong. Detectable because the verifier
+        // recomputes it from the resolved block instead of reading the field.
+        let block =
+            produce_dummy_block(9, None, vec![dispatch_naming_block_hash(b"hi", [0xab; 32])]);
+        assert!(
+            matches!(
+                verifier.verify_block(&block).await,
+                Err(CrossZoneVerifyError::Forged(_))
+            ),
+            "a delivery claiming a source block hash the peer block does not have is forged"
+        );
+    }
+
     #[tokio::test]
     async fn rejects_dispatch_with_no_matching_emission() {
         let verifier = verifier();
@@ -892,18 +941,52 @@ mod tests {
         // Mark the delivery seen, as the ingest loop does once the block applies.
         verifier.record_seen(keys).await;
 
-        // A payload that cannot re-derive, under the key just recorded. Accepted
-        // only by the seen-key short circuit, since the inbox no-ops it on
-        // chain; `unaccepted_dispatch_does_not_poison_seen` asserts the same
-        // input is rejected when the key was never recorded, which is what makes
-        // this one about the short circuit rather than re-derivation.
-        let replay = produce_dummy_block(10, None, vec![dispatch(b"forged")]);
+        // A payload that cannot re-derive, under the key just recorded, which
+        // now names the source block as well as the coordinates. Accepted only
+        // by the seen-key short circuit, since the inbox no-ops it on chain;
+        // `unaccepted_dispatch_does_not_poison_seen` asserts the same input is
+        // rejected when the key was never recorded, which is what makes this one
+        // about the short circuit rather than re-derivation.
+        let replay = produce_dummy_block(
+            10,
+            None,
+            vec![dispatch_naming_block_hash(
+                b"forged",
+                source_block_hash(b"hi"),
+            )],
+        );
         verifier
             .verify_block(&replay)
             .await
             .expect("a replay is accepted as an on-chain no-op");
     }
 
+    #[tokio::test]
+    async fn a_seen_coordinate_does_not_excuse_a_different_source_block() {
+        let verifier = verifier();
+        cache_chain(&verifier, peer_chain(b"hi")).await;
+
+        let first = produce_dummy_block(9, None, vec![dispatch(b"hi")]);
+        let keys = verifier.verify_block(&first).await.expect("first verifies");
+        verifier.record_seen(keys).await;
+
+        // Same coordinates as the delivery just seen, different source block.
+        // The inbox refuses rather than no-ops it, so skipping re-derivation
+        // would wave through a dispatch that parks the block.
+        let other = produce_dummy_block(
+            10,
+            None,
+            vec![dispatch_naming_block_hash(b"hi", [0xab; 32])],
+        );
+        assert!(
+            matches!(
+                verifier.verify_block(&other).await,
+                Err(CrossZoneVerifyError::Forged(_))
+            ),
+            "the seen set must agree with the guest on what counts as a replay"
+        );
+    }
+
     #[tokio::test]
     async fn unaccepted_dispatch_does_not_poison_seen() {
         // A dispatch verified in a block that never applies (e.g. one that parks)

lez/programs/bridge_lock/src/main.rs

@@ -4,7 +4,7 @@ use lee_core::{
     account::AccountWithMetadata,
     program::{AccountPostState, ChainedCall, Claim, ProgramInput, ProgramOutput, read_lee_inputs},
 };
-use wrapped_token_core::Instruction as WrappedInstruction;
+use wrapped_token_core::{Instruction as WrappedInstruction, MAX_MINT_AMOUNT};
 
 fn main() {
     let (
@@ -44,6 +44,13 @@ fn main() {
         mint_amount, amount,
         "locked amount must equal the wrapped mint amount"
     );
+    // Before the debit, not on the destination: nothing releases an escrow, so
+    // an amount the destination will not mint has to fail in the submitter's own
+    // transaction.
+    assert!(
+        amount <= MAX_MINT_AMOUNT,
+        "locked amount exceeds what the wrapped token will mint"
+    );
 
     // pre_states: [holder holding (authorized), escrow PDA, outbox PDA].
     let [holder, escrow, outbox] = <[AccountWithMetadata; 3]>::try_from(pre_states)

lez/programs/cross_zone_inbox/core/src/lib.rs

@@ -7,12 +7,12 @@ use lee_core::{
 };
 use serde::{Deserialize, Serialize};
 
-/// Source blocks per seen-set shard, so no single seen account grows without bound.
-pub const EPOCH_BLOCKS: u64 = 10_000;
-
 const MESSAGE_KEY_DOMAIN: [u8; 32] = *b"/LEZ/v0.3/CrossZoneMsgKey/00000/";
 const INBOX_CONFIG_SEED: [u8; 32] = *b"/LEZ/v0.3/CrossZoneInboxCfg/000/";
-const INBOX_SEEN_SEED_DOMAIN: [u8; 32] = *b"/LEZ/v0.3/CrossZoneInboxSeen/00/";
+/// `/01/` because `/00/` keyed shards by epoch: an epoch and a block id are
+/// indistinguishable under one domain. Belt and braces, since the image id
+/// already relocates every PDA in this crate whenever the crate changes.
+const INBOX_SEEN_SEED_DOMAIN: [u8; 32] = *b"/LEZ/v0.3/CrossZoneInboxSeen/01/";
 
 /// Raw 32-byte zone (channel) id; the host maps it to the zone-sdk `ChannelId`.
 pub type ZoneId = [u8; 32];
@@ -69,6 +69,13 @@ pub struct CrossZoneConfig {
 pub struct CrossZoneMessage {
     pub src_zone: ZoneId,
     pub src_block_id: u64,
+    /// The source block's recomputed hash, never the `header.hash` it declares.
+    ///
+    /// The signature does not cover that field, so a correctly signed block can
+    /// carry a bogus one. Both the watcher and the verifier hash the block's
+    /// contents themselves and fill this from that, so the two agree on it
+    /// without either trusting what the peer wrote.
+    pub src_block_hash: [u8; 32],
     pub src_tx_index: u32,
     pub src_program_id: ProgramId,
     pub target_program_id: ProgramId,
@@ -115,12 +122,37 @@ impl InboxConfig {
     }
 }
 
-/// The replay keys seen for one `(src_zone, epoch)` shard.
+/// What one peer block has already delivered.
+///
+/// Indices, not message keys: the shard's address already binds
+/// `(src_zone, src_block_id)`, so a key stored inside it adds nothing.
+///
+/// A shard costs an account plus a 36-byte header and breaks even against a
+/// shared shard at about five deliveries. What that buys is saturation
+/// resistance: at 32 bytes per delivery one peer block could overflow the
+/// account, and the guest's only answer is a panic that costs the message.
 #[derive(Clone, Debug, Default, PartialEq, Eq, BorshSerialize, BorshDeserialize)]
-pub struct SeenShard(pub BTreeSet<MessageKey>);
+pub struct SeenShard {
+    /// Recomputed hash of the peer block this shard records deliveries from.
+    /// All-zero until the first delivery claims it.
+    pub src_block_hash: [u8; 32],
+    /// Indices of that block's transactions already delivered.
+    pub delivered: BTreeSet<u32>,
+}
 
 impl SeenShard {
-    /// Decodes a shard from account data; empty data is an empty shard.
+    /// Deliveries one shard can hold before it exceeds `DATA_MAX_LENGTH`.
+    ///
+    /// Borsh is 32 bytes of hash, a 4-byte count, then 4 bytes per index, so
+    /// this is exactly the 100 KiB an account may carry.
+    ///
+    /// Out of reach only because of the L1 inscription cap: a block inscribes as
+    /// one op near 1.75 MiB and a minimal emitting transaction is about 257
+    /// bytes, capping a peer block near 7,100 deliveries. Raising that L1 cap
+    /// past roughly 6.3 MiB puts this back in reach.
+    pub const MAX_DELIVERIES: usize = 25_591;
+
+    /// Decodes a shard from account data; empty data is an unclaimed shard.
     pub fn from_bytes(bytes: &[u8]) -> borsh::io::Result<Self> {
         if bytes.is_empty() {
             return Ok(Self::default());
@@ -133,14 +165,32 @@ impl SeenShard {
         borsh::to_vec(self).expect("SeenShard serializes")
     }
 
+    /// Whether a delivery from the block with this hash may be recorded here.
+    ///
+    /// An unclaimed shard binds to its first claimant. Unclaimed is the whole
+    /// value being default, not the hash being zero, so a shard holding any
+    /// delivery can never read as unclaimed.
     #[must_use]
-    pub fn contains(&self, key: &MessageKey) -> bool {
-        self.0.contains(key)
+    pub fn binds(&self, src_block_hash: &[u8; 32]) -> bool {
+        *self == Self::default() || self.src_block_hash == *src_block_hash
     }
 
-    /// Inserts a key; returns true if it was newly inserted.
-    pub fn insert(&mut self, key: MessageKey) -> bool {
-        self.0.insert(key)
+    #[must_use]
+    pub fn contains(&self, src_tx_index: u32) -> bool {
+        self.delivered.contains(&src_tx_index)
+    }
+
+    /// Binds the shard if unclaimed and records the delivery; true if new.
+    ///
+    /// A non-binding hash records nothing. The guest already asserts
+    /// [`Self::binds`], so this is a backstop against a future caller rebinding
+    /// a claimed shard and erasing which peer block delivered what.
+    pub fn insert(&mut self, src_block_hash: [u8; 32], src_tx_index: u32) -> bool {
+        if !self.binds(&src_block_hash) {
+            return false;
+        }
+        self.src_block_hash = src_block_hash;
+        self.delivered.insert(src_tx_index)
     }
 }
 
@@ -207,7 +257,7 @@ pub const fn inbox_config_seed() -> PdaSeed {
     PdaSeed::new(INBOX_CONFIG_SEED)
 }
 
-/// The seen-set shard for the `(src_zone, epoch)` the message falls in.
+/// The seen-set shard for the peer block the message came from.
 #[must_use]
 pub fn inbox_seen_shard_account_id(
     inbox_id: ProgramId,
@@ -218,15 +268,17 @@ pub fn inbox_seen_shard_account_id(
 }
 
 /// Seed of the seen-shard PDA, exposed so the guest can claim the account.
+///
+/// One shard per peer block, so a peer cannot accumulate deliveries from many
+/// blocks into one account.
 #[must_use]
 pub fn inbox_seen_shard_seed(src_zone: &ZoneId, src_block_id: u64) -> PdaSeed {
     use risc0_zkvm::sha::{Impl, Sha256 as _};
 
-    let src_epoch = src_block_id.wrapping_div(EPOCH_BLOCKS);
     let mut bytes = [0_u8; 72];
     bytes[..32].copy_from_slice(&INBOX_SEEN_SEED_DOMAIN);
     bytes[32..64].copy_from_slice(src_zone);
-    bytes[64..].copy_from_slice(&src_epoch.to_le_bytes());
+    bytes[64..].copy_from_slice(&src_block_id.to_le_bytes());
 
     let seed: [u8; 32] = Impl::hash_bytes(&bytes)
         .as_bytes()
@@ -236,6 +288,8 @@ pub fn inbox_seen_shard_seed(src_zone: &ZoneId, src_block_id: u64) -> PdaSeed {
 }
 #[cfg(test)]
 mod tests {
+    use lee_core::account::data::DATA_MAX_LENGTH;
+
     use super::*;
 
     fn zone(b: u8) -> ZoneId {
@@ -308,15 +362,86 @@ mod tests {
     }
 
     #[test]
-    fn seen_shards_split_on_epoch_boundary() {
+    fn every_peer_block_gets_its_own_seen_shard() {
         let id: ProgramId = [9; 8];
         assert_eq!(
-            inbox_seen_shard_account_id(id, &zone(1), 0),
-            inbox_seen_shard_account_id(id, &zone(1), EPOCH_BLOCKS - 1),
+            inbox_seen_shard_account_id(id, &zone(1), 7),
+            inbox_seen_shard_account_id(id, &zone(1), 7),
         );
         assert_ne!(
-            inbox_seen_shard_account_id(id, &zone(1), EPOCH_BLOCKS - 1),
-            inbox_seen_shard_account_id(id, &zone(1), EPOCH_BLOCKS),
+            inbox_seen_shard_account_id(id, &zone(1), 7),
+            inbox_seen_shard_account_id(id, &zone(1), 8),
+        );
+        assert_ne!(
+            inbox_seen_shard_account_id(id, &zone(1), 7),
+            inbox_seen_shard_account_id(id, &zone(2), 7),
+        );
+    }
+
+    #[test]
+    fn a_shard_binds_to_the_first_block_that_claims_it() {
+        let mut shard = SeenShard::default();
+        assert!(shard.binds(&[1; 32]), "an unclaimed shard binds to anyone");
+        assert!(shard.binds(&[2; 32]));
+
+        shard.insert([1; 32], 0);
+        assert!(shard.binds(&[1; 32]), "and to that block thereafter");
+        assert!(
+            !shard.binds(&[2; 32]),
+            "a second block claiming the same block id cannot share this shard"
+        );
+    }
+
+    #[test]
+    fn a_shard_records_deliveries_by_transaction_index() {
+        let mut shard = SeenShard::default();
+        assert!(!shard.contains(3));
+        assert!(shard.insert([1; 32], 3));
+        assert!(shard.contains(3));
+        assert!(
+            !shard.insert([1; 32], 3),
+            "a replay of the same delivery records nothing new"
+        );
+        assert!(shard.insert([1; 32], 4));
+    }
+
+    #[test]
+    fn an_unclaimed_shard_reads_as_empty_and_round_trips() {
+        assert_eq!(
+            SeenShard::from_bytes(&[]).expect("empty data decodes"),
+            SeenShard::default(),
+            "an absent account is an unclaimed shard, not a decode failure"
+        );
+
+        let mut shard = SeenShard::default();
+        shard.insert([5; 32], 1);
+        shard.insert([5; 32], 9);
+        assert_eq!(
+            SeenShard::from_bytes(&shard.to_bytes()).expect("shard decodes"),
+            shard
+        );
+    }
+
+    #[test]
+    fn a_full_shard_fits_in_account_data() {
+        let mut shard = SeenShard::default();
+        for index in 0..SeenShard::MAX_DELIVERIES {
+            shard.insert([5; 32], u32::try_from(index).expect("index fits"));
+        }
+        let max = usize::try_from(DATA_MAX_LENGTH.as_u64()).expect("cap fits in usize");
+        assert_eq!(
+            shard.to_bytes().len(),
+            max,
+            "MAX_DELIVERIES is exactly what an account can carry"
+        );
+
+        shard.insert(
+            [5; 32],
+            u32::try_from(SeenShard::MAX_DELIVERIES).expect("index fits"),
+        );
+        assert!(
+            shard.to_bytes().len() > max,
+            "and one more does not fit, so the guest would fail rather than truncate"
         );
     }
 }

lez/programs/cross_zone_inbox/src/main.rs

@@ -1,6 +1,6 @@
 use cross_zone_inbox_core::{
     CrossZoneMessage, InboxConfig, Instruction, SeenShard, inbox_config_account_id,
-    inbox_config_seed, inbox_seen_shard_account_id, inbox_seen_shard_seed, message_key,
+    inbox_config_seed, inbox_seen_shard_account_id, inbox_seen_shard_seed,
 };
 use lee_core::{
     account::{Account, AccountWithMetadata},
@@ -94,16 +94,28 @@ fn dispatch(
         "No route from this source program to this target program for this peer"
     );
 
-    let key = message_key(&msg.src_zone, msg.src_block_id, msg.src_tx_index);
     let mut shard =
         SeenShard::from_bytes(&seen.account.data.clone().into_inner()).expect("seen shard decodes");
-    let already_seen = shard.contains(&key);
+
+    // One block id, one delivering block. The address binds the zone and block
+    // id but not which block claimed them, so an equivocating peer's two blocks
+    // at one id land here; the first binds the shard and the second aborts.
+    //
+    // Before the replay check, not after: reaching the replay branch first would
+    // turn a wrong-block delivery into a silent no-op, which the indexer's
+    // already-seen short circuit would then wave through.
+    assert!(
+        shard.binds(&msg.src_block_hash),
+        "Seen shard is bound to a different peer block at this block id"
+    );
+
+    let already_seen = shard.contains(msg.src_tx_index);
 
     // On replay this is a no-op: the seen shard is untouched and no call is made.
     let (seen_post, chained_calls) = if already_seen {
         (unchanged(&seen), vec![])
     } else {
-        shard.insert(key);
+        shard.insert(msg.src_block_hash, msg.src_tx_index);
         let mut seen_account = seen.account.clone();
         seen_account.data = shard
             .to_bytes()

lez/programs/wrapped_token/core/src/lib.rs

@@ -8,6 +8,18 @@ use lee_core::{
 };
 use serde::{Deserialize, Serialize};
 
+/// The most one mint may credit.
+///
+/// The peer zone chooses the amount and the balance is a `u128`, so unbounded
+/// one delivery pins a holding near the maximum, every later honest mint
+/// overflows into a guest panic, and the holding is bricked for inbound
+/// transfers at a cost of one message. The cap does not remove that ceiling, it
+/// makes reaching it cost 2^64 deliveries instead of one.
+///
+/// `u64::MAX` is the bridge's bound, not one native balances obey. `bridge_lock`
+/// refuses a larger amount at the source so it fails before escrowing.
+pub const MAX_MINT_AMOUNT: u128 = 0xFFFF_FFFF_FFFF_FFFF;
+
 const CONFIG_SEED_DOMAIN: [u8; 32] = *b"/LEZ/v0.3/WrappedTokenConfig/00/";
 const HOLDING_SEED_DOMAIN: [u8; 32] = *b"/LEZ/v0.3/WrappedTokenHold/00000";
 

lez/programs/wrapped_token/src/main.rs

@@ -3,8 +3,8 @@ use lee_core::{
     program::{AccountPostState, Claim, ProgramInput, ProgramOutput, read_lee_inputs},
 };
 use wrapped_token_core::{
-    Instruction, balance_bytes, config_account_id, config_seed, holding_account_id, holding_seed,
-    minter_bytes, read_balance, read_minter,
+    Instruction, MAX_MINT_AMOUNT, balance_bytes, config_account_id, config_seed,
+    holding_account_id, holding_seed, minter_bytes, read_balance, read_minter,
 };
 
 fn main() {
@@ -70,6 +70,11 @@ fn mint(
         "second account must be the recipient holding PDA"
     );
 
+    assert!(
+        amount <= MAX_MINT_AMOUNT,
+        "mint amount exceeds the per-mint cap"
+    );
+    // The backstop against accumulation, which the per-mint cap does not bound.
     let new_balance = read_balance(&holding.account.data.clone().into_inner())
         .checked_add(amount)
         .expect("wrapped-token balance overflow");

lez/sequencer/core/metrics/src/names.rs

@@ -7,3 +7,5 @@ pub const MEMPOOL_TRANSACTION_APPLICATION_TIME: &str =
     "mempool_transaction_application_time_seconds";
 pub const TRANSACTIONS_PER_BLOCK: &str = "transactions_per_block";
 pub const MEMPOOL_FAILED_TRANSACTIONS_TOTAL: &str = "mempool_failed_transactions_total";
+pub const CROSS_ZONE_DISPATCHES_RETIRED_TOTAL: &str = "cross_zone_dispatches_retired_total";
+pub const CROSS_ZONE_DEAD_LETTER_DISPATCHES: &str = "cross_zone_dead_letter_dispatches";

lez/sequencer/core/metrics/src/record.rs

@@ -48,6 +48,8 @@ impl From<common::transaction::TxKind> for TxKind {
 pub fn init() {
     blocks_produced_total_counter().increment(0);
     mempool_failed_transactions_total_counter().increment(0);
+    cross_zone_dispatches_retired_total_counter().increment(0);
+    record_cross_zone_dead_letter_dispatches(0);
     record_mempool_size(0);
     record_chain_height(0);
 
@@ -165,3 +167,26 @@ fn mempool_failed_transactions_total_counter() -> Counter {
 pub fn increment_mempool_failed_transactions_total() {
     mempool_failed_transactions_total_counter().increment(1);
 }
+
+fn cross_zone_dispatches_retired_total_counter() -> Counter {
+    counter!(
+        description: "Cross-zone deliveries this sequencer gave up on after repeated execution failures",
+        unit: Unit::Count,
+        names::CROSS_ZONE_DISPATCHES_RETIRED_TOTAL
+    )
+}
+
+pub fn increment_cross_zone_dispatches_retired_total() {
+    cross_zone_dispatches_retired_total_counter().increment(1);
+}
+
+/// Retained dead letters. A gauge, not a counter: eviction and reconciliation
+/// make this fall as well as rise.
+pub fn record_cross_zone_dead_letter_dispatches(count: usize) {
+    gauge!(
+        description: "Given-up-on cross-zone deliveries currently retained for inspection",
+        unit: Unit::Count,
+        names::CROSS_ZONE_DEAD_LETTER_DISPATCHES
+    )
+    .set(u64::try_from(count).expect("Dead letter count should fit into u64") as f64);
+}

lez/sequencer/core/src/config.rs

@@ -93,6 +93,16 @@ impl SequencerConfig {
 
         Ok(serde_json::from_reader(reader)?)
     }
+
+    /// Where this sequencer's database lives, suffixed with the channel id like
+    /// the indexer's, so several sequencers can share a home directory. Only the
+    /// database is per-channel; `bedrock_signing_key` stays unsuffixed, so
+    /// sequencers sharing a home share one Bedrock identity.
+    #[must_use]
+    pub fn db_path(&self) -> PathBuf {
+        self.home
+            .join(format!("rocksdb-{}", self.bedrock_config.channel_id))
+    }
 }
 
 const fn default_max_block_size() -> ByteSize {

lez/sequencer/core/src/cross_zone_watcher.rs

@@ -1,7 +1,7 @@
 use std::{sync::Arc, time::Duration};
 
 use common::{HashType, block::Block, transaction::LeeTransaction};
-use cross_zone::{build_dispatch_from_emission, extract_emission};
+use cross_zone::{EmissionSource, build_dispatch_from_emission, extract_emission};
 use cross_zone_inbox_core::{CrossZoneRoute, message_key, routes_permit};
 use futures::{Stream, StreamExt as _};
 use lee::{GENESIS_BLOCK_ID, PublicKey};
@@ -468,7 +468,7 @@ where
                         );
                     }
                     Link::Next(block_hash) => {
-                        if !record_block_deliveries(&block, peer, dbio) {
+                        if !record_block_deliveries(&block, block_hash, peer, dbio) {
                             // Recording a delivery is what makes it survive the
                             // mempool. Letting the pass finish here would move
                             // the floor past this slot on a store that just
@@ -545,7 +545,15 @@ fn advance_cursor(dbio: &RocksDBIO, peer_zone: [u8; 32], cursor: &mut Option<Slo
 /// Returns `false` if a delivery could not be recorded, which the caller turns
 /// into a stall: the record is the only thing standing between a durable read
 /// position and a lost message.
-fn record_block_deliveries(block: &Block, peer: &PeerContext, dbio: &RocksDBIO) -> bool {
+///
+/// `block_hash` is the value [`link_against`] recomputed from the block's own
+/// contents, not `block.header.hash`, which the signature does not cover.
+fn record_block_deliveries(
+    block: &Block,
+    block_hash: HashType,
+    peer: &PeerContext,
+    dbio: &RocksDBIO,
+) -> bool {
     let peer_zone = peer.peer_zone;
     let self_zone = peer.self_zone;
     let allowed_routes = peer.allowed_routes.as_slice();
@@ -583,10 +591,13 @@ fn record_block_deliveries(block: &Block, peer: &PeerContext, dbio: &RocksDBIO)
 
         let src_tx_index = u32::try_from(index).unwrap_or(u32::MAX);
         let dispatch = build_dispatch_from_emission(
-            peer_zone,
-            block.header.block_id,
-            src_tx_index,
-            message.program_id,
+            &EmissionSource {
+                src_zone: peer_zone,
+                src_block_id: block.header.block_id,
+                src_block_hash: block_hash.0,
+                src_tx_index,
+                src_program_id: message.program_id,
+            },
             emission.target_program_id,
             &emission.target_accounts,
             emission.payload,
@@ -1145,6 +1156,42 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn a_recorded_delivery_names_the_hash_the_watcher_validated() {
+        let (_dir, dbio) = store();
+        let mut cursor = None;
+        let mut tip = None;
+
+        consume_peer_stream(
+            stream::iter(vec![peer_block_msg(1, 0)]),
+            &peer_context(),
+            &dbio,
+            &mut cursor,
+            &mut tip,
+        )
+        .await;
+
+        let records = dbio.get_pending_cross_zone_dispatches().unwrap();
+        assert_eq!(records.len(), 1, "the delivery must be recorded");
+        let tx = borsh::from_slice::<LeeTransaction>(&records[0].transaction).unwrap();
+        let LeeTransaction::Public(public_tx) = tx else {
+            panic!("a dispatch is a public transaction");
+        };
+        let Ok(cross_zone_inbox_core::Instruction::Dispatch(msg)) =
+            risc0_zkvm::serde::from_slice(&public_tx.message().instruction_data)
+        else {
+            panic!("the recorded transaction is an inbox dispatch");
+        };
+
+        // The indexer recomputes this independently when it re-derives the same
+        // transaction; a different block here is what makes the two disagree.
+        assert_eq!(
+            msg.src_block_hash,
+            chain_block(1).recompute_hash().0,
+            "the delivery names the block the watcher read it from"
+        );
+    }
+
     #[tokio::test]
     async fn a_delivery_that_cannot_be_recorded_holds_the_floor() {
         let (_dir, dbio) = store();

lez/sequencer/core/src/lib.rs

@@ -32,11 +32,14 @@ use mempool::{MemPool, MemPoolHandle};
 pub use mock::SequencerCoreWithMockClients;
 use num_bigint::BigUint;
 pub use storage::error::DbError;
+// Re-exported because `cross_zone_dead_letters` returns it and the service
+// crate does not depend on `storage`, so it could not otherwise name the type.
+pub use storage::sequencer::sequencer_cells::DeadLetterDispatchRecord;
 use storage::sequencer::{
-    RocksDBIO, StoreUpdate,
+    DispatchFailure, RocksDBIO, StoreUpdate,
     sequencer_cells::{
-        PendingCrossZoneDispatchRecord, PendingDepositEventRecord, WithdrawalReconciliationKey,
-        ZoneAnchorRecord,
+        DispatchOrigin, PendingCrossZoneDispatchRecord, PendingDepositEventRecord,
+        WithdrawalReconciliationKey, ZoneAnchorRecord,
     },
 };
 
@@ -116,7 +119,7 @@ impl<BP: BlockPublisherTrait> SequencerCore<BP> {
     /// initializing its state with the accounts defined in the configuration file.
     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");
+        let db_path = config.db_path();
 
         if db_path.exists() {
             let store = SequencerStore::open_db(&db_path, signing_key).unwrap_or_else(|err| {
@@ -130,6 +133,14 @@ impl<BP: BlockPublisherTrait> SequencerCore<BP> {
                 .expect("Failed to read state from store");
             (store, state)
         } else {
+            let legacy = config.home.join("rocksdb");
+            if legacy.exists() {
+                warn!(
+                    "Ignoring pre-channel-suffix database at {}; rename it to {} to resume it",
+                    legacy.display(),
+                    db_path.display()
+                );
+            }
             warn!(
                 "Database not found at {}, starting from genesis",
                 db_path.display()
@@ -333,6 +344,7 @@ impl<BP: BlockPublisherTrait> SequencerCore<BP> {
         };
 
         sequencer_core_metrics::record_chain_height(sequencer_core.chain_height());
+        record_dead_letter_gauge(&sequencer_core.store.dbio());
 
         (sequencer_core, mempool_handle)
     }
@@ -794,18 +806,22 @@ impl<BP: BlockPublisherTrait> SequencerCore<BP> {
             (prev, height, chain.head_state().clone(), pending)
         };
 
-        if !settled.is_empty()
-            && let Err(err) = self
+        if !settled.is_empty() {
+            if let Err(err) = self
                 .store
                 .dbio()
                 .drop_settled_cross_zone_dispatches(&settled)
-        {
-            // Only bookkeeping: the deliveries themselves are irreversible, and
-            // the next turn tries again.
-            warn!(
-                "Failed to drop {} settled delivery record(s): {err:#}",
-                settled.len()
-            );
+            {
+                // Only bookkeeping: the deliveries themselves are irreversible,
+                // and the next turn tries again.
+                warn!(
+                    "Failed to drop {} settled delivery record(s): {err:#}",
+                    settled.len()
+                );
+            }
+            // A settled delivery may be one this node had given up on, which
+            // takes its dead letter with it.
+            record_dead_letter_gauge(&self.store.dbio());
         }
 
         let mut valid_transactions = Vec::new();
@@ -1074,8 +1090,8 @@ impl<BP: BlockPublisherTrait> SequencerCore<BP> {
     /// A delivery's payload and target accounts are chosen on the peer zone and
     /// validated by nobody in between, so one can fail for good; but a failure
     /// can equally be a property of the moment, so give up only after several.
-    /// Giving up drops the record, which is also what keeps a peer from growing
-    /// the pending list with deliveries that can never execute.
+    /// Giving up moves the record to the dead letter: a peer cannot grow the
+    /// pending list with deliveries that never execute, and it stays findable.
     fn count_dispatch_failure(&self, tx: &LeeTransaction) {
         let Some(message) = extract_cross_zone_dispatch(tx) else {
             return;
@@ -1085,17 +1101,37 @@ impl<BP: BlockPublisherTrait> SequencerCore<BP> {
             message.src_block_id,
             message.src_tx_index,
         );
+        let origin = DispatchOrigin {
+            src_zone: message.src_zone,
+            src_block_id: message.src_block_id,
+            src_tx_index: message.src_tx_index,
+        };
         match self
             .store
             .dbio()
-            .record_dispatch_failure(key, RETIRE_DISPATCH_AFTER_FAILURES)
+            .record_dispatch_failure(key, RETIRE_DISPATCH_AFTER_FAILURES, origin)
         {
-            Ok(true) => error!(
-                "Giving up on cross-zone delivery {} after {RETIRE_DISPATCH_AFTER_FAILURES} failed attempts; it will not be retried",
+            Ok(DispatchFailure::Retired(record)) => {
+                sequencer_core_metrics::increment_cross_zone_dispatches_retired_total();
+                record_dead_letter_gauge(&self.store.dbio());
+                error!(
+                    "Giving up on cross-zone delivery {} from peer zone {} block {} transaction {} ({} bytes) after {} failed attempts. This node will not retry it; unless another sequencer carries it, the message is not delivered. Kept in the dead letter.",
+                    hex::encode(key),
+                    hex::encode(origin.src_zone),
+                    origin.src_block_id,
+                    origin.src_tx_index,
+                    record.transaction_bytes,
+                    record.failed_attempts
+                );
+            }
+            Ok(DispatchFailure::Retried { failed_attempts }) => warn!(
+                "Cross-zone delivery {} failed to execute ({failed_attempts} of {RETIRE_DISPATCH_AFTER_FAILURES} attempts), will retry next block",
                 hex::encode(key)
             ),
-            Ok(false) => warn!(
-                "Cross-zone delivery {} failed to execute, will retry next block",
+            // Not a give-up: the ordinary case is a delivery that already
+            // settled, so its record is gone and there is nothing left to lose.
+            Ok(DispatchFailure::Absent) => debug!(
+                "Cross-zone delivery {} failed to execute but has no pending record; nothing to count",
                 hex::encode(key)
             ),
             Err(err) => error!(
@@ -1105,6 +1141,18 @@ impl<BP: BlockPublisherTrait> SequencerCore<BP> {
         }
     }
 
+    /// The deliveries this node has given up on, and how many times it has.
+    ///
+    /// Retained is read first so the pair can only skew towards a total that
+    /// leads its list, an ordinary evicted or settled state. The other order
+    /// would report entries against a total of zero.
+    pub fn cross_zone_dead_letters(&self) -> Result<(u64, Vec<DeadLetterDispatchRecord>), DbError> {
+        let dbio = self.store.dbio();
+        let retained = dbio.get_dead_letter_cross_zone_dispatches()?;
+        let total = dbio.get_dead_letter_cross_zone_dispatch_count()?;
+        Ok((total, retained))
+    }
+
     /// A weak reference to this sequencer's store, for a shutdown path that
     /// needs to observe the database actually closing rather than infer it.
     #[must_use]
@@ -1188,10 +1236,14 @@ fn deposit_already_minted(state: &lee::V03State, deposit_op_id: HashType) -> boo
 
 /// Whether a cross-zone delivery is already on the chain we are building on.
 ///
-/// The inbox records every delivered message key in a seen shard and no-ops a
-/// replay, so that shard is the same kind of answer the deposit receipt gives:
-/// state, not bookkeeping. An orphan reverts the entry with the block, so the
-/// next turn re-delivers with nothing of ours to unwind.
+/// The inbox records each peer block's delivered indices in that block's seen
+/// shard and no-ops a replay, so the shard is the same kind of answer the
+/// deposit receipt gives: state, not bookkeeping. An orphan reverts the entry
+/// with the block, so the next turn re-delivers with nothing to unwind.
+///
+/// Both halves matter. A shard bound to a different peer block is not this
+/// delivery's replay record, it is what will make it abort, and calling that
+/// delivered would drop the record instead of dead-lettering it.
 fn dispatch_already_delivered(state: &lee::V03State, message: &CrossZoneMessage) -> bool {
     let shard_id = cross_zone_inbox_core::inbox_seen_shard_account_id(
         programs::cross_zone_inbox().id(),
@@ -1200,15 +1252,27 @@ fn dispatch_already_delivered(state: &lee::V03State, message: &CrossZoneMessage)
     );
     state.get_account_by_id_ref(shard_id).is_some_and(|shard| {
         cross_zone_inbox_core::SeenShard::from_bytes(shard.data.as_ref()).is_ok_and(|seen| {
-            seen.contains(&cross_zone_inbox_core::message_key(
-                &message.src_zone,
-                message.src_block_id,
-                message.src_tx_index,
-            ))
+            seen.binds(&message.src_block_hash) && seen.contains(message.src_tx_index)
         })
     })
 }
 
+/// Publishes how many given-up-on deliveries are retained.
+///
+/// Read from the store because the list falls as well as rises (eviction, and
+/// reconciliation when a delivery settles elsewhere). Costs a read and a decode,
+/// so call it only where one of those can have happened.
+fn record_dead_letter_gauge(dbio: &RocksDBIO) {
+    match dbio.get_dead_letter_cross_zone_dispatches() {
+        Ok(records) => {
+            sequencer_core_metrics::record_cross_zone_dead_letter_dispatches(records.len());
+        }
+        Err(err) => {
+            warn!("Failed to read the cross-zone dead letter for its gauge: {err:#}");
+        }
+    }
+}
+
 /// Feed one channel delta into the follow state and mirror it to the store:
 /// revert orphaned, then apply and persist adopted and finalized blocks.
 /// Production builds on this same head. Wired to the publisher via
@@ -1399,6 +1463,9 @@ fn apply_follow_update(
     };
 
     sequencer_core_metrics::record_chain_height(head_height);
+    // The runtime reconcile path: finalizing another sequencer's block drops the
+    // dead letter of a delivery this node gave up on.
+    record_dead_letter_gauge(dbio);
 
     if outcome.accepted_deposits > 0 {
         info!(

lez/sequencer/core/src/tests.rs

@@ -25,7 +25,7 @@ use logos_blockchain_zone_sdk::sequencer::DepositInfo;
 use mempool::MemPoolHandle;
 use ping_core::{ReceiverInstruction, ping_record_pda};
 use storage::sequencer::sequencer_cells::{
-    PendingCrossZoneDispatchRecord, PendingDepositEventRecord,
+    DispatchOrigin, PendingCrossZoneDispatchRecord, PendingDepositEventRecord,
 };
 use tempfile::tempdir;
 use testnet_initial_state::{initial_pub_accounts_private_keys, initial_public_user_accounts};
@@ -208,16 +208,27 @@ fn ping_payload(payload: &[u8]) -> Vec<u8> {
 fn dispatch_tx(src_block_id: u64, payload: Vec<u8>) -> LeeTransaction {
     let receiver_id = programs::ping_receiver().id();
     LeeTransaction::Public(cross_zone::build_dispatch_from_emission(
-        PEER_ZONE,
-        src_block_id,
-        0,
-        programs::ping_sender().id(),
+        &cross_zone::EmissionSource {
+            src_zone: PEER_ZONE,
+            src_block_id,
+            src_block_hash: peer_block_hash(src_block_id),
+            src_tx_index: 0,
+            src_program_id: programs::ping_sender().id(),
+        },
         receiver_id,
         &[ping_record_pda(receiver_id).into_value()],
         payload,
     ))
 }
 
+/// A stand-in for the peer block's recomputed hash, distinct per block id. These
+/// records are seeded into the store, so no real block exists to hash.
+fn peer_block_hash(src_block_id: u64) -> [u8; 32] {
+    let mut hash = [0_u8; 32];
+    hash[..8].copy_from_slice(&src_block_id.to_le_bytes());
+    hash
+}
+
 /// The pending record the watcher would leave behind for that dispatch.
 fn dispatch_record(src_block_id: u64, payload: Vec<u8>) -> PendingCrossZoneDispatchRecord {
     let tx = dispatch_tx(src_block_id, payload);
@@ -285,7 +296,7 @@ async fn start_from_config_opens_existing_db_if_it_exists() {
     let genesis_block = genesis_hashable_data.into_pending_block(&signing_key);
 
     SequencerStore::create_db_with_genesis(
-        &config.home.join("rocksdb"),
+        &config.db_path(),
         &genesis_block,
         &genesis_state,
         signing_key,
@@ -305,7 +316,7 @@ async fn start_from_config_panics_when_db_open_returns_non_not_found_error() {
     let temp_dir = tempdir().unwrap();
     config.home = temp_dir.path().to_path_buf();
 
-    let db_path = config.home.join("rocksdb");
+    let db_path = config.db_path();
 
     std::fs::create_dir_all(&config.home).unwrap();
     // Force RocksDB open to fail with an IO error by placing a file at DB path.
@@ -337,7 +348,7 @@ async fn unfulfilled_deposit_events_are_drained_from_the_store_on_production() {
 
     {
         let signing_key = lee::PrivateKey::try_new(config.signing_key).unwrap();
-        let store = SequencerStore::open_db(&config.home.join("rocksdb"), signing_key).unwrap();
+        let store = SequencerStore::open_db(&config.db_path(), signing_key).unwrap();
 
         let inserted = store
             .dbio()
@@ -697,15 +708,45 @@ async fn a_dispatch_that_never_executes_is_given_up_on_after_repeated_failures()
         );
     }
 
-    // The attempt at the limit gives up on it, and giving up drops the record.
-    // Anything else leaves an entry no later block can ever remove, which is how
-    // a peer that can make deliveries fail would grow this list without bound.
+    // The attempt at the limit gives up on it, which takes the record out of the
+    // pending list. Anything else leaves an entry no later block can ever
+    // remove, which is how a peer that can make deliveries fail would grow this
+    // list without bound.
     sequencer.produce_new_block().await.unwrap();
     assert!(
         pending_dispatches(&sequencer).is_empty(),
-        "giving up on a delivery must drop its record, not flag it"
+        "giving up on a delivery must take its record out of the pending list"
     );
 
+    // The dead letter is the only record that this happened, and the origin is
+    // what identifies which message stopped being attempted.
+    let dbio = sequencer.store.dbio();
+    let dead_letters = dbio.get_dead_letter_cross_zone_dispatches().unwrap();
+    assert_eq!(dead_letters.len(), 1);
+    assert_eq!(
+        dead_letters[0].origin,
+        DispatchOrigin {
+            src_zone: PEER_ZONE,
+            src_block_id: 13,
+            src_tx_index: 0,
+        }
+    );
+    assert_eq!(
+        dead_letters[0].message_key,
+        cross_zone_inbox_core::message_key(&PEER_ZONE, 13, 0)
+    );
+    assert!(dead_letters[0].transaction_bytes > 0);
+    assert_eq!(
+        dead_letters[0].failed_attempts,
+        RETIRE_DISPATCH_AFTER_FAILURES
+    );
+    assert_eq!(dbio.get_dead_letter_cross_zone_dispatch_count().unwrap(), 1);
+
+    // The same view the RPC serves, so an operator sees what the store holds.
+    let (total_retired, retained) = sequencer.cross_zone_dead_letters().unwrap();
+    assert_eq!(total_retired, 1);
+    assert_eq!(retained, dead_letters);
+
     // And nothing re-feeds it, so it stops costing a guest execution per block.
     let block_id = sequencer.produce_new_block().await.unwrap();
     let block = sequencer.store.get_block_at_id(block_id).unwrap().unwrap();

lez/sequencer/service/protocol/Cargo.toml

@@ -13,4 +13,5 @@ lee.workspace = true
 lee_core.workspace = true
 
 hex.workspace = true
+serde.workspace = true
 serde_with.workspace = true

lez/sequencer/service/protocol/src/lib.rs

@@ -5,11 +5,36 @@ use std::{fmt::Display, str::FromStr};
 pub use common::{HashType, block::Block, transaction::LeeTransaction};
 pub use lee::{Account, AccountId, ProgramId};
 pub use lee_core::{BlockId, Commitment, CommitmentSetDigest, MembershipProof, account::Nonce};
+use serde::{Deserialize, Serialize};
 use serde_with::{DeserializeFromStr, SerializeDisplay};
 
 #[derive(Debug, Clone, PartialEq, Eq, Hash, SerializeDisplay, DeserializeFromStr)]
 pub struct ChannelId(pub [u8; 32]);
 
+/// A cross-zone delivery a sequencer gave up on after repeated failures.
+///
+/// Identifies the message rather than carrying it: zone, block id and tx index
+/// locate it on the peer's channel.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct CrossZoneDeadLetter {
+    pub message_key: HashType,
+    pub src_zone: ChannelId,
+    pub src_block_id: u64,
+    pub src_tx_index: u32,
+    pub failed_attempts: u32,
+    pub transaction_bytes: u32,
+}
+
+/// What a sequencer has given up delivering.
+///
+/// `total_retired` counts every give-up, `retained` only the ones still kept;
+/// they diverge on eviction and on reconciliation.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct CrossZoneDeadLetterReport {
+    pub total_retired: u64,
+    pub retained: Vec<CrossZoneDeadLetter>,
+}
+
 impl Display for ChannelId {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         let hex_string = hex::encode(self.0);

lez/sequencer/service/rpc/src/lib.rs

@@ -6,8 +6,8 @@ use jsonrpsee::types::ErrorObjectOwned;
 #[cfg(feature = "client")]
 pub use jsonrpsee::{core::ClientError, http_client::HttpClientBuilder as SequencerClientBuilder};
 use sequencer_service_protocol::{
-    Account, AccountId, Block, BlockId, ChannelId, Commitment, CommitmentSetDigest, HashType,
-    LeeTransaction, MembershipProof, Nonce, ProgramId,
+    Account, AccountId, Block, BlockId, ChannelId, Commitment, CommitmentSetDigest,
+    CrossZoneDeadLetterReport, HashType, LeeTransaction, MembershipProof, Nonce, ProgramId,
 };
 
 #[cfg(all(not(feature = "server"), not(feature = "client")))]
@@ -90,4 +90,13 @@ pub trait Rpc {
 
     #[method(name = "getChannelId")]
     async fn get_channel_id(&self) -> Result<ChannelId, ErrorObjectOwned>;
+
+    /// The cross-zone deliveries this sequencer has given up on.
+    ///
+    /// Its own method rather than folded into `checkHealth`: one undeliverable
+    /// peer message must not read as an unhealthy node.
+    #[method(name = "getCrossZoneDeadLetters")]
+    async fn get_cross_zone_dead_letters(
+        &self,
+    ) -> Result<CrossZoneDeadLetterReport, ErrorObjectOwned>;
 }

lez/sequencer/service/src/service.rs

@@ -12,8 +12,8 @@ use sequencer_core::{
     DbError, SequencerCore, TransactionOrigin, block_publisher::BlockPublisherTrait,
 };
 use sequencer_service_protocol::{
-    Account, AccountId, Block, BlockId, ChannelId, Commitment, CommitmentSetDigest, HashType,
-    MembershipProof, Nonce, ProgramId,
+    Account, AccountId, Block, BlockId, ChannelId, Commitment, CommitmentSetDigest,
+    CrossZoneDeadLetter, CrossZoneDeadLetterReport, HashType, MembershipProof, Nonce, ProgramId,
 };
 use tokio::sync::Mutex;
 
@@ -217,6 +217,32 @@ impl<BC: BlockPublisherTrait + Send + Sync + 'static> sequencer_service_rpc::Rpc
         let channel_id = self.sequencer.lock().await.block_publisher().channel_id();
         Ok(ChannelId(*channel_id.as_ref()))
     }
+
+    async fn get_cross_zone_dead_letters(
+        &self,
+    ) -> Result<CrossZoneDeadLetterReport, ErrorObjectOwned> {
+        let (total_retired, records) = self
+            .sequencer
+            .lock()
+            .await
+            .cross_zone_dead_letters()
+            .map_err(|err| internal_error(&err))?;
+
+        Ok(CrossZoneDeadLetterReport {
+            total_retired,
+            retained: records
+                .into_iter()
+                .map(|record| CrossZoneDeadLetter {
+                    message_key: HashType(record.message_key),
+                    src_zone: ChannelId(record.origin.src_zone),
+                    src_block_id: record.origin.src_block_id,
+                    src_tx_index: record.origin.src_tx_index,
+                    failed_attempts: record.failed_attempts,
+                    transaction_bytes: record.transaction_bytes,
+                })
+                .collect(),
+        })
+    }
 }
 
 fn internal_error(err: &DbError) -> ErrorObjectOwned {

lez/storage/src/sequencer/mod.rs

@@ -20,6 +20,8 @@ use crate::{
     cells::shared_cells::{BlockCell, FirstBlockCell, FirstBlockSetCell, LastBlockCell},
     error::DbError,
     sequencer::sequencer_cells::{
+        DeadLetterCrossZoneDispatchCountCell, DeadLetterCrossZoneDispatchesCellOwned,
+        DeadLetterCrossZoneDispatchesCellRef, DeadLetterDispatchRecord, DispatchOrigin,
         FinalBlockMetaCellOwned, FinalBlockMetaCellRef, FinalLeeStateCellOwned,
         FinalLeeStateCellRef, LEEStateCellOwned, LEEStateCellRef, LastFinalizedBlockIdCell,
         LatestBlockMetaCellOwned, LatestBlockMetaCellRef, PeerChainTip, PeerFloorCellOwned,
@@ -55,6 +57,12 @@ pub const DB_META_CROSS_ZONE_PEER_TIP_KEY: &str = "cross_zone_peer_tip";
 /// Key base for storing cross-zone deliveries the watcher has recorded but
 /// which are not yet known to be irreversibly delivered.
 pub const DB_META_PENDING_CROSS_ZONE_DISPATCHES_KEY: &str = "pending_cross_zone_dispatches";
+/// Key base for storing cross-zone deliveries this node has given up on.
+pub const DB_META_DEAD_LETTER_CROSS_ZONE_DISPATCHES_KEY: &str = "dead_letter_cross_zone_dispatches";
+/// Key base for counting every cross-zone delivery given up on, including ones
+/// since evicted from the retained list or reconciled out of it.
+pub const DB_META_DEAD_LETTER_CROSS_ZONE_DISPATCH_COUNT_KEY: &str =
+    "dead_letter_cross_zone_dispatch_count";
 
 /// Key base for counting unseen L2 withdraw intents.
 pub const DB_META_UNSEEN_WITHDRAW_COUNT_KEY: &str = "unseen_withdraw_count";
@@ -73,6 +81,17 @@ pub const DB_META_PUBLISHED_HIGH_WATER_KEY: &str = "published_high_water";
 /// delivery floor and reads the slot again later.
 pub const MAX_PENDING_CROSS_ZONE_DISPATCHES: usize = 4096;
 
+/// How many given-up-on cross-zone deliveries are kept for inspection.
+///
+/// A peer chooses how many deliveries fail, so this cannot be unbounded. The
+/// oldest evicts at the cap, and nothing is concealed by that: retirements are
+/// counted separately and the count does not evict.
+///
+/// An entry count bounds bytes only because a record identifies a delivery
+/// rather than carrying it. At a fixed 84 bytes each the list is 21 KB, which
+/// matters because it is one value rewritten under the block-production lock.
+pub const MAX_DEAD_LETTER_CROSS_ZONE_DISPATCHES: usize = 256;
+
 /// Key base for storing the LEE state.
 pub const DB_LEE_STATE_KEY: &str = "lee_state";
 /// Key base for storing the LEE state at the last L1-finalized block.
@@ -83,6 +102,20 @@ pub const DB_FINAL_BLOCK_META_KEY: &str = "final_block_meta";
 /// Name of state column family.
 pub const CF_LEE_STATE_NAME: &str = "cf_lee_state";
 
+/// What counting a failed production attempt did to a delivery's record.
+///
+/// Three outcomes rather than a bool: only one means this node stopped trying,
+/// and a settled delivery has no record, so it is [`Self::Absent`].
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum DispatchFailure {
+    /// Counted; the delivery is still pending and will be attempted again.
+    Retried { failed_attempts: u32 },
+    /// Given up on: moved out of the pending list and into the dead letter.
+    Retired(Box<DeadLetterDispatchRecord>),
+    /// No pending record, so nothing was counted and nothing was given up on.
+    Absent,
+}
+
 /// A single key/value entry from a column family, used inside [`DbDump`].
 #[derive(BorshSerialize, BorshDeserialize)]
 pub struct DbDumpEntry {
@@ -742,38 +775,103 @@ impl RocksDBIO {
         Ok(accepted)
     }
 
-    /// Counts a failed production attempt against a delivery, dropping its
-    /// record once it reaches `retire_at`. Returns whether it was dropped.
+    /// Counts a failed production attempt against a delivery, retiring it once
+    /// it reaches `retire_at`.
     ///
-    /// Dropped rather than flagged: a retired record is one the drain will never
-    /// turn into a block transaction again, so nothing would ever remove it, and
-    /// a peer that can make deliveries fail could grow the list without bound.
-    /// The delivery is given up on either way; this way the cost is a log line
-    /// rather than a permanent entry.
+    /// The pending list has to lose the record, or the drain re-feeds a
+    /// transaction that never executes for ever. The dead letter keeps the
+    /// delivery identifiable, since a dispatch that fails execution is left out
+    /// of the block and leaves no trace elsewhere, and it is bounded separately.
     ///
-    /// A delivery with no record is already retired as far as this is concerned:
-    /// there is nothing left to count against.
-    pub fn record_dispatch_failure(&self, message_key: [u8; 32], retire_at: u32) -> DbResult<bool> {
+    /// No pending record gives [`DispatchFailure::Absent`], not a retirement:
+    /// the ordinary shape of a delivery that settled and then failed a later
+    /// attempt.
+    pub fn record_dispatch_failure(
+        &self,
+        message_key: [u8; 32],
+        retire_at: u32,
+        origin: DispatchOrigin,
+    ) -> DbResult<DispatchFailure> {
         let _pending = self.lock_pending_records();
         let mut records = self.get_pending_cross_zone_dispatches()?;
         let Some(position) = records
             .iter()
             .position(|record| record.message_key == message_key)
         else {
-            return Ok(true);
+            return Ok(DispatchFailure::Absent);
         };
 
-        let attempts = {
+        let failed_attempts = {
             let record = &mut records[position];
             record.failed_attempts = record.failed_attempts.saturating_add(1);
             record.failed_attempts
         };
-        let retired = attempts >= retire_at;
-        if retired {
-            records.remove(position);
+        if failed_attempts < retire_at {
+            self.put_pending_cross_zone_dispatches(&records)?;
+            return Ok(DispatchFailure::Retried { failed_attempts });
         }
-        self.put_pending_cross_zone_dispatches(&records)?;
-        Ok(retired)
+
+        let retired = records.remove(position);
+        let dead_letter = DeadLetterDispatchRecord {
+            message_key,
+            origin,
+            failed_attempts,
+            transaction_bytes: u32::try_from(retired.transaction.len()).unwrap_or(u32::MAX),
+        };
+
+        // One entry per delivery, not per retirement. A watcher rebuilding a
+        // peer tip re-reads from the peer's genesis, and a never-executing
+        // delivery never reaches the seen-set, so the same one retires again;
+        // undeduped it would evict every other entry with copies of itself.
+        let mut dead_letters = self.get_dead_letter_cross_zone_dispatches()?;
+        if !dead_letters
+            .iter()
+            .any(|record| record.message_key == message_key)
+        {
+            dead_letters.push(dead_letter.clone());
+            while dead_letters.len() > MAX_DEAD_LETTER_CROSS_ZONE_DISPATCHES {
+                dead_letters.remove(0);
+            }
+        }
+        // Counted per retirement even so: the retained list evicts and drops
+        // settled entries, so its length is not how often this node gave up.
+        let count = self
+            .get_dead_letter_cross_zone_dispatch_count()?
+            .saturating_add(1);
+
+        // One batch: a crash between the two halves either loses the message
+        // silently or leaves the drain retrying a delivery already recorded as
+        // given up on.
+        let mut batch = WriteBatch::default();
+        self.put_pending_cross_zone_dispatches_batch(&records, &mut batch)?;
+        self.put_batch(
+            &DeadLetterCrossZoneDispatchesCellRef(&dead_letters),
+            (),
+            &mut batch,
+        )?;
+        self.put_batch(&DeadLetterCrossZoneDispatchCountCell(count), (), &mut batch)?;
+        self.db.write(batch).map_err(|rerr| {
+            DbError::rocksdb_cast_message(
+                rerr,
+                Some("Failed to retire a cross-zone dispatch into the dead letter".to_owned()),
+            )
+        })?;
+
+        Ok(DispatchFailure::Retired(Box::new(dead_letter)))
+    }
+
+    /// The cross-zone deliveries given up on and still retained, oldest first.
+    pub fn get_dead_letter_cross_zone_dispatches(&self) -> DbResult<Vec<DeadLetterDispatchRecord>> {
+        Ok(self
+            .get_opt::<DeadLetterCrossZoneDispatchesCellOwned>(())?
+            .map_or_else(Vec::new, |cell| cell.0))
+    }
+
+    /// Every cross-zone delivery given up on, including ones since evicted.
+    pub fn get_dead_letter_cross_zone_dispatch_count(&self) -> DbResult<u64> {
+        Ok(self
+            .get_opt::<DeadLetterCrossZoneDispatchCountCell>(())?
+            .map_or(0, |cell| cell.0))
     }
 
     /// Drops the records of deliveries that are settled for good, outside any
@@ -796,12 +894,52 @@ impl RocksDBIO {
         records.retain(|record| !to_remove.contains(&record.message_key));
         let removed = before.saturating_sub(records.len());
 
+        // Both lists in one batch, as in `record_dispatch_failure`: nothing
+        // recomputes these keys on a later pass to fix a torn write.
+        let mut batch = WriteBatch::default();
         if removed > 0 {
-            self.put_pending_cross_zone_dispatches(&records)?;
+            self.put_pending_cross_zone_dispatches_batch(&records, &mut batch)?;
+        }
+        self.stage_reconciled_dead_letters(&to_remove, &mut batch)?;
+        if !batch.is_empty() {
+            self.db.write(batch).map_err(|rerr| {
+                DbError::rocksdb_cast_message(
+                    rerr,
+                    Some("Failed to drop settled cross-zone dispatches".to_owned()),
+                )
+            })?;
         }
         Ok(removed)
     }
 
+    /// Stages the removal of dead letters whose delivery turned out to settle.
+    ///
+    /// Every sequencer gives up alone, against its own head, so a delivery this
+    /// one abandoned can still reach another's block. Nothing else removes an
+    /// entry, so without this it reports as abandoned for the store's lifetime.
+    ///
+    /// The count is deliberately not decremented: it records how often this node
+    /// gave up, which stays true.
+    fn stage_reconciled_dead_letters(
+        &self,
+        settled: &std::collections::HashSet<&[u8; 32]>,
+        batch: &mut WriteBatch,
+    ) -> DbResult<usize> {
+        let mut dead_letters = self.get_dead_letter_cross_zone_dispatches()?;
+        let before = dead_letters.len();
+        dead_letters.retain(|record| !settled.contains(&record.message_key));
+        let reconciled = before.saturating_sub(dead_letters.len());
+
+        if reconciled > 0 {
+            self.put_batch(
+                &DeadLetterCrossZoneDispatchesCellRef(&dead_letters),
+                (),
+                batch,
+            )?;
+        }
+        Ok(reconciled)
+    }
+
     /// Drops the pending records of deliveries that just became irreversible,
     /// staged into `batch` so they go with the update that made them so.
     ///
@@ -827,6 +965,10 @@ impl RocksDBIO {
         if removed > 0 {
             self.put_pending_cross_zone_dispatches_batch(&records, batch)?;
         }
+
+        // The ordinary case: another sequencer carried a delivery this node gave
+        // up on into a block that just became irreversible.
+        self.stage_reconciled_dead_letters(&to_remove, batch)?;
         Ok(removed)
     }
 

lez/storage/src/sequencer/sequencer_cells.rs

@@ -9,10 +9,12 @@ use crate::{
     sequencer::{
         CF_LEE_STATE_NAME, DB_FINAL_BLOCK_META_KEY, DB_FINAL_LEE_STATE_KEY, DB_LEE_STATE_KEY,
         DB_META_CROSS_ZONE_PEER_FLOOR_KEY, DB_META_CROSS_ZONE_PEER_TIP_KEY,
-        DB_META_LAST_FINALIZED_BLOCK_ID, DB_META_LATEST_BLOCK_META_KEY,
-        DB_META_PENDING_CROSS_ZONE_DISPATCHES_KEY, DB_META_PENDING_DEPOSIT_EVENTS_KEY,
-        DB_META_PUBLISHED_HIGH_WATER_KEY, DB_META_UNSEEN_WITHDRAW_COUNT_KEY,
-        DB_META_ZONE_CURSOR_KEY, DB_META_ZONE_SDK_CHECKPOINT_KEY,
+        DB_META_DEAD_LETTER_CROSS_ZONE_DISPATCH_COUNT_KEY,
+        DB_META_DEAD_LETTER_CROSS_ZONE_DISPATCHES_KEY, DB_META_LAST_FINALIZED_BLOCK_ID,
+        DB_META_LATEST_BLOCK_META_KEY, DB_META_PENDING_CROSS_ZONE_DISPATCHES_KEY,
+        DB_META_PENDING_DEPOSIT_EVENTS_KEY, DB_META_PUBLISHED_HIGH_WATER_KEY,
+        DB_META_UNSEEN_WITHDRAW_COUNT_KEY, DB_META_ZONE_CURSOR_KEY,
+        DB_META_ZONE_SDK_CHECKPOINT_KEY,
     },
 };
 
@@ -294,9 +296,9 @@ pub struct PendingCrossZoneDispatchRecord {
     /// A dispatch's payload and target accounts are chosen on the peer zone and
     /// validated by nobody in between, so one can fail for good. A failure can
     /// equally be a property of the moment, so a single one is not enough to
-    /// give up on a delivery. Once too many accumulate the record is dropped
-    /// rather than flagged, since a delivery nothing will retry is also a
-    /// delivery nothing would ever remove.
+    /// give up on a delivery. Once too many accumulate the record leaves this
+    /// list (the drain re-feeds it every turn) for a
+    /// [`DeadLetterDispatchRecord`], which keeps the delivery identifiable.
     pub failed_attempts: u32,
 }
 
@@ -347,6 +349,100 @@ impl SimpleWritableCell for PendingCrossZoneDispatchesCellRef<'_> {
     }
 }
 
+/// Which peer message a delivery carried, kept so a lost one can be traced back
+/// to the peer block it was in.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, BorshSerialize, BorshDeserialize)]
+pub struct DispatchOrigin {
+    pub src_zone: PeerZoneKey,
+    pub src_block_id: u64,
+    pub src_tx_index: u32,
+}
+
+/// A cross-zone delivery this node has given up on.
+///
+/// A dispatch that fails execution is left out of the block, so nothing on chain
+/// records that it was attempted; this is the only durable trace.
+///
+/// It identifies the message rather than carrying it: the peer block and index
+/// are enough to read it back off the channel, and the encoded transaction is
+/// peer-chosen and can exceed a whole block, which would leave the list bounded
+/// in entries but unbounded in bytes.
+///
+/// Giving up is this node's decision, not the network's, so an entry is dropped
+/// again if another sequencer carries the same delivery.
+#[derive(Debug, Clone, PartialEq, Eq, BorshSerialize, BorshDeserialize)]
+pub struct DeadLetterDispatchRecord {
+    pub message_key: [u8; 32],
+    pub origin: DispatchOrigin,
+    /// Attempts made before giving up, so the record carries the policy that was
+    /// in force at the time.
+    pub failed_attempts: u32,
+    /// Size of the delivery transaction that would not execute, the diagnostic
+    /// for size-related failures.
+    pub transaction_bytes: u32,
+}
+
+#[derive(BorshDeserialize)]
+pub struct DeadLetterCrossZoneDispatchesCellOwned(pub Vec<DeadLetterDispatchRecord>);
+
+impl SimpleStorableCell for DeadLetterCrossZoneDispatchesCellOwned {
+    type KeyParams = ();
+
+    const CELL_NAME: &'static str = DB_META_DEAD_LETTER_CROSS_ZONE_DISPATCHES_KEY;
+    const CF_NAME: &'static str = CF_META_NAME;
+}
+
+impl SimpleReadableCell for DeadLetterCrossZoneDispatchesCellOwned {}
+
+#[derive(BorshSerialize)]
+pub struct DeadLetterCrossZoneDispatchesCellRef<'records>(pub &'records [DeadLetterDispatchRecord]);
+
+impl SimpleStorableCell for DeadLetterCrossZoneDispatchesCellRef<'_> {
+    type KeyParams = ();
+
+    const CELL_NAME: &'static str = DB_META_DEAD_LETTER_CROSS_ZONE_DISPATCHES_KEY;
+    const CF_NAME: &'static str = CF_META_NAME;
+}
+
+impl SimpleWritableCell for DeadLetterCrossZoneDispatchesCellRef<'_> {
+    fn value_constructor(&self) -> DbResult<Vec<u8>> {
+        borsh::to_vec(&self).map_err(|err| {
+            DbError::borsh_cast_message(
+                err,
+                Some("Failed to serialize dead-letter cross-zone dispatches cell".to_owned()),
+            )
+        })
+    }
+}
+
+/// Deliveries given up on since this store was created.
+///
+/// Separate from the retained list, which evicts at its cap and drops settled
+/// entries: a node that gave up hundreds of times would otherwise look like one
+/// that gave up at the cap.
+#[derive(BorshSerialize, BorshDeserialize)]
+pub struct DeadLetterCrossZoneDispatchCountCell(pub u64);
+
+impl SimpleStorableCell for DeadLetterCrossZoneDispatchCountCell {
+    type KeyParams = ();
+
+    const CELL_NAME: &'static str = DB_META_DEAD_LETTER_CROSS_ZONE_DISPATCH_COUNT_KEY;
+    const CF_NAME: &'static str = CF_META_NAME;
+}
+
+impl SimpleReadableCell for DeadLetterCrossZoneDispatchCountCell {}
+
+impl SimpleWritableCell for DeadLetterCrossZoneDispatchCountCell {
+    fn value_constructor(&self) -> DbResult<Vec<u8>> {
+        borsh::to_vec(&self).map_err(|err| {
+            DbError::borsh_cast_message(
+                err,
+                Some("Failed to serialize dead-letter cross-zone dispatch count".to_owned()),
+            )
+        })
+    }
+}
+
 #[derive(BorshDeserialize)]
 pub struct PendingDepositEventsCellOwned(pub Vec<PendingDepositEventRecord>);
 

lez/storage/src/sequencer/tests.rs

@@ -41,6 +41,15 @@ fn dispatch_record(seed: u8) -> PendingCrossZoneDispatchRecord {
     PendingCrossZoneDispatchRecord::recorded([seed; 32], vec![seed; 4])
 }
 
+/// The peer coordinates a dead letter carries, distinct per seed.
+fn dispatch_origin(seed: u8) -> DispatchOrigin {
+    DispatchOrigin {
+        src_zone: [seed; 32],
+        src_block_id: u64::from(seed),
+        src_tx_index: u32::from(seed),
+    }
+}
+
 /// A distinct message key per index, for filling the pending list.
 fn key_from_index(index: usize) -> [u8; 32] {
     let mut key = [0_u8; 32];
@@ -606,7 +615,7 @@ fn finalized_dispatch_records_are_removed_by_message_key() {
 }
 
 #[test]
-fn record_dispatch_failure_drops_the_record_at_the_limit() {
+fn record_dispatch_failure_retires_the_record_at_the_limit() {
     let temp_dir = tempdir().unwrap();
     let (dbio, _genesis) = dbio_with_genesis(temp_dir.path());
 
@@ -616,36 +625,229 @@ fn record_dispatch_failure_drops_the_record_at_the_limit() {
     dbio.add_pending_cross_zone_dispatches(vec![record, survivor.clone()])
         .unwrap();
 
-    assert!(!dbio.record_dispatch_failure(key, 3).unwrap());
     assert_eq!(
-        dbio.get_pending_cross_zone_dispatches().unwrap()[0].failed_attempts,
-        1,
+        dbio.record_dispatch_failure(key, 3, dispatch_origin(1))
+            .unwrap(),
+        DispatchFailure::Retried { failed_attempts: 1 },
         "a failure short of the limit is counted, not given up on"
     );
-    assert!(!dbio.record_dispatch_failure(key, 3).unwrap());
-    assert!(
-        dbio.record_dispatch_failure(key, 3).unwrap(),
-        "the third failure is the one it is given up on"
+    assert_eq!(
+        dbio.get_pending_cross_zone_dispatches().unwrap()[0].failed_attempts,
+        1
     );
+    assert_eq!(
+        dbio.record_dispatch_failure(key, 3, dispatch_origin(1))
+            .unwrap(),
+        DispatchFailure::Retried { failed_attempts: 2 }
+    );
+    let DispatchFailure::Retired(retired) = dbio
+        .record_dispatch_failure(key, 3, dispatch_origin(1))
+        .unwrap()
+    else {
+        panic!("the third failure is the one it is given up on");
+    };
+    assert_eq!(retired.message_key, key);
+    assert_eq!(retired.origin, dispatch_origin(1));
+    assert_eq!(retired.failed_attempts, 3);
 
-    // Dropped rather than flagged: a delivery the drain will never feed into a
-    // block again is one nothing would ever remove, so flagging it would let a
-    // peer that can make deliveries fail grow the list without bound.
+    // It has to leave the pending list, which the drain re-feeds every turn, or
+    // a delivery that can never execute would be retried for ever.
     assert_eq!(
         dbio.get_pending_cross_zone_dispatches().unwrap(),
         vec![survivor],
-        "giving up on a delivery drops its record and leaves the others alone"
+        "giving up on a delivery takes its record out and leaves the others alone"
     );
 
-    // A key with no record reads as given up on: there is nothing left to count
-    // against, and nothing will feed it into a block.
-    assert!(
-        dbio.record_dispatch_failure(key, 3).unwrap(),
-        "a failure against a dropped delivery must not re-create its record"
+    // A key with no record is not a give-up: nothing was counted and nothing was
+    // abandoned. This is the shape of a delivery that settled and then failed a
+    // later attempt.
+    assert_eq!(
+        dbio.record_dispatch_failure(key, 3, dispatch_origin(1))
+            .unwrap(),
+        DispatchFailure::Absent,
+        "a failure against a retired delivery must not re-create its record"
     );
     assert_eq!(dbio.get_pending_cross_zone_dispatches().unwrap().len(), 1);
 }
 
+#[test]
+fn a_retired_dispatch_moves_into_the_dead_letter_identified_by_its_origin() {
+    let temp_dir = tempdir().unwrap();
+    let (dbio, _genesis) = dbio_with_genesis(temp_dir.path());
+
+    let record = dispatch_record(7);
+    let key = record.message_key;
+    let encoded_len = u32::try_from(record.transaction.len()).unwrap();
+    dbio.add_pending_cross_zone_dispatches(vec![record])
+        .unwrap();
+
+    assert!(
+        dbio.get_dead_letter_cross_zone_dispatches()
+            .unwrap()
+            .is_empty()
+    );
+    for _ in 0..3 {
+        dbio.record_dispatch_failure(key, 3, dispatch_origin(7))
+            .unwrap();
+    }
+
+    let dead_letters = dbio.get_dead_letter_cross_zone_dispatches().unwrap();
+    assert_eq!(dead_letters.len(), 1);
+    assert_eq!(dead_letters[0].message_key, key);
+    assert_eq!(
+        dead_letters[0].origin,
+        dispatch_origin(7),
+        "the peer coordinates are what let the message be read back off the peer channel"
+    );
+    assert_eq!(dead_letters[0].transaction_bytes, encoded_len);
+    assert_eq!(dbio.get_dead_letter_cross_zone_dispatch_count().unwrap(), 1);
+}
+
+#[test]
+fn a_dead_letter_is_dropped_once_its_delivery_settles_elsewhere() {
+    let temp_dir = tempdir().unwrap();
+    let (dbio, _genesis) = dbio_with_genesis(temp_dir.path());
+
+    let record = dispatch_record(7);
+    let key = record.message_key;
+    dbio.add_pending_cross_zone_dispatches(vec![record])
+        .unwrap();
+    dbio.record_dispatch_failure(key, 1, dispatch_origin(7))
+        .unwrap();
+    assert_eq!(
+        dbio.get_dead_letter_cross_zone_dispatches().unwrap().len(),
+        1
+    );
+
+    // A delivery this node gave up on can still reach another sequencer's block.
+    dbio.drop_settled_cross_zone_dispatches(&[key]).unwrap();
+    assert!(
+        dbio.get_dead_letter_cross_zone_dispatches()
+            .unwrap()
+            .is_empty()
+    );
+
+    // The count is how often this node gave up, which stays true whatever
+    // happened next, and is what keeps the list readable as "still outstanding".
+    assert_eq!(dbio.get_dead_letter_cross_zone_dispatch_count().unwrap(), 1);
+}
+
+#[test]
+fn a_dead_letter_is_dropped_by_the_settlement_path_inside_a_store_update() {
+    let temp_dir = tempdir().unwrap();
+    let (dbio, genesis) = dbio_with_genesis(temp_dir.path());
+
+    let record = dispatch_record(7);
+    let key = record.message_key;
+    dbio.add_pending_cross_zone_dispatches(vec![record])
+        .unwrap();
+    dbio.record_dispatch_failure(key, 1, dispatch_origin(7))
+        .unwrap();
+    assert_eq!(
+        dbio.get_dead_letter_cross_zone_dispatches().unwrap().len(),
+        1
+    );
+
+    // The ordinary route, unlike the standalone drop: a block carrying the
+    // delivery becomes irreversible and the update that records that also
+    // reconciles the dead letter, in the same batch.
+    let block2 = produce_dummy_block(2, Some(genesis.header.hash), vec![]);
+    dbio.store_update(&StoreUpdate {
+        blocks: &[(&block2, true)],
+        remove_dispatch_records: &[key],
+        ..StoreUpdate::new(&state_with_balance(200))
+    })
+    .unwrap();
+
+    assert!(
+        dbio.get_dead_letter_cross_zone_dispatches()
+            .unwrap()
+            .is_empty()
+    );
+    assert_eq!(dbio.get_dead_letter_cross_zone_dispatch_count().unwrap(), 1);
+}
+
+#[test]
+fn one_delivery_that_always_fails_takes_one_dead_letter_slot() {
+    let temp_dir = tempdir().unwrap();
+    let (dbio, _genesis) = dbio_with_genesis(temp_dir.path());
+
+    // A watcher rebuilding a peer tip re-reads from genesis, so the same
+    // never-executing delivery retires repeatedly (see `record_dispatch_failure`).
+    let key = key_from_index(1);
+    let other = key_from_index(2);
+    dbio.add_pending_cross_zone_dispatches(vec![PendingCrossZoneDispatchRecord::recorded(
+        other,
+        vec![1, 2, 3, 4],
+    )])
+    .unwrap();
+    dbio.record_dispatch_failure(other, 1, dispatch_origin(2))
+        .unwrap();
+
+    for _ in 0..5 {
+        dbio.add_pending_cross_zone_dispatches(vec![PendingCrossZoneDispatchRecord::recorded(
+            key,
+            vec![1, 2, 3, 4],
+        )])
+        .unwrap();
+        dbio.record_dispatch_failure(key, 1, dispatch_origin(1))
+            .unwrap();
+    }
+
+    let dead_letters = dbio.get_dead_letter_cross_zone_dispatches().unwrap();
+    assert_eq!(
+        dead_letters.len(),
+        2,
+        "one entry per delivery, not per retirement"
+    );
+    assert_eq!(
+        dead_letters[0].message_key, other,
+        "the other message is not evicted"
+    );
+
+    // The count still measures give-ups, so the repetition remains visible.
+    assert_eq!(dbio.get_dead_letter_cross_zone_dispatch_count().unwrap(), 6);
+}
+
+#[test]
+fn dead_letters_evict_the_oldest_at_the_cap_but_keep_counting() {
+    let temp_dir = tempdir().unwrap();
+    let (dbio, _genesis) = dbio_with_genesis(temp_dir.path());
+
+    let retirements = MAX_DEAD_LETTER_CROSS_ZONE_DISPATCHES + 3;
+    for index in 0..retirements {
+        let key = key_from_index(index);
+        dbio.add_pending_cross_zone_dispatches(vec![PendingCrossZoneDispatchRecord::recorded(
+            key,
+            vec![1, 2, 3, 4],
+        )])
+        .unwrap();
+        dbio.record_dispatch_failure(key, 1, dispatch_origin(1))
+            .unwrap();
+    }
+
+    let dead_letters = dbio.get_dead_letter_cross_zone_dispatches().unwrap();
+    assert_eq!(dead_letters.len(), MAX_DEAD_LETTER_CROSS_ZONE_DISPATCHES);
+    assert_eq!(
+        dead_letters[0].message_key,
+        key_from_index(3),
+        "the oldest retained entry is the fourth retirement, the first three having been evicted"
+    );
+    assert_eq!(
+        dead_letters[dead_letters.len() - 1].message_key,
+        key_from_index(retirements - 1),
+        "the newest retirement is kept"
+    );
+
+    // What eviction must not do is hide that the evicted ones happened: a node
+    // that lost hundreds of messages would otherwise look like one that lost the
+    // cap.
+    assert_eq!(
+        dbio.get_dead_letter_cross_zone_dispatch_count().unwrap(),
+        u64::try_from(retirements).unwrap()
+    );
+}
+
 #[test]
 fn repeated_withdrawal_key_in_one_update_folds_once_per_occurrence() {
     let temp_dir = tempdir().unwrap();

monitoring/grafana/dashboards/sequencer.json

@@ -346,6 +346,86 @@
       ],
       "title": "Submitted vs failed transactions (per minute)",
       "type": "timeseries"
+    },
+    {
+      "datasource": { "type": "prometheus", "uid": "prometheus" },
+      "fieldConfig": {
+        "defaults": { "color": { "mode": "fixed", "fixedColor": "red" }, "unit": "short", "decimals": 0 },
+        "overrides": [ ]
+      },
+      "gridPos": { "h": 7, "w": 6, "x": 0, "y": 41 },
+      "id": 11,
+      "options": {
+        "colorMode": "value",
+        "graphMode": "area",
+        "reduceOptions": { "calcs": [ "lastNotNull" ], "fields": "", "values": false }
+      },
+      "targets": [
+        {
+          "datasource": { "type": "prometheus", "uid": "prometheus" },
+          "expr": "cross_zone_dispatches_retired_total",
+          "legendFormat": "given up on",
+          "refId": "A"
+        }
+      ],
+      "title": "Cross-zone deliveries given up on since startup",
+      "type": "stat"
+    },
+    {
+      "datasource": { "type": "prometheus", "uid": "prometheus" },
+      "fieldConfig": {
+        "defaults": { "color": { "mode": "fixed", "fixedColor": "orange" }, "unit": "short", "decimals": 0 },
+        "overrides": [ ]
+      },
+      "gridPos": { "h": 7, "w": 6, "x": 6, "y": 41 },
+      "id": 12,
+      "options": {
+        "colorMode": "value",
+        "graphMode": "area",
+        "reduceOptions": { "calcs": [ "lastNotNull" ], "fields": "", "values": false }
+      },
+      "targets": [
+        {
+          "datasource": { "type": "prometheus", "uid": "prometheus" },
+          "expr": "cross_zone_dead_letter_dispatches",
+          "legendFormat": "retained",
+          "refId": "A"
+        }
+      ],
+      "title": "Dead letters retained",
+      "type": "stat"
+    },
+    {
+      "datasource": { "type": "prometheus", "uid": "prometheus" },
+      "fieldConfig": {
+        "defaults": {
+          "custom": { "drawStyle": "line", "lineWidth": 1, "fillOpacity": 10 },
+          "unit": "short",
+          "min": 0.0
+        },
+        "overrides": [
+          {
+            "matcher": { "id": "byName", "options": "given up on" },
+            "properties": [ { "id": "color", "value": { "mode": "fixed", "fixedColor": "red" } } ]
+          }
+        ]
+      },
+      "gridPos": { "h": 7, "w": 12, "x": 12, "y": 41 },
+      "id": 13,
+      "options": {
+        "legend": { "displayMode": "list", "placement": "bottom", "calcs": [ "last", "max" ] },
+        "tooltip": { "mode": "single" }
+      },
+      "targets": [
+        {
+          "datasource": { "type": "prometheus", "uid": "prometheus" },
+          "expr": "rate(cross_zone_dispatches_retired_total[$__rate_interval]) * 60",
+          "legendFormat": "given up on",
+          "refId": "A"
+        }
+      ],
+      "title": "Cross-zone deliveries given up on (per minute)",
+      "type": "timeseries"
     }
   ],
   "refresh": "5s",

test_fixtures/src/bin/regenerate_test_fixture.rs

@@ -76,7 +76,9 @@ async fn generate_prebuilt_fixture(dest: &Path) -> Result<()> {
     drop(wallet);
     drop(sequencer_handle);
 
-    let db_path = temp_sequencer_dir.path().join("rocksdb");
+    let db_path = temp_sequencer_dir
+        .path()
+        .join(format!("rocksdb-{}", config::bedrock_channel_id()));
     let store = open_store_with_retry(&db_path)
         .await
         .context("Failed to reopen sequencer store after shutdown")?;

test_fixtures/src/setup.rs

@@ -118,8 +118,9 @@ impl SequencerSetup {
             genesis
         } else {
             let dump = load_prebuilt_dump()?;
-            // `SequencerCore::open_or_create_store` looks for `<home>/rocksdb`.
-            let dst = home.join("rocksdb");
+            // `SequencerCore::open_or_create_store` looks for the channel-suffixed
+            // db under its home, so the restore has to land on the same name.
+            let dst = home.join(format!("rocksdb-{channel_id}"));
             let _store = SequencerStore::restore_db_from_dump(
                 &dst,
                 &dump,

tools/dashboard_gen/src/dashboards/sequencer.rs

@@ -192,4 +192,44 @@ pub fn dashboard() -> Dashboard {
                     ),
             ],
         )
+        .row(
+            7,
+            [
+                // A failed dispatch is left out of the block, so nothing on chain
+                // records it. These panels are the only signal.
+                Panel::stat("Cross-zone deliveries given up on since startup")
+                    .width(6)
+                    .unit(Unit::Short)
+                    .decimals(0)
+                    .color(Color::fixed("red"))
+                    .target(
+                        Target::new(
+                            sequencer_core_metrics::names::CROSS_ZONE_DISPATCHES_RETIRED_TOTAL,
+                        )
+                        .legend("given up on"),
+                    ),
+                Panel::stat("Dead letters retained")
+                    .width(6)
+                    .unit(Unit::Short)
+                    .decimals(0)
+                    .color(Color::fixed("orange"))
+                    .target(
+                        Target::new(
+                            sequencer_core_metrics::names::CROSS_ZONE_DEAD_LETTER_DISPATCHES,
+                        )
+                        .legend("retained"),
+                    ),
+                Panel::timeseries("Cross-zone deliveries given up on (per minute)")
+                    .width(12)
+                    .unit(Unit::Short)
+                    .min(0.0)
+                    .target(rate_per_min(
+                        sequencer_core_metrics::names::CROSS_ZONE_DISPATCHES_RETIRED_TOTAL,
+                        "given up on",
+                    ))
+                    .with_override(
+                        FieldOverride::by_name("given up on").color(Color::fixed("red")),
+                    ),
+            ],
+        )
 }