diff --git a/Cargo.lock b/Cargo.lock index 73156027f..d67437d7c 100644 --- a/Cargo.lock +++ b/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", ] diff --git a/Justfile b/Justfile index 4741f6c84..607b92817 100644 --- a/Justfile +++ b/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 diff --git a/README.md b/README.md index 401fff157..79a624ba9 100644 --- a/README.md +++ b/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-` 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-` ### 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. diff --git a/artifacts/lez/programs/bridge_lock.bin b/artifacts/lez/programs/bridge_lock.bin index f141b9aa6..04626a0dd 100644 Binary files a/artifacts/lez/programs/bridge_lock.bin and b/artifacts/lez/programs/bridge_lock.bin differ diff --git a/artifacts/lez/programs/cross_zone_inbox.bin b/artifacts/lez/programs/cross_zone_inbox.bin index c2f18db31..2d69904d3 100644 Binary files a/artifacts/lez/programs/cross_zone_inbox.bin and b/artifacts/lez/programs/cross_zone_inbox.bin differ diff --git a/artifacts/lez/programs/wrapped_token.bin b/artifacts/lez/programs/wrapped_token.bin index 930f44277..416cb2b5b 100644 Binary files a/artifacts/lez/programs/wrapped_token.bin and b/artifacts/lez/programs/wrapped_token.bin differ diff --git a/build_utils/src/lib.rs b/build_utils/src/lib.rs index 1323d830e..6753e845d 100644 --- a/build_utils/src/lib.rs +++ b/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()); diff --git a/integration_tests/Cargo.toml b/integration_tests/Cargo.toml index 225d75d4d..f365e8561 100644 --- a/integration_tests/Cargo.toml +++ b/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 diff --git a/integration_tests/tests/cross_zone_ingress_guard.rs b/integration_tests/tests/cross_zone_ingress_guard.rs index 86731477f..1da931379 100644 --- a/integration_tests/tests/cross_zone_ingress_guard.rs +++ b/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(), diff --git a/integration_tests/tests/cross_zone_state_machine.rs b/integration_tests/tests/cross_zone_state_machine.rs index f9d92a312..119d09cb2 100644 --- a/integration_tests/tests/cross_zone_state_machine.rs +++ b/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 { + mint_payload_of(LOCK_AMOUNT) +} + +fn mint_payload_of(amount: u128) -> Vec { 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 { + 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 = 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" + ); +} diff --git a/integration_tests/tests/sequencer_bootstrap.rs b/integration_tests/tests/sequencer_bootstrap.rs index c4468c1b5..dc6d72739 100644 --- a/integration_tests/tests/sequencer_bootstrap.rs +++ b/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. { diff --git a/lez/cross_zone/src/lib.rs b/lez/cross_zone/src/lib.rs index 38f35d24c..7ea3fdf0f 100644 --- a/lez/cross_zone/src/lib.rs +++ b/lez/cross_zone/src/lib.rs @@ -31,6 +31,21 @@ pub struct Emission { pub payload: Vec, } +/// 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, ) -> 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, diff --git a/lez/indexer/core/src/cross_zone_verifier.rs b/lez/indexer/core/src/cross_zone_verifier.rs index 9dcfe2eb5..1cc48ee17 100644 --- a/lez/indexer/core/src/cross_zone_verifier.rs +++ b/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, peers: PeerBlocks, - seen: Arc>>, + seen: Arc>>, } 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, CrossZoneVerifyError> { + pub async fn verify_block(&self, block: &Block) -> Result, 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) { + pub async fn record_seen(&self, keys: Vec) { 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, } +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) diff --git a/lez/programs/bridge_lock/src/main.rs b/lez/programs/bridge_lock/src/main.rs index 8b176ee56..ec4ac7b4a 100644 --- a/lez/programs/bridge_lock/src/main.rs +++ b/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) diff --git a/lez/programs/cross_zone_inbox/core/src/lib.rs b/lez/programs/cross_zone_inbox/core/src/lib.rs index b3ea8d596..c0dcf64b2 100644 --- a/lez/programs/cross_zone_inbox/core/src/lib.rs +++ b/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); +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, +} 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 { 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" ); } } diff --git a/lez/programs/cross_zone_inbox/src/main.rs b/lez/programs/cross_zone_inbox/src/main.rs index 4dec4478b..4d2181c23 100644 --- a/lez/programs/cross_zone_inbox/src/main.rs +++ b/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() diff --git a/lez/programs/wrapped_token/core/src/lib.rs b/lez/programs/wrapped_token/core/src/lib.rs index a95a5ca0a..2d5e0d775 100644 --- a/lez/programs/wrapped_token/core/src/lib.rs +++ b/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"; diff --git a/lez/programs/wrapped_token/src/main.rs b/lez/programs/wrapped_token/src/main.rs index 19095e393..88cb17887 100644 --- a/lez/programs/wrapped_token/src/main.rs +++ b/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"); diff --git a/lez/sequencer/core/metrics/src/names.rs b/lez/sequencer/core/metrics/src/names.rs index 92b1595eb..980f2c8ce 100644 --- a/lez/sequencer/core/metrics/src/names.rs +++ b/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"; diff --git a/lez/sequencer/core/metrics/src/record.rs b/lez/sequencer/core/metrics/src/record.rs index 96b3304ed..e6f7e8d33 100644 --- a/lez/sequencer/core/metrics/src/record.rs +++ b/lez/sequencer/core/metrics/src/record.rs @@ -48,6 +48,8 @@ impl From 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); +} diff --git a/lez/sequencer/core/src/config.rs b/lez/sequencer/core/src/config.rs index bce4a2624..a1fe50823 100644 --- a/lez/sequencer/core/src/config.rs +++ b/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 { diff --git a/lez/sequencer/core/src/cross_zone_watcher.rs b/lez/sequencer/core/src/cross_zone_watcher.rs index e21641a6a..0ddc4b6d3 100644 --- a/lez/sequencer/core/src/cross_zone_watcher.rs +++ b/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 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::(&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(); diff --git a/lez/sequencer/core/src/lib.rs b/lez/sequencer/core/src/lib.rs index 93de58b05..81a9cc268 100644 --- a/lez/sequencer/core/src/lib.rs +++ b/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 SequencerCore { /// 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 SequencerCore { .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 SequencerCore { }; 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 SequencerCore { (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 SequencerCore { /// 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 SequencerCore { 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 SequencerCore { } } + /// 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), 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!( diff --git a/lez/sequencer/core/src/tests.rs b/lez/sequencer/core/src/tests.rs index ae0a6c31b..ceef2d4ae 100644 --- a/lez/sequencer/core/src/tests.rs +++ b/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 { fn dispatch_tx(src_block_id: u64, payload: Vec) -> 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) -> 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(); diff --git a/lez/sequencer/service/protocol/Cargo.toml b/lez/sequencer/service/protocol/Cargo.toml index ced19e755..1eb413d0b 100644 --- a/lez/sequencer/service/protocol/Cargo.toml +++ b/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 diff --git a/lez/sequencer/service/protocol/src/lib.rs b/lez/sequencer/service/protocol/src/lib.rs index ce669d312..37f70415b 100644 --- a/lez/sequencer/service/protocol/src/lib.rs +++ b/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, +} + impl Display for ChannelId { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { let hex_string = hex::encode(self.0); diff --git a/lez/sequencer/service/rpc/src/lib.rs b/lez/sequencer/service/rpc/src/lib.rs index a1d2acfb4..6af9cb807 100644 --- a/lez/sequencer/service/rpc/src/lib.rs +++ b/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; + + /// 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; } diff --git a/lez/sequencer/service/src/service.rs b/lez/sequencer/service/src/service.rs index e55735c07..6afe696be 100644 --- a/lez/sequencer/service/src/service.rs +++ b/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 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 { + 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 { diff --git a/lez/storage/src/sequencer/mod.rs b/lez/storage/src/sequencer/mod.rs index c771ea556..90f074145 100644 --- a/lez/storage/src/sequencer/mod.rs +++ b/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), + /// 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 { + /// 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 { 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> { + Ok(self + .get_opt::(())? + .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 { + Ok(self + .get_opt::(())? + .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 { + 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) } diff --git a/lez/storage/src/sequencer/sequencer_cells.rs b/lez/storage/src/sequencer/sequencer_cells.rs index 098d561bd..8cbfe012d 100644 --- a/lez/storage/src/sequencer/sequencer_cells.rs +++ b/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); + +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> { + 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> { + 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); diff --git a/lez/storage/src/sequencer/tests.rs b/lez/storage/src/sequencer/tests.rs index 385f7ffec..2d4d2ed34 100644 --- a/lez/storage/src/sequencer/tests.rs +++ b/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(); diff --git a/monitoring/grafana/dashboards/sequencer.json b/monitoring/grafana/dashboards/sequencer.json index ede0ae4bd..afc6c2756 100644 --- a/monitoring/grafana/dashboards/sequencer.json +++ b/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", diff --git a/test_fixtures/fixtures/prebuilt_sequencer_db.dump b/test_fixtures/fixtures/prebuilt_sequencer_db.dump index 05d5b4f40..b5c47d6a1 100644 Binary files a/test_fixtures/fixtures/prebuilt_sequencer_db.dump and b/test_fixtures/fixtures/prebuilt_sequencer_db.dump differ diff --git a/test_fixtures/src/bin/regenerate_test_fixture.rs b/test_fixtures/src/bin/regenerate_test_fixture.rs index 038d4b0cd..434a4ccf5 100644 --- a/test_fixtures/src/bin/regenerate_test_fixture.rs +++ b/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")?; diff --git a/test_fixtures/src/setup.rs b/test_fixtures/src/setup.rs index 5d185a2d7..612f6b8db 100644 --- a/test_fixtures/src/setup.rs +++ b/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 `/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, diff --git a/tools/dashboard_gen/src/dashboards/sequencer.rs b/tools/dashboard_gen/src/dashboards/sequencer.rs index 0520251c0..b75f120d3 100644 --- a/tools/dashboard_gen/src/dashboards/sequencer.rs +++ b/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")), + ), + ], + ) }