logos-blockchain/lssa
1
0
Fork 0
mirror of https://github.com/logos-blockchain/lssa.git synced 2026-08-01 20:43:14 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(cross-zone): record peer deliveries instead of injecting them

Browse Source
This commit is contained in:
moudyellaz 2026-07-31 16:01:41 +02:00
parent 2d201f7bac
commit c14bfe3f9f
7 changed files with 2015 additions and 109 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
Cargo.lock generated
View File

@ -9307,6 +9307,7 @@ dependencies = [
"logos-blockchain-zone-sdk",
"mempool",
"num-bigint 0.4.6",
"ping_core",
"programs",
"rand 0.8.6",
"risc0-zkvm",

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

@ -57,3 +57,4 @@ test_programs.workspace = true
lee = { workspace = true, features = ["test-utils"] }
key_protocol.workspace = true
token_core.workspace = true
ping_core.workspace = true

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

@ -9,10 +9,12 @@ use common::{
use lee::V03State;
use lee_core::BlockId;
use log::info;
use logos_blockchain_zone_sdk::sequencer::SequencerCheckpoint;
use logos_blockchain_zone_sdk::{Slot, sequencer::SequencerCheckpoint};
use storage::sequencer::{
RocksDBIO,
sequencer_cells::{PendingDepositEventRecord, WithdrawalReconciliationKey, ZoneAnchorRecord},
sequencer_cells::{
PeerZoneKey, PendingDepositEventRecord, WithdrawalReconciliationKey, ZoneAnchorRecord,
},
};
pub use storage::{DbResult, sequencer::DbDump};
@ -233,6 +235,36 @@ pub(crate) fn block_to_transactions_map(block: &Block) -> HashMap<HashType, u64>
.collect()
}
/// A cross-zone watcher's delivery floor on `peer_zone`'s channel.
///
/// The highest slot every message of which was delivered, or `None` before it
/// has delivered anything from that peer. Stored as a little-endian `u64`.
///
/// Free functions rather than only [`SequencerStore`] methods because each
/// watcher runs as its own spawned task and holds an `Arc<RocksDBIO>`;
/// `SequencerStore` is not `Clone`.
pub fn get_cross_zone_peer_floor(dbio: &RocksDBIO, peer_zone: PeerZoneKey) -> Result<Option<Slot>> {
let Some(bytes) = dbio.get_cross_zone_peer_floor_bytes(peer_zone)? else {
return Ok(None);
};
let bytes: [u8; 8] = bytes.as_slice().try_into().with_context(|| {
format!(
"Stored cross-zone peer floor is {} bytes, expected 8",
bytes.len()
)
})?;
Ok(Some(Slot::new(u64::from_le_bytes(bytes))))
}
pub fn set_cross_zone_peer_floor(
dbio: &RocksDBIO,
peer_zone: PeerZoneKey,
floor: Slot,
) -> Result<()> {
dbio.put_cross_zone_peer_floor_bytes(peer_zone, &floor.to_le_bytes())?;
Ok(())
}
#[cfg(test)]
mod tests {
use common::{block::HashableBlockData, test_utils::sequencer_sign_key_for_testing};

1022
lez/sequencer/core/src/cross_zone_watcher.rs
View File

File diff suppressed because it is too large Load Diff

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

@ -16,6 +16,7 @@ use common::{
transaction::{LeeTransaction, clock_invocation},
};
use config::{GenesisAction, SequencerConfig};
use cross_zone_inbox_core::CrossZoneMessage;
use futures::StreamExt as _;
use itertools::Itertools as _;
use lee::{AccountId, PublicTransaction, public_transaction::Message};
@ -33,7 +34,10 @@ use num_bigint::BigUint;
pub use storage::error::DbError;
use storage::sequencer::{
RocksDBIO, StoreUpdate,
sequencer_cells::{PendingDepositEventRecord, WithdrawalReconciliationKey, ZoneAnchorRecord},
sequencer_cells::{
PendingCrossZoneDispatchRecord, PendingDepositEventRecord, WithdrawalReconciliationKey,
ZoneAnchorRecord,
},
};
use crate::{
@ -51,6 +55,22 @@ pub mod cross_zone_watcher;
pub mod mock;
pub mod task_group;
/// Failed production attempts before a cross-zone dispatch is given up on.
///
/// One attempt per block, so this is tens of seconds of retrying. Enough for a
/// failure that is not the message's fault to clear, short enough that a message
/// which will never execute stops being retried.
const RETIRE_DISPATCH_AFTER_FAILURES: u32 = 3;
/// Cross-zone deliveries one block may carry.
///
/// Each one costs a guest execution whether it succeeds or fails, and what
/// queues them up is chosen by peer zones. Without a bound, a backlog decides
/// how long a block takes to build and leaves no room for user transactions,
/// since store-drained work is taken before the mempool. The rest wait one
/// block; nothing is dropped.
const MAX_DISPATCHES_PER_BLOCK: usize = 16;
/// The origin of a transaction.
#[derive(Clone, Copy)]
pub enum TransactionOrigin {
@ -73,6 +93,10 @@ pub struct SequencerCore<BP: BlockPublisherTrait = ZoneSdkPublisher> {
mempool: MemPool<(TransactionOrigin, LeeTransaction)>,
sequencer_config: SequencerConfig,
block_publisher: BP,
/// Cross-zone watchers, stopped when this sequencer is dropped. They hold a
/// store handle, so leaving them running would keep the `RocksDB` lock held
/// and make the home directory unopenable by a restarting sequencer.
watchers: TaskGroup,
}
impl<BP: BlockPublisherTrait> SequencerCore<BP> {
@ -204,14 +228,17 @@ impl<BP: BlockPublisherTrait> SequencerCore<BP> {
// Cross-zone messaging: start a watcher per configured peer. The inbox
// config account is seeded into genesis state in `build_genesis_state`.
if let Some(cross_zone) = &config.cross_zone {
cross_zone_watcher::spawn_watchers(
&config.bedrock_config,
cross_zone,
config.block_create_timeout,
&mempool_handle,
);
}
let watchers = config
.cross_zone
.as_ref()
.map_or_else(TaskGroup::default, |cross_zone| {
cross_zone_watcher::spawn_watchers(
&config.bedrock_config,
cross_zone,
config.block_create_timeout,
&store.dbio(),
)
});
// Before producing, verify our local state still belongs to the chain
// the channel serves and replay any channel blocks we are missing
// (e.g. from other sequencers).
@ -269,6 +296,7 @@ impl<BP: BlockPublisherTrait> SequencerCore<BP> {
mempool,
sequencer_config: config,
block_publisher,
watchers,
};
(sequencer_core, mempool_handle)
@ -425,6 +453,12 @@ impl<BP: BlockPublisherTrait> SequencerCore<BP> {
.context("Failed to read stored block")?
{
Some(stored) if stored.header.hash == block_hash => {
// Already applied, but the channel serving it is what makes
// it irreversible, so its deliveries are settled and their
// records are owed nothing. Without this a restart leaves a
// record for every delivery it already published, and
// nothing downstream would ever remove them.
settle_reconstructed_deliveries(store, &stored);
store
.set_zone_anchor(&record)
.context("Failed to persist zone anchor")?;
@ -466,6 +500,9 @@ impl<BP: BlockPublisherTrait> SequencerCore<BP> {
.iter()
.filter_map(extract_bridge_deposit_id)
.collect();
// The same for the deliveries it carries: the inbox has seen them, so
// the drain would skip them anyway, and the records are owed nothing.
let finalized_dispatch_keys = settled_dispatch_keys(&store.dbio(), block);
// The tip meta stays pinned to the head tip even when the reconstructed
// block lands below it, and the anchor only advances if the block
@ -479,6 +516,7 @@ impl<BP: BlockPublisherTrait> SequencerCore<BP> {
head_tip: head_tip.as_ref(),
final_snapshot: final_meta.as_ref().map(|meta| (chain.final_state(), meta)),
remove_deposit_records: &finalized_deposit_ids,
remove_dispatch_records: &finalized_dispatch_keys,
zone_anchor: Some(&record),
..StoreUpdate::new(chain.head_state())
})
@ -640,8 +678,42 @@ impl<BP: BlockPublisherTrait> SequencerCore<BP> {
fn build_block_from_mempool(&mut self) -> Result<BlockWithMeta> {
let now = Instant::now();
// Build on the head: its tip is the parent, its state the validation base.
let (prev_block_hash, new_block_height, mut working_state) = {
// Decoded outside the chain lock, and read before it is taken: the usual
// case is no delivery records at all, and decoding is the expensive part.
// One that does not decode is dropped rather than kept, since nothing
// will ever turn those bytes into a block transaction.
let mut settled = Vec::new();
let recorded_dispatches: Vec<_> = self
.store
.dbio()
.get_pending_cross_zone_dispatches()
.context("Failed to load pending cross-zone dispatches")?
.into_iter()
.filter_map(
|record| match borsh::from_slice::<LeeTransaction>(&record.transaction) {
Ok(tx) => {
let message = extract_cross_zone_dispatch(&tx);
Some((record.message_key, message, tx))
}
Err(err) => {
warn!(
"Dropping pending cross-zone dispatch {} that does not decode: {err:#}",
hex::encode(record.message_key)
);
settled.push(record.message_key);
None
}
},
)
.collect();
// Build on the head: its tip is the parent, its state the validation
// base.
//
// The delivery records are classified in here rather than after, so the
// final state can be read by reference. Cloning it cost a full state
// copy on every block of every zone, cross-zone or not.
let (prev_block_hash, new_block_height, mut working_state, pending_dispatches) = {
let chain = self.chain.lock().expect("chain state mutex poisoned");
let tip = chain.head_tip();
let height = tip.as_ref().map_or(GENESIS_BLOCK_ID, |head| {
@ -650,9 +722,43 @@ impl<BP: BlockPublisherTrait> SequencerCore<BP> {
.expect("block id should not overflow")
});
let prev = tip.map_or(HashType([0; 32]), |head| head.hash);
(prev, height, chain.head_state().clone())
// Three outcomes per record. Already in the final state means the
// delivery is irreversible, so the record is dropped; that is the
// only thing that removes a record the watcher re-added after its
// delivery had already settled, which it does whenever it re-reads a
// slot it has consumed. Already in the head state but not the final
// one means the delivery is on this chain but could still orphan, so
// the record is skipped and kept. Otherwise it goes in this block.
let mut pending: VecDeque<LeeTransaction> = VecDeque::new();
for (key, message, tx) in recorded_dispatches {
match message {
Some(message) if dispatch_already_delivered(chain.final_state(), &message) => {
settled.push(key);
}
Some(message) if dispatch_already_delivered(chain.head_state(), &message) => {}
_ if pending.len() >= MAX_DISPATCHES_PER_BLOCK => {}
_ => pending.push_back(tx),
}
}
(prev, height, chain.head_state().clone(), pending)
};
if !settled.is_empty()
&& 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()
);
}
let mut valid_transactions = Vec::new();
let mut withdrawals = Vec::new();
@ -666,7 +772,7 @@ impl<BP: BlockPublisherTrait> SequencerCore<BP> {
// build on — it was minted by us or by a peer whose block we adopted.
// An orphan reverts the receipt with the block, so the next turn
// re-mints without any bookkeeping of our own.
let mut pending_deposits: VecDeque<LeeTransaction> = self
let pending_deposits: VecDeque<LeeTransaction> = self
.store
.get_pending_deposit_events()
.context("Failed to load pending deposit events")?
@ -693,10 +799,13 @@ impl<BP: BlockPublisherTrait> SequencerCore<BP> {
let clock_tx = clock_invocation(new_block_timestamp);
let clock_lee_tx = LeeTransaction::Public(clock_tx.clone());
// Pending deposit mints first, then user work. `from_store` is not the
// same as a `Sequencer` origin — the cross-zone watcher pushes those
// into the mempool too.
while let Some((origin, tx, from_store)) = pending_deposits
// Everything drained from the store first, then user work. `from_store`
// is not the same as a `Sequencer` origin: it says the transaction has a
// record behind it and so needs no requeue, where the origin only says
// it was not submitted by a user.
let mut pending_from_store = pending_deposits;
pending_from_store.extend(pending_dispatches);
while let Some((origin, tx, from_store)) = pending_from_store
.pop_front()
.map(|tx| (TransactionOrigin::Sequencer, tx, true))
.or_else(|| self.mempool.pop().map(|(origin, tx)| (origin, tx, false)))
@ -721,12 +830,39 @@ impl<BP: BlockPublisherTrait> SequencerCore<BP> {
.len();
if block_size > max_block_size {
// Would a block carrying nothing but this still be too big? Then
// it does not fit in any block and deferring it defers it for
// ever. A store-drained transaction is at the head of the queue
// every turn, so breaking here would stop production reaching
// anything behind it, including the whole mempool, permanently.
// Count it against the delivery instead so it is given up on.
//
// Measured on its own rather than from `block_size`, which also
// counts whatever this block already holds: a transaction that
// merely does not fit *today* is the ordinary deferral below.
if from_store
&& !self.fits_in_an_empty_block(
&tx,
&clock_lee_tx,
new_block_height,
prev_block_hash,
new_block_timestamp,
)?
{
error!(
"Sequencer-drained transaction {tx_hash} cannot fit in any block under the \
{max_block_size} byte limit; giving up on it rather than stalling production",
);
self.count_dispatch_failure(&tx);
continue;
}
warn!(
"Transaction with hash {tx_hash} deferred to next block: \
block size {block_size} bytes would exceed limit of {max_block_size} bytes",
);
// A deposit mint needs no requeue: its record stays unfulfilled
// in the store and is drained again on the next turn.
// Anything drained from the store needs no requeue: its record
// stays there and is drained again on the next turn.
if !from_store {
self.mempool.push_front((origin, tx));
}
@ -742,6 +878,11 @@ impl<BP: BlockPublisherTrait> SequencerCore<BP> {
&mut withdrawals,
) {
valid_transactions.push(tx);
} else {
// A failed transaction is simply left out of the block, except a
// dispatch: that one is re-fed from the store every turn, so one
// that can never execute would fail on every block for ever.
self.count_dispatch_failure(&tx);
}
if valid_transactions.len() >= self.sequencer_config.max_num_tx_in_block {
@ -830,6 +971,72 @@ impl<BP: BlockPublisherTrait> SequencerCore<BP> {
&self.block_publisher
}
/// Whether a block carrying nothing but `tx` and the clock would be within
/// the size limit.
///
/// Distinguishes "does not fit in this block" from "does not fit in any
/// block". The first is an ordinary deferral; the second, for a transaction
/// the store re-feeds every turn, is a permanent stall unless it is given up
/// on.
fn fits_in_an_empty_block(
&self,
tx: &LeeTransaction,
clock_tx: &LeeTransaction,
block_id: u64,
prev_block_hash: HashType,
timestamp: u64,
) -> Result<bool> {
let alone = HashableBlockData {
block_id,
transactions: vec![tx.clone(), clock_tx.clone()],
prev_block_hash,
timestamp,
};
let size = borsh::to_vec(&alone)
.context("Failed to serialize block for size check")?
.len();
let max = usize::try_from(self.sequencer_config.max_block_size.as_u64())
.expect("`max_block_size` should fit into usize");
Ok(size <= max)
}
/// Counts one failed production attempt against `tx` if it is a cross-zone
/// delivery, giving up on it once too many accumulate.
///
/// 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.
fn count_dispatch_failure(&self, tx: &LeeTransaction) {
let Some(message) = extract_cross_zone_dispatch(tx) else {
return;
};
let key = cross_zone_inbox_core::message_key(
&message.src_zone,
message.src_block_id,
message.src_tx_index,
);
match self
.store
.dbio()
.record_dispatch_failure(key, RETIRE_DISPATCH_AFTER_FAILURES)
{
Ok(true) => error!(
"Giving up on cross-zone delivery {} after {RETIRE_DISPATCH_AFTER_FAILURES} failed attempts; it will not be retried",
hex::encode(key)
),
Ok(false) => warn!(
"Cross-zone delivery {} failed to execute, will retry next block",
hex::encode(key)
),
Err(err) => error!(
"Failed to count the failed attempt for cross-zone delivery {}: {err:#}",
hex::encode(key)
),
}
}
/// 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]
@ -845,7 +1052,10 @@ impl<BP: BlockPublisherTrait> SequencerCore<BP> {
/// what a restart does.
#[must_use]
pub fn background_tasks(&self) -> Vec<TaskGroup> {
vec![self.block_publisher.background_tasks()]
vec![
self.watchers.clone(),
self.block_publisher.background_tasks(),
]
}
/// Whether this sequencer is currently authorized to write to the channel.
@ -877,6 +1087,29 @@ fn deposit_already_minted(state: &lee::V03State, deposit_op_id: HashType) -> boo
.is_some_and(|receipt| *receipt != lee::Account::default())
}
/// 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.
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(),
&message.src_zone,
message.src_block_id,
);
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,
))
})
})
}
/// 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
@ -995,6 +1228,14 @@ fn apply_follow_update(
.filter_map(extract_bridge_deposit_id)
.collect();
// The same for cross-zone deliveries, keyed by message key: a record
// goes once its own delivery is irreversible, never because another
// block finalized at its height.
let finalized_dispatch_keys: Vec<[u8; 32]> = irreversible
.iter()
.flat_map(|block| settled_dispatch_keys(dbio, block))
.collect();
// A persist failure is fatal: the in-memory chain has already advanced,
// and continuing would leave a permanent gap in the store. The `panic!`
// ends the drive task, whose cancellation halts the node.
@ -1007,6 +1248,7 @@ fn apply_follow_update(
finalized_up_to: last_finalized,
new_deposit_events: &deposit_records,
remove_deposit_records: &finalized_deposit_ids,
remove_dispatch_records: &finalized_dispatch_keys,
consumed_withdrawals: &consumed_withdrawals,
..StoreUpdate::new(chain.head_state())
})
@ -1246,6 +1488,104 @@ fn is_sequencer_only_tx(tx: &LeeTransaction) -> bool {
if is_sequencer_only_program(tx.message().program_id))
}
/// The cross-zone message an inbox dispatch delivers, or `None` if `tx` is not
/// a dispatch.
#[must_use]
fn extract_cross_zone_dispatch(tx: &LeeTransaction) -> Option<CrossZoneMessage> {
let LeeTransaction::Public(tx) = tx else {
return None;
};
let message = tx.message();
if message.program_id != programs::cross_zone_inbox().id() {
return None;
}
match risc0_zkvm::serde::from_slice::<cross_zone_inbox_core::Instruction, u32>(
&message.instruction_data,
) {
Ok(cross_zone_inbox_core::Instruction::Dispatch(msg)) => Some(msg),
Ok(cross_zone_inbox_core::Instruction::InitConfig(_)) | Err(_) => None,
}
}
/// The content-addressed key of the message an inbox dispatch delivers.
///
/// A delivery in an irreversible block settles its pending record, so the record
/// is dropped by identity rather than by the height it happened to land at.
#[must_use]
fn extract_cross_zone_dispatch_key(tx: &LeeTransaction) -> Option<[u8; 32]> {
extract_cross_zone_dispatch(tx).map(|msg| {
cross_zone_inbox_core::message_key(&msg.src_zone, msg.src_block_id, msg.src_tx_index)
})
}
/// The keys of the deliveries `block` carries, reporting any whose transaction
/// is not the one we recorded for that key.
///
/// The key covers `(src_zone, src_block_id, src_tx_index)` and nothing about the
/// payload, and so does the inbox's own replay check, so a sequencer that
/// publishes a dispatch with the right key and a forged payload settles our
/// correct record along with it. The forgery is caught downstream by the
/// indexer, which re-derives every delivery and halts, but the local record is
/// the last copy of what we believed and it is about to be dropped either way.
/// Saying so in the log is what makes the halt diagnosable.
fn settled_dispatch_keys(dbio: &RocksDBIO, block: &Block) -> Vec<[u8; 32]> {
let recorded = dbio.get_pending_cross_zone_dispatches().unwrap_or_default();
let (keys, forged) = classify_settled_deliveries(&recorded, block);
for key in forged {
error!(
"Cross-zone delivery {} settled with a transaction that is not the one this node recorded for that key. The message key does not cover the payload, so a peer's sequencer can publish a different delivery under it.",
hex::encode(key)
);
}
keys
}
/// Splits the deliveries `block` carries into every settled key, and the subset
/// whose transaction is not the one `recorded` holds for that key.
///
/// Separated from the logging so the detection is testable: a forged delivery
/// leaves no trace in state that differs from an honest one, precisely because
/// the key does not cover the payload.
fn classify_settled_deliveries(
recorded: &[PendingCrossZoneDispatchRecord],
block: &Block,
) -> (Vec<[u8; 32]>, Vec<[u8; 32]>) {
let mut keys = Vec::new();
let mut forged = Vec::new();
for tx in &block.body.transactions {
let Some(key) = extract_cross_zone_dispatch_key(tx) else {
continue;
};
let mismatched = recorded
.iter()
.find(|record| record.message_key == key)
.is_some_and(|record| {
borsh::to_vec(tx).is_ok_and(|encoded| encoded != record.transaction)
});
if mismatched {
forged.push(key);
}
keys.push(key);
}
(keys, forged)
}
/// Drops the records of deliveries carried by a reconstructed block.
///
/// A persist failure is only logged: the deliveries are already irreversible, so
/// the worst case is a record the next drain drops instead.
fn settle_reconstructed_deliveries(store: &SequencerStore, block: &Block) {
let keys = settled_dispatch_keys(&store.dbio(), block);
if keys.is_empty() {
return;
}
if let Err(err) = store.dbio().drop_settled_cross_zone_dispatches(&keys) {
warn!("Failed to settle reconstructed delivery records: {err:#}");
}
}
#[must_use]
fn extract_bridge_deposit_id(tx: &LeeTransaction) -> Option<HashType> {
let LeeTransaction::Public(tx) = tx else {

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

@ -4,7 +4,7 @@ use std::{pin::pin, time::Duration};
use common::{
HashType,
block::{BedrockStatus, HashableBlockData},
block::{BedrockStatus, Block, HashableBlockData},
test_utils::sequencer_sign_key_for_testing,
transaction::{LeeTransaction, clock_invocation},
};
@ -29,23 +29,32 @@ use logos_blockchain_core::mantle::{
};
use logos_blockchain_zone_sdk::sequencer::DepositInfo;
use mempool::MemPoolHandle;
use storage::sequencer::sequencer_cells::PendingDepositEventRecord;
use ping_core::{ReceiverInstruction, ping_record_pda};
use storage::sequencer::sequencer_cells::{
PendingCrossZoneDispatchRecord, PendingDepositEventRecord,
};
use tempfile::tempdir;
use testnet_initial_state::{initial_pub_accounts_private_keys, initial_public_user_accounts};
use crate::{
TransactionOrigin, apply_follow_update,
MAX_DISPATCHES_PER_BLOCK, RETIRE_DISPATCH_AFTER_FAILURES, TransactionOrigin,
apply_follow_update,
block_publisher::FollowUpdate,
block_store::SequencerStore,
build_bridge_deposit_tx_from_event, build_genesis_state,
config::{BedrockConfig, GenesisAction, SequencerConfig},
deposit_already_minted, is_sequencer_only_program,
build_bridge_deposit_tx_from_event, build_genesis_state, classify_settled_deliveries,
config::{BedrockConfig, CrossZoneConfig, CrossZonePeer, GenesisAction, SequencerConfig},
deposit_already_minted, dispatch_already_delivered, extract_cross_zone_dispatch,
extract_cross_zone_dispatch_key, is_sequencer_only_program,
mock::{SequencerCoreWithMockClients, mock_checkpoint},
resubmittable_txs,
};
mod reconstruction;
/// The peer zone a cross-zone test receives from. Distinct from the test
/// channel id (`[0; 32]`), which the inbox guest rejects as a source.
const PEER_ZONE: [u8; 32] = [0xbe_u8; 32];
#[derive(borsh::BorshSerialize)]
struct DepositMetadataForEncoding {
recipient_id: lee::AccountId,
@ -162,6 +171,80 @@ fn tx_is_bridge_deposit(
)
}
/// A config that receives `ping_receiver` messages from [`PEER_ZONE`], so
/// `build_genesis_state` seeds the inbox config PDA and a delivery has an
/// allowlist to pass.
fn cross_zone_test_config() -> SequencerConfig {
SequencerConfig {
cross_zone: Some(CrossZoneConfig {
peers: vec![CrossZonePeer {
channel_id: PEER_ZONE,
allowed_targets: vec![programs::ping_receiver().id()],
expected_block_signing_pubkey: None,
}],
}),
..setup_sequencer_config()
}
}
/// A `ping_receiver::Record` instruction as risc0 words, little-endian: the wire
/// form an emitter on the peer zone puts in the message payload.
fn ping_payload(payload: &[u8]) -> Vec<u8> {
risc0_zkvm::serde::to_vec(&ReceiverInstruction::Record {
payload: payload.to_vec(),
})
.expect("ping instruction serializes")
.iter()
.flat_map(|word| word.to_le_bytes())
.collect()
}
/// The dispatch transaction for a message at index 0 of [`PEER_ZONE`] block
/// `src_block_id`. Built through the same builder the watcher uses, so a change
/// to the encoding shows up here rather than passing silently.
fn dispatch_tx(src_block_id: u64, payload: Vec<u8>) -> LeeTransaction {
let receiver_id = programs::ping_receiver().id();
LeeTransaction::Public(cross_zone::build_dispatch_from_emission(
PEER_ZONE,
src_block_id,
0,
programs::ping_sender().id(),
receiver_id,
&[ping_record_pda(receiver_id).into_value()],
payload,
))
}
/// The pending record the watcher would leave behind for that dispatch.
fn dispatch_record(src_block_id: u64, payload: Vec<u8>) -> PendingCrossZoneDispatchRecord {
let tx = dispatch_tx(src_block_id, payload);
PendingCrossZoneDispatchRecord::recorded(
cross_zone_inbox_core::message_key(&PEER_ZONE, src_block_id, 0),
borsh::to_vec(&tx).expect("dispatch encodes"),
)
}
/// The message keys of the deliveries a block carries.
fn dispatches_in(block: &Block) -> Vec<[u8; 32]> {
block
.body
.transactions
.iter()
.filter_map(extract_cross_zone_dispatch_key)
.collect()
}
/// The pending dispatch records a sequencer still holds.
fn pending_dispatches(
sequencer: &SequencerCoreWithMockClients,
) -> Vec<PendingCrossZoneDispatchRecord> {
sequencer
.store
.dbio()
.get_pending_cross_zone_dispatches()
.expect("pending dispatches readable")
}
#[tokio::test]
async fn start_from_config() {
let config = setup_sequencer_config();
@ -477,6 +560,363 @@ async fn a_replayed_deposit_mint_no_ops_in_the_guest() {
);
}
#[tokio::test]
async fn recorded_dispatches_are_drained_from_the_store_on_production() {
let payload = b"hello-cross-zone".to_vec();
let record = dispatch_record(7, ping_payload(&payload));
let key = record.message_key;
let (mut sequencer, _mempool_handle) =
SequencerCoreWithMockClients::start_from_config(cross_zone_test_config()).await;
assert_eq!(
sequencer
.store
.dbio()
.add_pending_cross_zone_dispatches(vec![record])
.unwrap(),
1
);
// The delivery never goes through the mempool: the record is the queue, and
// production drains it. That is what makes the window between the watcher's
// durable read cursor and a block carrying the dispatch survivable.
assert!(
sequencer.mempool.pop().is_none(),
"deliveries are drained from the store, never queued in the mempool"
);
let block_id = sequencer.produce_new_block().await.unwrap();
let block = sequencer
.store
.get_block_at_id(block_id)
.unwrap()
.expect("produced block is stored");
assert_eq!(
dispatches_in(&block),
vec![key],
"the drained delivery should be included in the produced block"
);
let record_id = ping_record_pda(programs::ping_receiver().id());
assert_eq!(
sequencer.with_state(|state| state.get_account_by_id(record_id).data.into_inner()),
payload,
"the dispatch must reach its target program, not just sit in the block"
);
// The record stays until the delivery finalizes; re-delivery is prevented by
// the inbox seen-set now in head state, not by any marker on the record.
assert_eq!(
pending_dispatches(&sequencer)
.iter()
.map(|record| record.message_key)
.collect::<Vec<_>>(),
vec![key],
"the record remains until the delivery becomes irreversible"
);
}
#[tokio::test]
async fn a_delivered_dispatch_is_skipped_on_the_next_turn() {
// The seen-set is what replaces the submitted mark: the drain asks the state
// it is building on whether the inbox has already taken this message, so a
// record that outlives its delivery costs one skipped drain, not a replay.
let record = dispatch_record(11, ping_payload(b"once"));
let key = record.message_key;
let (mut sequencer, _mempool_handle) =
SequencerCoreWithMockClients::start_from_config(cross_zone_test_config()).await;
sequencer
.store
.dbio()
.add_pending_cross_zone_dispatches(vec![record])
.unwrap();
let first = sequencer.produce_new_block().await.unwrap();
let second = sequencer.produce_new_block().await.unwrap();
let delivered_in = |block_id: u64| {
dispatches_in(
&sequencer
.store
.get_block_at_id(block_id)
.unwrap()
.expect("produced block is stored"),
)
};
assert_eq!(delivered_in(first), vec![key]);
assert!(
delivered_in(second).is_empty(),
"the inbox seen-set must keep the drain from re-delivering"
);
let message = extract_cross_zone_dispatch(&dispatch_tx(11, ping_payload(b"once")))
.expect("the dispatch carries a cross-zone message");
assert!(
sequencer.with_state(|state| dispatch_already_delivered(state, &message)),
"the seen shard in head state is what the skip reads"
);
}
#[tokio::test]
async fn a_dispatch_that_never_executes_is_given_up_on_after_repeated_failures() {
// A payload that is not `u32`-aligned: the inbox guest rejects it outright,
// so this is a delivery that can never execute however often it is retried.
// Its content is chosen on the peer zone and validated by nobody in between,
// so without a give-up policy it would fail on every block for ever.
let record = dispatch_record(13, b"odd".to_vec());
let (mut sequencer, _mempool_handle) =
SequencerCoreWithMockClients::start_from_config(cross_zone_test_config()).await;
sequencer
.store
.dbio()
.add_pending_cross_zone_dispatches(vec![record])
.unwrap();
for attempt in 1..RETIRE_DISPATCH_AFTER_FAILURES {
let block_id = sequencer.produce_new_block().await.unwrap();
let block = sequencer
.store
.get_block_at_id(block_id)
.unwrap()
.expect("produced block is stored");
assert!(
dispatches_in(&block).is_empty(),
"a dispatch that fails to execute must not reach the block"
);
let records = pending_dispatches(&sequencer);
assert_eq!(records.len(), 1);
assert_eq!(
records[0].failed_attempts, attempt,
"the counter advances once per block, not once per process start"
);
}
// 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.
sequencer.produce_new_block().await.unwrap();
assert!(
pending_dispatches(&sequencer).is_empty(),
"giving up on a delivery must drop its record, not flag it"
);
// 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();
assert!(dispatches_in(&block).is_empty());
assert!(pending_dispatches(&sequencer).is_empty());
}
#[tokio::test]
async fn a_redelivered_record_is_dropped_once_its_delivery_is_irreversible() {
// The watcher persists its floor only at slot boundaries, so a crash inside
// a slot makes the next run re-read it and re-record deliveries that have
// already settled. Their keys are in the inbox seen-set for good, so no
// future block will ever carry them and the settlement path cannot reach
// them. The drain dropping them is the only thing that does.
let record = dispatch_record(29, ping_payload(b"again"));
let key = record.message_key;
let (mut sequencer, mempool_handle) =
SequencerCoreWithMockClients::start_from_config(cross_zone_test_config()).await;
sequencer
.store
.dbio()
.add_pending_cross_zone_dispatches(vec![record.clone()])
.unwrap();
let block_id = sequencer.produce_new_block().await.unwrap();
let delivery_block = sequencer.store.get_block_at_id(block_id).unwrap().unwrap();
assert_eq!(dispatches_in(&delivery_block), vec![key]);
apply_follow_update(
&sequencer.store.dbio(),
&sequencer.chain(),
&mempool_handle,
FollowUpdate {
finalized: vec![(MsgId::from(delivery_block.header.hash.0), delivery_block)],
..empty_follow_update()
},
);
assert!(pending_dispatches(&sequencer).is_empty());
// The watcher re-reads the slot and records it again.
sequencer
.store
.dbio()
.add_pending_cross_zone_dispatches(vec![record])
.unwrap();
assert_eq!(pending_dispatches(&sequencer).len(), 1);
let block_id = sequencer.produce_new_block().await.unwrap();
let block = sequencer.store.get_block_at_id(block_id).unwrap().unwrap();
assert!(
dispatches_in(&block).is_empty(),
"the delivery is already on the chain, so it must not be delivered again"
);
assert!(
pending_dispatches(&sequencer).is_empty(),
"a record whose delivery is already irreversible must be dropped, not kept for ever"
);
}
#[tokio::test]
async fn a_delivery_still_reversible_keeps_its_record() {
// The counterpart to the test above, and the reason the drain checks two
// states rather than one. In head but not yet final means the delivery can
// still orphan, so skipping it is right but dropping its record would lose
// the delivery when it does.
let record = dispatch_record(31, ping_payload(b"pending"));
let key = record.message_key;
let (mut sequencer, _mempool_handle) =
SequencerCoreWithMockClients::start_from_config(cross_zone_test_config()).await;
sequencer
.store
.dbio()
.add_pending_cross_zone_dispatches(vec![record])
.unwrap();
sequencer.produce_new_block().await.unwrap();
sequencer.produce_new_block().await.unwrap();
assert_eq!(
pending_dispatches(&sequencer)
.iter()
.map(|record| record.message_key)
.collect::<Vec<_>>(),
vec![key],
"nothing has finalized, so the record must survive in case the block orphans"
);
}
#[test]
fn a_settled_delivery_that_is_not_the_one_we_recorded_is_reported() {
// The message key covers (src_zone, src_block_id, src_tx_index) and nothing
// about the payload, and so does the inbox's own replay check. So a peer's
// sequencer can publish a delivery under a key we hold with a payload we
// never saw, and it settles our correct record along with it. The indexer
// catches the forgery and halts; this record is the last local copy of what
// we believed, so the mismatch has to be reported before it is dropped.
let honest = dispatch_record(53, ping_payload(b"honest"));
let key = honest.message_key;
let forged = dispatch_tx(53, ping_payload(b"forged"));
assert_eq!(
extract_cross_zone_dispatch_key(&forged),
Some(key),
"the forged delivery must share the key, or it proves nothing"
);
let block = common::test_utils::produce_dummy_block(2, None, vec![forged]);
let (keys, mismatched) = classify_settled_deliveries(std::slice::from_ref(&honest), &block);
assert_eq!(keys, vec![key], "the record is settled either way");
assert_eq!(
mismatched,
vec![key],
"a delivery that differs from the one recorded under that key must be reported"
);
// The honest case must stay quiet, or the report is noise.
let honest_block = common::test_utils::produce_dummy_block(
2,
None,
vec![dispatch_tx(53, ping_payload(b"honest"))],
);
let (keys, mismatched) = classify_settled_deliveries(&[honest], &honest_block);
assert_eq!(keys, vec![key]);
assert!(mismatched.is_empty());
}
#[tokio::test]
async fn a_delivery_too_large_for_any_block_does_not_stall_production() {
// A store-drained transaction is at the head of the queue every turn, so one
// that cannot fit in any block would defer itself for ever and, because the
// deferral breaks the loop, stop production ever reaching the mempool behind
// it. The peer chooses the payload, so this is theirs to trigger.
let record = dispatch_record(41, ping_payload(&[7_u8; 8192]));
let mut config = cross_zone_test_config();
config.max_block_size = bytesize::ByteSize::kib(4);
let (mut sequencer, mempool_handle) =
SequencerCoreWithMockClients::start_from_config(config).await;
sequencer
.store
.dbio()
.add_pending_cross_zone_dispatches(vec![record])
.unwrap();
let user_tx = common::test_utils::create_transaction_native_token_transfer(
initial_public_user_accounts()[0].account_id,
0,
initial_public_user_accounts()[1].account_id,
10,
&create_signing_key_for_account1(),
);
mempool_handle
.push((TransactionOrigin::User, user_tx.clone()))
.await
.unwrap();
// Production must get past it to the mempool in the very first block.
let block_id = sequencer.produce_new_block().await.unwrap();
let block = sequencer.store.get_block_at_id(block_id).unwrap().unwrap();
assert!(
block.body.transactions.contains(&user_tx),
"an oversized drained delivery must not stop production reaching the mempool"
);
assert!(dispatches_in(&block).is_empty());
// And it is given up on rather than retried for ever.
for _ in 1..RETIRE_DISPATCH_AFTER_FAILURES {
sequencer.produce_new_block().await.unwrap();
}
assert!(
pending_dispatches(&sequencer).is_empty(),
"a delivery that fits in no block must be given up on"
);
}
#[tokio::test]
async fn a_delivery_backlog_is_spread_across_blocks() {
// Each delivery costs a guest execution and peers decide how many queue up,
// so an unbounded drain would let a backlog decide how long a block takes to
// build and leave no room for user work, since store-drained transactions
// are taken before the mempool.
let backlog = MAX_DISPATCHES_PER_BLOCK + 3;
let records: Vec<_> = (0..backlog)
.map(|index| {
let src_block_id = 100 + u64::try_from(index).expect("test index fits");
dispatch_record(src_block_id, ping_payload(b"backlog"))
})
.collect();
let mut config = cross_zone_test_config();
config.max_num_tx_in_block = backlog + 10;
let (mut sequencer, _mempool_handle) =
SequencerCoreWithMockClients::start_from_config(config).await;
sequencer
.store
.dbio()
.add_pending_cross_zone_dispatches(records)
.unwrap();
let block_id = sequencer.produce_new_block().await.unwrap();
let block = sequencer.store.get_block_at_id(block_id).unwrap().unwrap();
assert_eq!(
dispatches_in(&block).len(),
MAX_DISPATCHES_PER_BLOCK,
"one block must not carry an unbounded number of deliveries"
);
// Deferred, not dropped: the rest go in the next block.
let block_id = sequencer.produce_new_block().await.unwrap();
let block = sequencer.store.get_block_at_id(block_id).unwrap().unwrap();
assert_eq!(dispatches_in(&block).len(), 3);
}
#[test]
fn transaction_pre_check_pass() {
let tx = common::test_utils::produce_dummy_empty_transaction();
@ -1774,6 +2214,98 @@ async fn follow_finalized_own_block_moves_final_tier_and_marks_store() {
assert!(matches!(stored.bedrock_status, BedrockStatus::Finalized));
}
#[tokio::test]
async fn follow_finalized_delivery_drops_its_pending_record() {
// The record exists to bridge the gap between the watcher's durable read
// cursor and a block that carries the delivery. Once that block is
// irreversible the delivery cannot be lost any more, so the record is owed
// nothing and goes with the same update that made the block irreversible.
let record = dispatch_record(17, ping_payload(b"settled"));
let key = record.message_key;
let (mut sequencer, mempool_handle) =
SequencerCoreWithMockClients::start_from_config(cross_zone_test_config()).await;
sequencer
.store
.dbio()
.add_pending_cross_zone_dispatches(vec![record])
.unwrap();
let block_id = sequencer.produce_new_block().await.unwrap();
let delivery_block = sequencer.store.get_block_at_id(block_id).unwrap().unwrap();
assert_eq!(dispatches_in(&delivery_block), vec![key]);
assert_eq!(
pending_dispatches(&sequencer).len(),
1,
"including the delivery is not enough to settle its record"
);
apply_follow_update(
&sequencer.store.dbio(),
&sequencer.chain(),
&mempool_handle,
FollowUpdate {
finalized: vec![(MsgId::from(delivery_block.header.hash.0), delivery_block)],
..empty_follow_update()
},
);
assert!(
pending_dispatches(&sequencer).is_empty(),
"a delivery in an irreversible block settles its record"
);
}
#[tokio::test]
async fn a_parked_finalized_block_does_not_drop_a_dispatch_record() {
// Keyed by message key, not by height: a finalized block the final tier
// parks never became irreversible, so nothing it happens to sit above may
// settle a record. Dropping one here would lose the delivery for good, since
// the watcher's floor has already moved past the peer block it came from.
let record = dispatch_record(19, ping_payload(b"parked"));
let key = record.message_key;
let delivery = dispatch_tx(19, ping_payload(b"parked"));
let (mut sequencer, mempool_handle) =
SequencerCoreWithMockClients::start_from_config(cross_zone_test_config()).await;
sequencer
.store
.dbio()
.add_pending_cross_zone_dispatches(vec![record])
.unwrap();
let tx = common::test_utils::produce_dummy_empty_transaction();
mempool_handle
.push((TransactionOrigin::User, tx))
.await
.unwrap();
sequencer.produce_new_block().await.unwrap();
// A skip-ahead block carrying the same delivery: not in head and linking to
// nothing we hold, so the final tier parks it instead of applying it.
let parked =
common::test_utils::produce_dummy_block(9, Some(HashType([44; 32])), vec![delivery]);
apply_follow_update(
&sequencer.store.dbio(),
&sequencer.chain(),
&mempool_handle,
FollowUpdate {
finalized: vec![(MsgId::from([9; 32]), parked)],
..empty_follow_update()
},
);
assert_eq!(
pending_dispatches(&sequencer)
.iter()
.map(|record| record.message_key)
.collect::<Vec<_>>(),
vec![key],
"a parked finalized block must not drop its delivery's record"
);
}
#[tokio::test]
async fn follow_finalized_backfill_block_is_applied_and_marked_finalized() {
let config = setup_sequencer_config();

140
lez/sequencer/core/src/tests/reconstruction.rs
View File

@ -586,6 +586,146 @@ async fn reconstruction_reconciles_already_finished_deposit() {
);
}
/// A cross-zone delivery whose record is still pending locally, but whose block
/// arrives already finalized on the channel. Reconstruction must settle the
/// record on the way through: the delivery is permanently reflected in the
/// reconstructed state (the inbox seen shard), so the next production neither
/// re-delivers it nor leaves a record nothing will ever drop.
#[tokio::test]
async fn reconstructed_delivery_settles_its_pending_record() {
let payload = b"reconstructed".to_vec();
let record = dispatch_record(23, ping_payload(&payload));
let key = record.message_key;
// Sequencer A produces the block that carries the delivery.
let config_a = cross_zone_test_config();
let (mut seq_a, _mempool_a) =
SequencerCoreWithMockClients::start_from_config(config_a.clone()).await;
seq_a
.block_store()
.dbio()
.add_pending_cross_zone_dispatches(vec![record.clone()])
.unwrap();
seq_a.produce_new_block().await.unwrap();
let tip_a = seq_a.block_store().latest_block_meta().unwrap().unwrap();
let messages = channel_from_store(seq_a.block_store(), 10);
let tip_slot = messages.last().unwrap().1;
let channel_id = config_a.bedrock_config.channel_id;
// Sequencer B holds the same record, as its own watcher would after reading
// the peer block, and reconstructs A's chain from a fresh store.
let (mut seq_b, _mempool_b) =
SequencerCoreWithMockClients::start_from_config(cross_zone_test_config()).await;
assert_eq!(
seq_b
.block_store()
.dbio()
.add_pending_cross_zone_dispatches(vec![record])
.unwrap(),
1
);
let mock_b = MockBlockPublisher::with_canned_channel(channel_id, Some(tip_slot), messages);
SequencerCore::<MockBlockPublisher>::verify_and_reconstruct(
&mock_b,
&seq_b.store,
&seq_b.chain,
true,
)
.await
.expect("reconstruct");
let tip_b = seq_b.block_store().latest_block_meta().unwrap().unwrap();
assert_eq!(tip_b.id, tip_a.id);
assert_eq!(tip_b.hash, tip_a.hash);
assert!(
seq_b
.block_store()
.dbio()
.get_pending_cross_zone_dispatches()
.unwrap()
.is_empty(),
"reconstruction must settle the record of a delivery it replayed"
);
// The delivery landed exactly once, and the next turn does not re-emit it.
let record_id = ping_record_pda(programs::ping_receiver().id());
assert_eq!(
seq_b.with_state(|state| state.get_account_by_id(record_id).data.into_inner()),
payload,
"the reconstructed delivery must reach its target program"
);
seq_b.produce_new_block().await.unwrap();
let produced = seq_b
.block_store()
.get_block_at_id(tip_b.id + 1)
.unwrap()
.expect("produced block present");
assert!(
!dispatches_in(&produced).contains(&key),
"the reconstructed delivery must not be re-emitted"
);
}
/// A delivery this node published itself, served back by the channel at or below
/// its own tip. That path verifies the block matches and returns early, so it is
/// reached on every restart. It must still settle the delivery's record: the
/// channel serving the block is what makes it irreversible, and nothing later
/// will ever put that key in a block again.
#[tokio::test]
async fn a_verified_own_block_settles_its_delivery_records() {
let record = dispatch_record(37, ping_payload(b"verified"));
let key = record.message_key;
let (mut seq, _mempool) =
SequencerCoreWithMockClients::start_from_config(cross_zone_test_config()).await;
seq.block_store()
.dbio()
.add_pending_cross_zone_dispatches(vec![record])
.unwrap();
let block_id = seq.produce_new_block().await.unwrap();
let block = seq
.block_store()
.get_block_at_id(block_id)
.unwrap()
.unwrap();
assert_eq!(dispatches_in(&block), vec![key]);
assert_eq!(
seq.block_store()
.dbio()
.get_pending_cross_zone_dispatches()
.unwrap()
.len(),
1,
"producing the block is not what settles the record"
);
// The channel serves our own chain back, tip included.
let messages = channel_from_store(seq.block_store(), 10);
let tip_slot = messages.last().unwrap().1;
let mock = MockBlockPublisher::with_canned_channel(
seq.sequencer_config.bedrock_config.channel_id,
Some(tip_slot),
messages,
);
SequencerCore::<MockBlockPublisher>::verify_and_reconstruct(
&mock, &seq.store, &seq.chain, true,
)
.await
.expect("reconstruct");
assert!(
seq.block_store()
.dbio()
.get_pending_cross_zone_dispatches()
.unwrap()
.is_empty(),
"a delivery the channel confirms must not leave a record nothing can remove"
);
}
#[tokio::test]
async fn committed_local_against_missing_channel_fails_without_anchor() {
// A sequencer that has committed blocks — a non-genesis tip plus a persisted