mirror of
https://github.com/logos-blockchain/lssa.git
synced 2026-08-01 12:33:24 +00:00
Merge pull request #652 from logos-blockchain/moudy/durable-watcher-cursor
feat(cross-zone): persist each watcher's peer read cursor
This commit is contained in:
Moudy
GitHub
commit
39bdec0d44
1
Cargo.lock
generated
1
Cargo.lock
generated
@ -9307,6 +9307,7 @@ dependencies = [
|
||||
"logos-blockchain-zone-sdk",
|
||||
"mempool",
|
||||
"num-bigint 0.4.6",
|
||||
"ping_core",
|
||||
"programs",
|
||||
"rand 0.8.6",
|
||||
"risc0-zkvm",
|
||||
|
||||
227
integration_tests/tests/cross_zone_watcher_restart.rs
Normal file
227
integration_tests/tests/cross_zone_watcher_restart.rs
Normal file
@ -0,0 +1,227 @@
|
||||
#![expect(
|
||||
clippy::tests_outside_test_module,
|
||||
reason = "top-level test functions are conventional for integration tests"
|
||||
)]
|
||||
|
||||
//! A sequencer restart must resume its cross-zone watcher from the persisted
|
||||
//! per-peer delivery floor instead of re-reading the peer channel from genesis.
|
||||
//!
|
||||
//! Re-reading is safe (the dispatch key is content-addressed and the inbox
|
||||
//! no-ops a replay) so on-chain state cannot tell the two apart. What does tell
|
||||
//! them apart is the transactions: a watcher that lost its cursor re-injects
|
||||
//! every already-delivered dispatch, which shows up as inbox transactions in
|
||||
//! blocks produced after the restart. This is the only test that covers the
|
||||
//! wiring from `spawn_watchers` through the store, so a silent regression here
|
||||
//! would leave every other test green while the feature does nothing.
|
||||
|
||||
use std::time::Duration;
|
||||
|
||||
use anyhow::{Context as _, Result};
|
||||
use common::transaction::LeeTransaction;
|
||||
use cross_zone_outbox_core::outbox_pda;
|
||||
use integration_tests::{
|
||||
config::{self, SequencerPartialConfig},
|
||||
setup::{SequencerSetup, sequencer_client, setup_bedrock_node},
|
||||
};
|
||||
use lee::{AccountId, PublicTransaction, public_transaction::Message};
|
||||
use lee_core::program::ProgramId;
|
||||
use ping_core::{ReceiverInstruction, SenderInstruction, ping_record_pda};
|
||||
use sequencer_core::config::{CrossZoneConfig, CrossZonePeer};
|
||||
use sequencer_service_rpc::{RpcClient as _, SequencerClient};
|
||||
use tokio::test;
|
||||
|
||||
const DELIVERY_TIMEOUT: Duration = Duration::from_secs(480);
|
||||
/// Blocks zone B must produce after the restart before we judge the watcher.
|
||||
/// A watcher that lost its cursor re-reads the peer channel on its first pass,
|
||||
/// so a handful of blocks is ample room for the replay to appear.
|
||||
const BLOCKS_AFTER_RESTART: u64 = 5;
|
||||
const RESTART_TIMEOUT: Duration = Duration::from_secs(240);
|
||||
const PING_PAYLOAD: &[u8] = b"hello-cross-zone";
|
||||
|
||||
#[test]
|
||||
async fn restarted_watcher_resumes_instead_of_replaying_the_peer_channel() -> Result<()> {
|
||||
// Declared first so it outlives both zones (drops run in reverse order).
|
||||
let (_bedrock, bedrock_addr) = setup_bedrock_node()
|
||||
.await
|
||||
.context("Failed to set up shared Bedrock node")?;
|
||||
|
||||
let partial = SequencerPartialConfig::default();
|
||||
let channel_a = config::bedrock_channel_id();
|
||||
let channel_b = config::bedrock_channel_id_b();
|
||||
let zone_a: [u8; 32] = *channel_a.as_ref();
|
||||
let zone_b: [u8; 32] = *channel_b.as_ref();
|
||||
let receiver_id = programs::ping_receiver().id();
|
||||
|
||||
let cross_zone = CrossZoneConfig {
|
||||
peers: vec![CrossZonePeer {
|
||||
channel_id: zone_a,
|
||||
allowed_targets: vec![receiver_id],
|
||||
expected_block_signing_pubkey: None,
|
||||
}],
|
||||
};
|
||||
|
||||
let (seq_a, _seq_a_home) = SequencerSetup::new(partial, bedrock_addr)
|
||||
.with_channel_id(channel_a)
|
||||
.with_genesis(vec![])
|
||||
.setup()
|
||||
.await
|
||||
.context("Failed to set up zone A sequencer")?;
|
||||
|
||||
// Zone B keeps an explicit home so it can be restarted on the same store.
|
||||
let home_b = tempfile::tempdir().context("Failed to create zone B home")?;
|
||||
let mut seq_b = SequencerSetup::new(partial, bedrock_addr)
|
||||
.with_channel_id(channel_b)
|
||||
.with_genesis(vec![])
|
||||
.with_cross_zone(cross_zone.clone())
|
||||
.setup_at(home_b.path())
|
||||
.await
|
||||
.context("Failed to set up zone B sequencer")?;
|
||||
|
||||
// Deliver one ping, so the peer channel holds a dispatch worth replaying.
|
||||
sequencer_client(seq_a.addr())?
|
||||
.send_transaction(build_ping_tx(zone_b, receiver_id))
|
||||
.await
|
||||
.context("Failed to submit ping on zone A")?;
|
||||
let record_id = ping_record_pda(receiver_id);
|
||||
let delivered = wait_for_delivery(sequencer_client(seq_b.addr())?, record_id).await?;
|
||||
assert_eq!(
|
||||
delivered, PING_PAYLOAD,
|
||||
"Zone B must record the payload before the restart"
|
||||
);
|
||||
|
||||
let tip_before = sequencer_client(seq_b.addr())?.get_last_block_id().await?;
|
||||
|
||||
// Restart zone B on the same home. Zone A stays quiet from here, so any
|
||||
// inbox transaction after the restart is a replay, not a new delivery.
|
||||
//
|
||||
// `shutdown` rather than `drop`: dropping aborts the main loop without
|
||||
// awaiting it and leaves the watchers and the publisher's drive task holding
|
||||
// the store, so the reopen below would race the `RocksDB` lock.
|
||||
seq_b.shutdown().await;
|
||||
seq_b = SequencerSetup::new(partial, bedrock_addr)
|
||||
.with_channel_id(channel_b)
|
||||
.with_genesis(vec![])
|
||||
.with_cross_zone(cross_zone)
|
||||
.setup_at(home_b.path())
|
||||
.await
|
||||
.context("Failed to restart zone B sequencer")?;
|
||||
let client_b = sequencer_client(seq_b.addr())?;
|
||||
|
||||
let tip_after = wait_for_block_id(
|
||||
&client_b,
|
||||
tip_before.saturating_add(BLOCKS_AFTER_RESTART),
|
||||
RESTART_TIMEOUT,
|
||||
)
|
||||
.await?;
|
||||
|
||||
let replayed = count_inbox_transactions(&client_b, tip_before.saturating_add(1), tip_after)
|
||||
.await
|
||||
.context("Failed to scan zone B blocks after the restart")?;
|
||||
assert_eq!(
|
||||
replayed,
|
||||
0,
|
||||
"a restarted watcher must resume from its persisted delivery floor; found {replayed} inbox transaction(s) in blocks {}..={tip_after}, which means it re-read the peer channel from genesis",
|
||||
tip_before.saturating_add(1)
|
||||
);
|
||||
|
||||
// The delivery itself must survive the restart untouched.
|
||||
let account = client_b.get_account(record_id).await?;
|
||||
assert_eq!(
|
||||
account.data.into_inner(),
|
||||
PING_PAYLOAD,
|
||||
"the delivered payload must survive the restart"
|
||||
);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Counts inbox transactions across `from..=to`, the signature of a re-injected
|
||||
/// dispatch.
|
||||
async fn count_inbox_transactions(client: &SequencerClient, from: u64, to: u64) -> Result<usize> {
|
||||
let inbox_id = programs::cross_zone_inbox().id();
|
||||
let mut count = 0_usize;
|
||||
for block_id in from..=to {
|
||||
let Some(block) = client.get_block(block_id).await? else {
|
||||
continue;
|
||||
};
|
||||
for tx in &block.body.transactions {
|
||||
if let LeeTransaction::Public(public_tx) = tx
|
||||
&& public_tx.message().program_id == inbox_id
|
||||
{
|
||||
count = count.saturating_add(1);
|
||||
}
|
||||
}
|
||||
}
|
||||
Ok(count)
|
||||
}
|
||||
|
||||
/// Waits until the sequencer's tip reaches `target`, returning the tip.
|
||||
async fn wait_for_block_id(
|
||||
client: &SequencerClient,
|
||||
target: u64,
|
||||
timeout: Duration,
|
||||
) -> Result<u64> {
|
||||
let wait = async {
|
||||
loop {
|
||||
let tip = client.get_last_block_id().await?;
|
||||
if tip >= target {
|
||||
return Ok::<u64, anyhow::Error>(tip);
|
||||
}
|
||||
tokio::time::sleep(Duration::from_secs(2)).await;
|
||||
}
|
||||
};
|
||||
tokio::time::timeout(timeout, wait)
|
||||
.await
|
||||
.context("Zone B did not produce enough blocks after the restart")?
|
||||
}
|
||||
|
||||
/// Builds a top-level `ping_sender` transaction that chains into the outbox to emit
|
||||
/// a message carrying a `ping_receiver::Record` instruction for the target zone.
|
||||
fn build_ping_tx(target_zone: [u8; 32], receiver_id: ProgramId) -> LeeTransaction {
|
||||
let outbox_id = programs::cross_zone_outbox().id();
|
||||
let ordinal = 0;
|
||||
|
||||
let words = risc0_zkvm::serde::to_vec(&ReceiverInstruction::Record {
|
||||
payload: PING_PAYLOAD.to_vec(),
|
||||
})
|
||||
.expect("serialize ping instruction");
|
||||
let payload: Vec<u8> = words.iter().flat_map(|word| word.to_le_bytes()).collect();
|
||||
|
||||
let send = SenderInstruction::Send {
|
||||
outbox_program_id: outbox_id,
|
||||
target_zone,
|
||||
target_program_id: receiver_id,
|
||||
target_accounts: vec![ping_record_pda(receiver_id).into_value()],
|
||||
payload,
|
||||
ordinal,
|
||||
};
|
||||
|
||||
let outbox_account = outbox_pda(outbox_id, &target_zone, ordinal);
|
||||
let message = Message::try_new(
|
||||
programs::ping_sender().id(),
|
||||
vec![outbox_account],
|
||||
vec![],
|
||||
send,
|
||||
)
|
||||
.expect("build ping message");
|
||||
LeeTransaction::Public(PublicTransaction::new(
|
||||
message,
|
||||
lee::public_transaction::WitnessSet::from_raw_parts(vec![]),
|
||||
))
|
||||
}
|
||||
|
||||
/// Polls zone B's sequencer until the ping record PDA holds a payload.
|
||||
async fn wait_for_delivery(client: SequencerClient, record_id: AccountId) -> Result<Vec<u8>> {
|
||||
let wait = async {
|
||||
loop {
|
||||
let account = client.get_account(record_id).await?;
|
||||
let data = account.data.into_inner();
|
||||
if !data.is_empty() {
|
||||
return Ok::<Vec<u8>, anyhow::Error>(data);
|
||||
}
|
||||
tokio::time::sleep(Duration::from_secs(3)).await;
|
||||
}
|
||||
};
|
||||
tokio::time::timeout(DELIVERY_TIMEOUT, wait)
|
||||
.await
|
||||
.context("Zone B did not record the cross-zone payload in time")?
|
||||
}
|
||||
@ -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
|
||||
|
||||
@ -1,4 +1,4 @@
|
||||
use std::{sync::Arc, time::Duration};
|
||||
use std::time::Duration;
|
||||
|
||||
use anyhow::{Context as _, Result, anyhow, ensure};
|
||||
use common::block::Block;
|
||||
@ -34,13 +34,10 @@ use logos_blockchain_zone_sdk::{
|
||||
ZoneSequencer,
|
||||
},
|
||||
};
|
||||
use tokio::{
|
||||
sync::{mpsc, oneshot, watch},
|
||||
task::JoinHandle,
|
||||
};
|
||||
use tokio::sync::{mpsc, oneshot, watch};
|
||||
use tokio_util::sync::CancellationToken;
|
||||
|
||||
use crate::config::BedrockConfig;
|
||||
use crate::{config::BedrockConfig, task_group::TaskGroup};
|
||||
|
||||
/// Channel capacity for the publish inbox. One publish per produced block, drained
|
||||
/// in microseconds by the drive task — 32 is huge headroom and just provides
|
||||
@ -120,6 +117,14 @@ pub trait BlockPublisherTrait: Sized {
|
||||
/// are processed past that point, so the node must halt.
|
||||
fn driver_cancellation(&self) -> CancellationToken;
|
||||
|
||||
/// The publisher's background tasks, for a caller that needs to know when
|
||||
/// they have actually stopped. Its sinks capture a store handle, so the
|
||||
/// `RocksDB` lock outlives the sequencer until the drive task is gone.
|
||||
/// Empty by default, for publishers that run no tasks.
|
||||
fn background_tasks(&self) -> TaskGroup {
|
||||
TaskGroup::default()
|
||||
}
|
||||
|
||||
/// Current channel frontier slot on the connected chain, or `None` if the
|
||||
/// channel does not exist there. Drives the startup frontier check.
|
||||
async fn channel_tip_slot(&self) -> Result<Option<Slot>>;
|
||||
@ -143,19 +148,12 @@ pub struct ZoneSdkPublisher {
|
||||
turn_rx: watch::Receiver<TurnNotification>,
|
||||
// Cancelled when the drive task ends for any reason, including a panic.
|
||||
driver_cancellation: CancellationToken,
|
||||
// Aborts the drive task when the last clone is dropped.
|
||||
_drive_task: Arc<DriveTaskGuard>,
|
||||
// Stops the drive task when the last clone is dropped, and lets a shutdown
|
||||
// path wait until it has actually stopped.
|
||||
drive_task: TaskGroup,
|
||||
indexer: ZoneIndexer<NodeHttpClient>,
|
||||
}
|
||||
|
||||
struct DriveTaskGuard(JoinHandle<()>);
|
||||
|
||||
impl Drop for DriveTaskGuard {
|
||||
fn drop(&mut self) {
|
||||
self.0.abort();
|
||||
}
|
||||
}
|
||||
|
||||
impl BlockPublisherTrait for ZoneSdkPublisher {
|
||||
async fn new(
|
||||
config: &BedrockConfig,
|
||||
@ -318,7 +316,7 @@ impl BlockPublisherTrait for ZoneSdkPublisher {
|
||||
command_tx,
|
||||
turn_rx,
|
||||
driver_cancellation,
|
||||
_drive_task: Arc::new(DriveTaskGuard(drive_task)),
|
||||
drive_task: TaskGroup::new(vec![drive_task]),
|
||||
})
|
||||
}
|
||||
|
||||
@ -359,6 +357,10 @@ impl BlockPublisherTrait for ZoneSdkPublisher {
|
||||
self.driver_cancellation.clone()
|
||||
}
|
||||
|
||||
fn background_tasks(&self) -> TaskGroup {
|
||||
self.drive_task.clone()
|
||||
}
|
||||
|
||||
async fn channel_tip_slot(&self) -> Result<Option<Slot>> {
|
||||
Ok(self
|
||||
.node
|
||||
|
||||
@ -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};
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
@ -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,12 +34,16 @@ 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::{
|
||||
block_publisher::{BlockPublisherTrait, MsgId, ZoneSdkPublisher},
|
||||
block_store::SequencerStore,
|
||||
task_group::{StoreRelease, TaskGroup},
|
||||
};
|
||||
|
||||
pub mod block_publisher;
|
||||
@ -48,6 +53,23 @@ pub mod cross_zone_watcher;
|
||||
|
||||
#[cfg(feature = "mock")]
|
||||
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)]
|
||||
@ -71,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> {
|
||||
@ -202,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).
|
||||
@ -267,6 +296,7 @@ impl<BP: BlockPublisherTrait> SequencerCore<BP> {
|
||||
mempool,
|
||||
sequencer_config: config,
|
||||
block_publisher,
|
||||
watchers,
|
||||
};
|
||||
|
||||
(sequencer_core, mempool_handle)
|
||||
@ -423,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")?;
|
||||
@ -464,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
|
||||
@ -477,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())
|
||||
})
|
||||
@ -638,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| {
|
||||
@ -648,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();
|
||||
|
||||
@ -664,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")?
|
||||
@ -691,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)))
|
||||
@ -719,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));
|
||||
}
|
||||
@ -740,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 {
|
||||
@ -828,6 +971,93 @@ 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]
|
||||
pub fn store_release(&self) -> StoreRelease {
|
||||
StoreRelease::new(&self.store.dbio())
|
||||
}
|
||||
|
||||
/// Every background task that holds this sequencer's store handle.
|
||||
///
|
||||
/// Taken before the core is shared, so a shutdown path can wait for them
|
||||
/// without owning the core. Until all of them have stopped the `RocksDB`
|
||||
/// lock is still held and the home directory cannot be reopened, which is
|
||||
/// what a restart does.
|
||||
#[must_use]
|
||||
pub fn background_tasks(&self) -> Vec<TaskGroup> {
|
||||
vec![
|
||||
self.watchers.clone(),
|
||||
self.block_publisher.background_tasks(),
|
||||
]
|
||||
}
|
||||
|
||||
/// Whether this sequencer is currently authorized to write to the channel.
|
||||
#[must_use]
|
||||
pub fn is_our_turn(&self) -> bool {
|
||||
@ -857,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
|
||||
@ -975,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.
|
||||
@ -987,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())
|
||||
})
|
||||
@ -1226,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 {
|
||||
|
||||
137
lez/sequencer/core/src/task_group.rs
Normal file
137
lez/sequencer/core/src/task_group.rs
Normal file
@ -0,0 +1,137 @@
|
||||
//! A set of background tasks that can be stopped and waited on.
|
||||
|
||||
use std::sync::{Arc, Mutex, MutexGuard, PoisonError, Weak};
|
||||
|
||||
use log::warn;
|
||||
use storage::sequencer::RocksDBIO;
|
||||
use tokio::task::JoinHandle;
|
||||
|
||||
/// Background tasks owned by one component, stoppable on demand and stopped
|
||||
/// anyway when the last handle goes away.
|
||||
///
|
||||
/// `JoinHandle::abort` only *requests* cancellation, and dropping a handle
|
||||
/// detaches rather than cancels, so neither on its own says when a task has
|
||||
/// actually stopped. That matters because these tasks hold a store handle:
|
||||
/// until they are gone the `RocksDB` lock is still held and a restarting
|
||||
/// sequencer cannot reopen its home directory. [`TaskGroup::shutdown`] is the
|
||||
/// answer to "have they stopped yet"; the `Drop` below stays as the best-effort
|
||||
/// path for panics and tests that never call it.
|
||||
///
|
||||
/// Cloneable so the owner can keep it (tying task lifetime to its own) while a
|
||||
/// shutdown path elsewhere holds a clone.
|
||||
#[derive(Clone, Default)]
|
||||
pub struct TaskGroup(Arc<TaskGroupInner>);
|
||||
|
||||
#[derive(Default)]
|
||||
struct TaskGroupInner(Mutex<Vec<JoinHandle<()>>>);
|
||||
|
||||
/// A weak handle to the store, for observing when it is finally closed.
|
||||
///
|
||||
/// Every strong reference lives inside a task or a server that shutdown stops,
|
||||
/// but the last drop runs on whichever thread owned it, not on the one awaiting
|
||||
/// shutdown. Watching the count is the difference between knowing the database
|
||||
/// file is closed and assuming it from another crate's drop order.
|
||||
pub struct StoreRelease(Weak<RocksDBIO>);
|
||||
|
||||
impl StoreRelease {
|
||||
#[must_use]
|
||||
pub fn new(store: &Arc<RocksDBIO>) -> Self {
|
||||
Self(Arc::downgrade(store))
|
||||
}
|
||||
|
||||
/// How many holders are left. Zero means the store is closed.
|
||||
#[must_use]
|
||||
pub fn holders(&self) -> usize {
|
||||
self.0.strong_count()
|
||||
}
|
||||
}
|
||||
|
||||
impl Drop for TaskGroupInner {
|
||||
fn drop(&mut self) {
|
||||
for task in Self::take(&self.0) {
|
||||
task.abort();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl TaskGroupInner {
|
||||
/// Empties the handle list, so a second shutdown (or a drop after one) is a
|
||||
/// no-op rather than a second abort.
|
||||
fn handles(handles: &Mutex<Vec<JoinHandle<()>>>) -> MutexGuard<'_, Vec<JoinHandle<()>>> {
|
||||
handles.lock().unwrap_or_else(PoisonError::into_inner)
|
||||
}
|
||||
|
||||
fn take(handles: &Mutex<Vec<JoinHandle<()>>>) -> Vec<JoinHandle<()>> {
|
||||
std::mem::take(&mut *Self::handles(handles))
|
||||
}
|
||||
}
|
||||
|
||||
impl TaskGroup {
|
||||
/// Takes ownership of already-spawned tasks.
|
||||
#[must_use]
|
||||
pub fn new(handles: Vec<JoinHandle<()>>) -> Self {
|
||||
Self(Arc::new(TaskGroupInner(Mutex::new(handles))))
|
||||
}
|
||||
|
||||
/// Whether any task has ended on its own.
|
||||
///
|
||||
/// These tasks run for the lifetime of the sequencer, so a finished one is a
|
||||
/// task that panicked, and whatever it was doing is not happening any more.
|
||||
#[must_use]
|
||||
pub fn any_finished(&self) -> bool {
|
||||
TaskGroupInner::handles(&self.0.0)
|
||||
.iter()
|
||||
.any(JoinHandle::is_finished)
|
||||
}
|
||||
|
||||
/// Stops every task and waits for it to finish.
|
||||
///
|
||||
/// Returns only once the runtime has dropped each task's future, so whatever
|
||||
/// they held (a store handle, a network client) is released by the time this
|
||||
/// returns. Cancellation is the expected outcome, so it is not reported; a
|
||||
/// panic is, since it means the task died on its own terms earlier.
|
||||
pub async fn shutdown(&self) {
|
||||
let handles = TaskGroupInner::take(&self.0.0);
|
||||
for handle in handles {
|
||||
handle.abort();
|
||||
if let Err(err) = handle.await
|
||||
&& err.is_panic()
|
||||
{
|
||||
warn!("Background task panicked before shutdown: {err}");
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[tokio::test]
|
||||
async fn a_task_that_ends_on_its_own_is_visible() {
|
||||
let group = TaskGroup::new(vec![tokio::spawn(async {})]);
|
||||
// A watcher only ends by panicking, so "finished" is the signal that a
|
||||
// peer's deliveries have stopped happening.
|
||||
tokio::task::yield_now().await;
|
||||
assert!(group.any_finished());
|
||||
|
||||
let running = TaskGroup::new(vec![tokio::spawn(std::future::pending())]);
|
||||
assert!(!running.any_finished());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn shutdown_ends_a_task_that_would_never_end_on_its_own() {
|
||||
let group = TaskGroup::new(vec![tokio::spawn(std::future::pending())]);
|
||||
|
||||
// The watchers and the drive task are infinite loops, so awaiting one
|
||||
// without cancelling it first hangs here for ever.
|
||||
tokio::time::timeout(std::time::Duration::from_secs(5), group.shutdown())
|
||||
.await
|
||||
.expect("shutdown must not hang on a task that never finishes by itself");
|
||||
|
||||
// Shutting down twice is a no-op rather than a second abort.
|
||||
tokio::time::timeout(std::time::Duration::from_secs(5), group.shutdown())
|
||||
.await
|
||||
.expect("a second shutdown must return immediately");
|
||||
}
|
||||
}
|
||||
@ -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();
|
||||
|
||||
@ -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
|
||||
|
||||
@ -12,7 +12,11 @@ use sequencer_core::SequencerCore;
|
||||
#[cfg(feature = "standalone")]
|
||||
use sequencer_core::SequencerCoreWithMockClients as SequencerCore;
|
||||
pub use sequencer_core::config::*;
|
||||
use sequencer_core::{TransactionOrigin, block_publisher::BlockPublisherTrait as _};
|
||||
use sequencer_core::{
|
||||
TransactionOrigin,
|
||||
block_publisher::BlockPublisherTrait as _,
|
||||
task_group::{StoreRelease, TaskGroup},
|
||||
};
|
||||
use sequencer_service_rpc::RpcServer as _;
|
||||
use tokio::{sync::Mutex, task::JoinHandle};
|
||||
use tokio_util::sync::CancellationToken;
|
||||
@ -26,12 +30,19 @@ const REQUEST_BODY_MAX_SIZE: ByteSize = ByteSize::mib(10);
|
||||
/// Implements `Drop` to ensure all tasks are aborted and the RPC server is stopped when dropped.
|
||||
pub struct SequencerHandle {
|
||||
addr: SocketAddr,
|
||||
/// Option because of `Drop` which forbids to simply move out of `self` in `stopped()`.
|
||||
server_handle: Option<ServerHandle>,
|
||||
server_handle: ServerHandle,
|
||||
main_loop_handle: JoinHandle<Result<Never>>,
|
||||
/// Cancelled when the publisher's drive task terminates (e.g. a panicked
|
||||
/// persist sink); no channel events are processed past that point.
|
||||
driver_cancellation: CancellationToken,
|
||||
/// The core's background tasks, taken before the core was shared. This
|
||||
/// handle owns no reference to the core itself, so without these there is
|
||||
/// nothing to wait on: aborting the main loop only starts the teardown.
|
||||
background_tasks: Vec<TaskGroup>,
|
||||
/// The store, weakly. Every strong reference lives inside something this
|
||||
/// handle stops, so watching the count go to zero is how shutdown knows the
|
||||
/// database file is actually closed rather than assuming it from drop order.
|
||||
store: StoreRelease,
|
||||
}
|
||||
|
||||
impl SequencerHandle {
|
||||
@ -40,29 +51,73 @@ impl SequencerHandle {
|
||||
server_handle: ServerHandle,
|
||||
main_loop_handle: JoinHandle<Result<Never>>,
|
||||
driver_cancellation: CancellationToken,
|
||||
background_tasks: Vec<TaskGroup>,
|
||||
store: StoreRelease,
|
||||
) -> Self {
|
||||
Self {
|
||||
addr,
|
||||
server_handle: Some(server_handle),
|
||||
server_handle,
|
||||
main_loop_handle,
|
||||
driver_cancellation,
|
||||
background_tasks,
|
||||
store,
|
||||
}
|
||||
}
|
||||
|
||||
/// Stops the sequencer and waits for every part of it to be gone.
|
||||
///
|
||||
/// `Drop` alone cannot do this: it aborts the main loop without awaiting it,
|
||||
/// and the core lives behind `Arc`s held by that task and the RPC server, so
|
||||
/// after a plain drop the store is still open for an unbounded stretch. That
|
||||
/// is why restarting a sequencer on the same home directory used to need a
|
||||
/// sleep, and why an in-process restart could fail outright with a `RocksDB`
|
||||
/// lock error.
|
||||
///
|
||||
/// Order matters: the main loop stops first so nothing new is produced while
|
||||
/// the publisher is torn down, then the background tasks that hold the store,
|
||||
/// then the server. Consuming `self` drops the last references, so the store
|
||||
/// is closed by the time this returns.
|
||||
pub async fn shutdown(mut self) {
|
||||
self.main_loop_handle.abort();
|
||||
if let Err(err) = (&mut self.main_loop_handle).await
|
||||
&& err.is_panic()
|
||||
{
|
||||
error!("Sequencer main loop panicked before shutdown: {err}");
|
||||
}
|
||||
|
||||
for tasks in &self.background_tasks {
|
||||
tasks.shutdown().await;
|
||||
}
|
||||
|
||||
if let Err(err) = self.server_handle.stop() {
|
||||
error!("An error occurred while stopping Sequencer RPC server: {err}");
|
||||
}
|
||||
self.server_handle.clone().stopped().await;
|
||||
|
||||
// Nothing this handle owns holds the store, so waiting here rather than
|
||||
// after the drop is the same thing, and it keeps the guarantee inside
|
||||
// the call the caller awaits.
|
||||
wait_for_store_release(&self.store).await;
|
||||
}
|
||||
|
||||
/// Wait for any of the sequencer tasks to fail and return the error.
|
||||
#[expect(
|
||||
clippy::integer_division_remainder_used,
|
||||
reason = "Generated by select! macro, can't be easily rewritten to avoid this lint"
|
||||
)]
|
||||
pub async fn failed(mut self) -> Result<Never> {
|
||||
pub async fn failed(&mut self) -> Result<Never> {
|
||||
let Self {
|
||||
addr: _,
|
||||
server_handle,
|
||||
main_loop_handle,
|
||||
driver_cancellation,
|
||||
} = &mut self;
|
||||
background_tasks: _,
|
||||
store: _,
|
||||
} = self;
|
||||
|
||||
let server_handle = server_handle.take().expect("Server handle is set");
|
||||
// Cloned rather than taken: `stopped()` consumes a handle, and taking
|
||||
// this one would leave `shutdown` with no way to stop the server.
|
||||
let server_handle = server_handle.clone();
|
||||
tokio::select! {
|
||||
() = server_handle.stopped() => {
|
||||
Err(anyhow!("RPC Server stopped"))
|
||||
@ -89,11 +144,16 @@ impl SequencerHandle {
|
||||
server_handle,
|
||||
main_loop_handle,
|
||||
driver_cancellation,
|
||||
background_tasks,
|
||||
store: _,
|
||||
} = self;
|
||||
|
||||
let stopped = server_handle.as_ref().is_none_or(ServerHandle::is_stopped)
|
||||
let stopped = server_handle.is_stopped()
|
||||
|| main_loop_handle.is_finished()
|
||||
|| driver_cancellation.is_cancelled();
|
||||
|| driver_cancellation.is_cancelled()
|
||||
// A watcher only ends by panicking, and a peer whose deliveries have
|
||||
// silently stopped is exactly what this predicate exists to catch.
|
||||
|| background_tasks.iter().any(TaskGroup::any_finished);
|
||||
!stopped
|
||||
}
|
||||
|
||||
@ -110,20 +170,46 @@ impl Drop for SequencerHandle {
|
||||
server_handle,
|
||||
main_loop_handle,
|
||||
driver_cancellation: _,
|
||||
background_tasks: _,
|
||||
store: _,
|
||||
} = self;
|
||||
|
||||
main_loop_handle.abort();
|
||||
|
||||
let Some(handle) = server_handle else {
|
||||
return;
|
||||
};
|
||||
|
||||
if let Err(err) = handle.stop() {
|
||||
if let Err(err) = server_handle.stop() {
|
||||
error!("An error occurred while stopping Sequencer RPC server: {err}");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Waits until nothing holds the store any more.
|
||||
///
|
||||
/// Everything that holds one lives inside a task or a server this handle has
|
||||
/// already stopped, but the last drop happens on whichever thread ran them, not
|
||||
/// on this one. Without this the caller can reopen the database a moment too
|
||||
/// early and hit a `RocksDB` lock error, which is the kind of failure that shows
|
||||
/// up as an occasional flake rather than a bug.
|
||||
async fn wait_for_store_release(store: &StoreRelease) {
|
||||
/// Long enough for a drop that is already in flight, short enough that a
|
||||
/// leak is reported rather than hung on.
|
||||
const RELEASE_TIMEOUT: Duration = Duration::from_secs(10);
|
||||
const POLL: Duration = Duration::from_millis(10);
|
||||
|
||||
let released = tokio::time::timeout(RELEASE_TIMEOUT, async {
|
||||
while store.holders() > 0 {
|
||||
tokio::time::sleep(POLL).await;
|
||||
}
|
||||
})
|
||||
.await;
|
||||
|
||||
if released.is_err() {
|
||||
error!(
|
||||
"Sequencer store still held by {} reference(s) after shutdown; something outlived the tasks it should have died with",
|
||||
store.holders()
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn run(config: SequencerConfig, listen_addr: SocketAddr) -> Result<SequencerHandle> {
|
||||
let block_timeout = config.block_create_timeout;
|
||||
let max_block_size = config.max_block_size;
|
||||
@ -134,6 +220,11 @@ pub async fn run(config: SequencerConfig, listen_addr: SocketAddr) -> Result<Seq
|
||||
info!("Sequencer core set up");
|
||||
|
||||
let driver_cancellation = sequencer_core.block_publisher().driver_cancellation();
|
||||
// Taken while the core is still owned here: once it is behind the `Arc`
|
||||
// below, the only owners are the RPC server and the main loop task, and
|
||||
// neither hands it back.
|
||||
let background_tasks = sequencer_core.background_tasks();
|
||||
let store = sequencer_core.store_release();
|
||||
let seq_core_wrapped = Arc::new(Mutex::new(sequencer_core));
|
||||
let mempool_handle_for_server = mempool_handle.clone();
|
||||
|
||||
@ -156,6 +247,8 @@ pub async fn run(config: SequencerConfig, listen_addr: SocketAddr) -> Result<Seq
|
||||
server_handle,
|
||||
main_loop_handle,
|
||||
driver_cancellation,
|
||||
background_tasks,
|
||||
store,
|
||||
))
|
||||
}
|
||||
|
||||
|
||||
@ -6,6 +6,7 @@ use std::{
|
||||
use anyhow::Result;
|
||||
use clap::Parser;
|
||||
use log::{error, info};
|
||||
use tokio::signal::unix::{SignalKind, signal};
|
||||
use tokio_util::sync::CancellationToken;
|
||||
|
||||
#[derive(Debug, Parser)]
|
||||
@ -40,15 +41,13 @@ async fn main() -> Result<()> {
|
||||
home,
|
||||
} = Args::parse();
|
||||
|
||||
// TODO: handle this cancellation token more gracefully within Sequencer service
|
||||
// similar to how we do in Indexer
|
||||
let cancellation_token = listen_for_shutdown_signal();
|
||||
|
||||
let mut config = sequencer_service::SequencerConfig::from_path(&config_path)?;
|
||||
if let Some(home) = home {
|
||||
config.home = home;
|
||||
}
|
||||
let sequencer_handle =
|
||||
let mut sequencer_handle =
|
||||
sequencer_service::run(config, SocketAddr::new(listen_address, port)).await?;
|
||||
|
||||
tokio::select! {
|
||||
@ -60,21 +59,50 @@ async fn main() -> Result<()> {
|
||||
}
|
||||
}
|
||||
|
||||
// Stop the watchers, the publisher's drive task, the block loop and the RPC
|
||||
// server, and wait for each. Dropping the handle only asks; the store stays
|
||||
// open for an unbounded stretch after that, so a restart can find its own
|
||||
// home directory locked, and a watcher can be killed between recording a
|
||||
// delivery and handing it over.
|
||||
sequencer_handle.shutdown().await;
|
||||
|
||||
info!("Sequencer shutdown complete");
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Cancelled on Ctrl-C or `SIGTERM`.
|
||||
///
|
||||
/// `SIGTERM` is what a container runtime sends first, so without it every
|
||||
/// orchestrated stop is the ungraceful path.
|
||||
#[expect(
|
||||
clippy::integer_division_remainder_used,
|
||||
reason = "Generated by select! macro, can't be easily rewritten to avoid this lint"
|
||||
)]
|
||||
fn listen_for_shutdown_signal() -> CancellationToken {
|
||||
let cancellation_token = CancellationToken::new();
|
||||
let cancellation_token_clone = cancellation_token.clone();
|
||||
|
||||
tokio::spawn(async move {
|
||||
if let Err(err) = tokio::signal::ctrl_c().await {
|
||||
error!("Failed to listen for Ctrl-C signal: {err}");
|
||||
return;
|
||||
let mut terminate = match signal(SignalKind::terminate()) {
|
||||
Ok(terminate) => terminate,
|
||||
Err(err) => {
|
||||
error!("Failed to listen for SIGTERM: {err}");
|
||||
return;
|
||||
}
|
||||
};
|
||||
|
||||
tokio::select! {
|
||||
result = tokio::signal::ctrl_c() => match result {
|
||||
Ok(()) => info!("Received Ctrl-C signal"),
|
||||
Err(err) => {
|
||||
error!("Failed to listen for Ctrl-C signal: {err}");
|
||||
return;
|
||||
}
|
||||
},
|
||||
_ = terminate.recv() => info!("Received SIGTERM"),
|
||||
}
|
||||
info!("Received Ctrl-C signal");
|
||||
|
||||
cancellation_token_clone.cancel();
|
||||
});
|
||||
|
||||
|
||||
@ -1,4 +1,8 @@
|
||||
use std::{collections::BTreeMap, path::Path, sync::Arc};
|
||||
use std::{
|
||||
collections::BTreeMap,
|
||||
path::Path,
|
||||
sync::{Arc, Mutex, MutexGuard, PoisonError},
|
||||
};
|
||||
|
||||
use borsh::{BorshDeserialize, BorshSerialize};
|
||||
use common::{
|
||||
@ -18,7 +22,9 @@ use crate::{
|
||||
sequencer::sequencer_cells::{
|
||||
FinalBlockMetaCellOwned, FinalBlockMetaCellRef, FinalLeeStateCellOwned,
|
||||
FinalLeeStateCellRef, LEEStateCellOwned, LEEStateCellRef, LastFinalizedBlockIdCell,
|
||||
LatestBlockMetaCellOwned, LatestBlockMetaCellRef, PendingDepositEventRecord,
|
||||
LatestBlockMetaCellOwned, LatestBlockMetaCellRef, PeerFloorCellOwned, PeerFloorCellRef,
|
||||
PeerZoneKey, PendingCrossZoneDispatchRecord, PendingCrossZoneDispatchesCellOwned,
|
||||
PendingCrossZoneDispatchesCellRef, PendingDepositEventRecord,
|
||||
PendingDepositEventsCellOwned, PendingDepositEventsCellRef, UnseenWithdrawCountCell,
|
||||
WithdrawalReconciliationKey, ZoneAnchorCell, ZoneAnchorRecord, ZoneSdkCheckpointCellOwned,
|
||||
ZoneSdkCheckpointCellRef,
|
||||
@ -40,9 +46,25 @@ pub const DB_META_ZONE_CURSOR_KEY: &str = "zone_cursor";
|
||||
/// Key base for storing queued deposit events that were not yet
|
||||
/// fulfilled on L2.
|
||||
pub const DB_META_PENDING_DEPOSIT_EVENTS_KEY: &str = "pending_deposit_events";
|
||||
/// Key base for storing a cross-zone watcher's delivery floor on one peer
|
||||
/// channel (opaque bytes). Keyed per peer zone.
|
||||
pub const DB_META_CROSS_ZONE_PEER_FLOOR_KEY: &str = "cross_zone_peer_floor";
|
||||
/// 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 counting unseen L2 withdraw intents.
|
||||
pub const DB_META_UNSEEN_WITHDRAW_COUNT_KEY: &str = "unseen_withdraw_count";
|
||||
|
||||
/// How many cross-zone deliveries may be pending at once.
|
||||
///
|
||||
/// The whole list is a single value, read on every block and rewritten on every
|
||||
/// change, and what fills it is chosen by peer zones rather than by us. Refusing
|
||||
/// to record past this bound turns "a peer decides how large our store gets"
|
||||
/// into "a peer's messages wait", since a watcher that cannot record holds its
|
||||
/// delivery floor and reads the slot again later.
|
||||
pub const MAX_PENDING_CROSS_ZONE_DISPATCHES: usize = 4096;
|
||||
|
||||
/// 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.
|
||||
@ -123,6 +145,8 @@ pub struct StoreUpdate<'update> {
|
||||
pub new_deposit_events: &'update [PendingDepositEventRecord],
|
||||
/// Deposit op ids whose mint finalized: their pending records are dropped.
|
||||
pub remove_deposit_records: &'update [HashType],
|
||||
/// Message keys whose delivery finalized: their pending records are dropped.
|
||||
pub remove_dispatch_records: &'update [[u8; 32]],
|
||||
/// L1 withdraw events to reconcile against the local unseen counters.
|
||||
pub consumed_withdrawals: &'update [WithdrawalReconciliationKey],
|
||||
/// L2 withdraw intents this update raises, awaiting their L1 event.
|
||||
@ -146,6 +170,7 @@ impl<'update> StoreUpdate<'update> {
|
||||
finalized_up_to: None,
|
||||
new_deposit_events: &[],
|
||||
remove_deposit_records: &[],
|
||||
remove_dispatch_records: &[],
|
||||
consumed_withdrawals: &[],
|
||||
new_withdraw_intents: &[],
|
||||
zone_anchor: None,
|
||||
@ -165,8 +190,21 @@ pub struct StoreUpdateOutcome {
|
||||
pub unmatched_withdrawals: Vec<WithdrawalReconciliationKey>,
|
||||
}
|
||||
|
||||
#[expect(
|
||||
clippy::partial_pub_fields,
|
||||
reason = "the pending-record lock is an implementation detail and must stay private"
|
||||
)]
|
||||
pub struct RocksDBIO {
|
||||
pub db: DBWithThreadMode<MultiThreaded>,
|
||||
/// Serializes the read-modify-write cycles over the pending cross-zone
|
||||
/// dispatch list.
|
||||
///
|
||||
/// The list is a single value holding the whole `Vec`, and three tasks
|
||||
/// rewrite it: the watcher recording a delivery, the production loop
|
||||
/// counting a failed attempt, and the publisher's drive task settling
|
||||
/// finalized deliveries. Rocksdb makes the write atomic, not the cycle, so
|
||||
/// without this the writer that read first silently drops the others.
|
||||
pending_records: Mutex<()>,
|
||||
}
|
||||
|
||||
impl DBIO for RocksDBIO {
|
||||
@ -176,6 +214,18 @@ impl DBIO for RocksDBIO {
|
||||
}
|
||||
|
||||
impl RocksDBIO {
|
||||
/// Held across a pending-record read-modify-write. See
|
||||
/// [`RocksDBIO::pending_records`].
|
||||
///
|
||||
/// A poisoned lock is recovered rather than propagated: the records behind
|
||||
/// it are a plain `Vec` that a panicking writer cannot leave half-written,
|
||||
/// since the write is a single rocksdb put.
|
||||
fn lock_pending_records(&self) -> MutexGuard<'_, ()> {
|
||||
self.pending_records
|
||||
.lock()
|
||||
.unwrap_or_else(PoisonError::into_inner)
|
||||
}
|
||||
|
||||
pub fn open(path: &Path) -> DbResult<Self> {
|
||||
let db_opts = Options::default();
|
||||
Self::open_inner(path, &db_opts)
|
||||
@ -289,7 +339,10 @@ impl RocksDBIO {
|
||||
additional_info: Some("Failed to open or create DB".to_owned()),
|
||||
})?;
|
||||
|
||||
let dbio = Self { db };
|
||||
let dbio = Self {
|
||||
db,
|
||||
pending_records: Mutex::new(()),
|
||||
};
|
||||
Ok(dbio)
|
||||
}
|
||||
|
||||
@ -539,6 +592,185 @@ impl RocksDBIO {
|
||||
Ok(accepted)
|
||||
}
|
||||
|
||||
/// One cross-zone watcher's delivery floor on `peer_zone`'s channel, or
|
||||
/// `None` before it has delivered anything from that peer.
|
||||
pub fn get_cross_zone_peer_floor_bytes(
|
||||
&self,
|
||||
peer_zone: PeerZoneKey,
|
||||
) -> DbResult<Option<Vec<u8>>> {
|
||||
Ok(self
|
||||
.get_opt::<PeerFloorCellOwned>(peer_zone)?
|
||||
.map(|cell| cell.0))
|
||||
}
|
||||
|
||||
pub fn put_cross_zone_peer_floor_bytes(
|
||||
&self,
|
||||
peer_zone: PeerZoneKey,
|
||||
bytes: &[u8],
|
||||
) -> DbResult<()> {
|
||||
self.put(&PeerFloorCellRef(bytes), peer_zone)
|
||||
}
|
||||
|
||||
pub fn get_pending_cross_zone_dispatches(
|
||||
&self,
|
||||
) -> DbResult<Vec<PendingCrossZoneDispatchRecord>> {
|
||||
Ok(self
|
||||
.get_opt::<PendingCrossZoneDispatchesCellOwned>(())?
|
||||
.map_or_else(Vec::new, |cell| cell.0))
|
||||
}
|
||||
|
||||
fn put_pending_cross_zone_dispatches(
|
||||
&self,
|
||||
records: &[PendingCrossZoneDispatchRecord],
|
||||
) -> DbResult<()> {
|
||||
self.put(&PendingCrossZoneDispatchesCellRef(records), ())
|
||||
}
|
||||
|
||||
fn put_pending_cross_zone_dispatches_batch(
|
||||
&self,
|
||||
records: &[PendingCrossZoneDispatchRecord],
|
||||
batch: &mut WriteBatch,
|
||||
) -> DbResult<()> {
|
||||
self.put_batch(&PendingCrossZoneDispatchesCellRef(records), (), batch)
|
||||
}
|
||||
|
||||
/// Records every delivery one peer block carries, in a single write.
|
||||
///
|
||||
/// Returns how many were new. Ones already recorded are skipped, so a slot
|
||||
/// the watcher re-reads is not double-tracked.
|
||||
///
|
||||
/// Batched rather than one call per delivery because the whole list is one
|
||||
/// value: recording a block's messages one at a time rewrites the list once
|
||||
/// per message, which is quadratic in a block that carries many.
|
||||
///
|
||||
/// Fails without writing anything if the list would exceed
|
||||
/// [`MAX_PENDING_CROSS_ZONE_DISPATCHES`]. The caller's floor then stays put
|
||||
/// and the slot is read again later, which is the difference between
|
||||
/// backpressure and an unbounded list a peer controls the size of.
|
||||
pub fn add_pending_cross_zone_dispatches(
|
||||
&self,
|
||||
dispatches: Vec<PendingCrossZoneDispatchRecord>,
|
||||
) -> DbResult<usize> {
|
||||
if dispatches.is_empty() {
|
||||
return Ok(0);
|
||||
}
|
||||
|
||||
let _pending = self.lock_pending_records();
|
||||
let mut records = self.get_pending_cross_zone_dispatches()?;
|
||||
let before = records.len();
|
||||
|
||||
for dispatch in dispatches {
|
||||
if records
|
||||
.iter()
|
||||
.any(|record| record.message_key == dispatch.message_key)
|
||||
{
|
||||
continue;
|
||||
}
|
||||
records.push(dispatch);
|
||||
}
|
||||
|
||||
let accepted = records.len().saturating_sub(before);
|
||||
if accepted == 0 {
|
||||
return Ok(0);
|
||||
}
|
||||
if records.len() > MAX_PENDING_CROSS_ZONE_DISPATCHES {
|
||||
return Err(DbError::db_interaction_error(format!(
|
||||
"Refusing to hold more than {MAX_PENDING_CROSS_ZONE_DISPATCHES} pending cross-zone deliveries; {before} already pending"
|
||||
)));
|
||||
}
|
||||
|
||||
self.put_pending_cross_zone_dispatches(&records)?;
|
||||
Ok(accepted)
|
||||
}
|
||||
|
||||
/// Counts a failed production attempt against a delivery, dropping its
|
||||
/// record once it reaches `retire_at`. Returns whether it was dropped.
|
||||
///
|
||||
/// 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.
|
||||
///
|
||||
/// A delivery with no record is already retired as far as this is concerned:
|
||||
/// there is nothing left to count against.
|
||||
pub fn record_dispatch_failure(&self, message_key: [u8; 32], retire_at: u32) -> DbResult<bool> {
|
||||
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);
|
||||
};
|
||||
|
||||
let 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);
|
||||
}
|
||||
self.put_pending_cross_zone_dispatches(&records)?;
|
||||
Ok(retired)
|
||||
}
|
||||
|
||||
/// Drops the records of deliveries that are settled for good, outside any
|
||||
/// store update.
|
||||
///
|
||||
/// The settlement path in [`Self::store_update`] catches a delivery as its
|
||||
/// block becomes irreversible. This catches the ones that path cannot: a
|
||||
/// record re-added after its delivery had already settled, which the watcher
|
||||
/// does whenever it re-reads a slot it has already consumed. Nothing would
|
||||
/// ever put such a key in a block again, so without this it stays for ever.
|
||||
pub fn drop_settled_cross_zone_dispatches(&self, message_keys: &[[u8; 32]]) -> DbResult<usize> {
|
||||
if message_keys.is_empty() {
|
||||
return Ok(0);
|
||||
}
|
||||
|
||||
let _pending = self.lock_pending_records();
|
||||
let to_remove: std::collections::HashSet<&[u8; 32]> = message_keys.iter().collect();
|
||||
let mut records = self.get_pending_cross_zone_dispatches()?;
|
||||
let before = records.len();
|
||||
records.retain(|record| !to_remove.contains(&record.message_key));
|
||||
let removed = before.saturating_sub(records.len());
|
||||
|
||||
if removed > 0 {
|
||||
self.put_pending_cross_zone_dispatches(&records)?;
|
||||
}
|
||||
Ok(removed)
|
||||
}
|
||||
|
||||
/// Drops the pending records of deliveries that just became irreversible,
|
||||
/// staged into `batch` so they go with the update that made them so.
|
||||
///
|
||||
/// Removal only, unlike [`Self::stage_pending_deposit_events`]: a delivery is
|
||||
/// recorded by the watcher through
|
||||
/// [`Self::add_pending_cross_zone_dispatch`], on its own task and outside
|
||||
/// any store update, so nothing ever adds one here.
|
||||
fn stage_removed_dispatches(
|
||||
&self,
|
||||
remove_keys: &[[u8; 32]],
|
||||
batch: &mut WriteBatch,
|
||||
) -> DbResult<usize> {
|
||||
if remove_keys.is_empty() {
|
||||
return Ok(0);
|
||||
}
|
||||
|
||||
let to_remove: std::collections::HashSet<&[u8; 32]> = remove_keys.iter().collect();
|
||||
let mut records = self.get_pending_cross_zone_dispatches()?;
|
||||
let before = records.len();
|
||||
records.retain(|record| !to_remove.contains(&record.message_key));
|
||||
let removed = before.saturating_sub(records.len());
|
||||
|
||||
if removed > 0 {
|
||||
self.put_pending_cross_zone_dispatches_batch(&records, batch)?;
|
||||
}
|
||||
Ok(removed)
|
||||
}
|
||||
|
||||
/// Stages the unseen-withdraw decrements for one update into `batch`,
|
||||
/// returning one entry per occurrence that matched no local counter.
|
||||
///
|
||||
@ -875,6 +1107,7 @@ impl RocksDBIO {
|
||||
/// one, most carry nothing else) must not drag a full state serialization
|
||||
/// with it.
|
||||
pub fn store_update(&self, update: &StoreUpdate<'_>) -> DbResult<StoreUpdateOutcome> {
|
||||
let _pending = self.lock_pending_records();
|
||||
let StoreUpdate {
|
||||
checkpoint,
|
||||
blocks,
|
||||
@ -884,6 +1117,7 @@ impl RocksDBIO {
|
||||
finalized_up_to,
|
||||
new_deposit_events,
|
||||
remove_deposit_records,
|
||||
remove_dispatch_records,
|
||||
consumed_withdrawals,
|
||||
new_withdraw_intents,
|
||||
zone_anchor,
|
||||
@ -940,6 +1174,7 @@ impl RocksDBIO {
|
||||
remove_deposit_records,
|
||||
&mut batch,
|
||||
)?;
|
||||
self.stage_removed_dispatches(remove_dispatch_records, &mut batch)?;
|
||||
let unmatched_withdrawals =
|
||||
self.stage_consumed_withdrawals(consumed_withdrawals, &mut batch)?;
|
||||
self.stage_new_withdraw_intents(new_withdraw_intents, &mut batch)?;
|
||||
|
||||
@ -8,7 +8,8 @@ use crate::{
|
||||
error::DbError,
|
||||
sequencer::{
|
||||
CF_LEE_STATE_NAME, DB_FINAL_BLOCK_META_KEY, DB_FINAL_LEE_STATE_KEY, DB_LEE_STATE_KEY,
|
||||
DB_META_LAST_FINALIZED_BLOCK_ID, DB_META_LATEST_BLOCK_META_KEY,
|
||||
DB_META_CROSS_ZONE_PEER_FLOOR_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_UNSEEN_WITHDRAW_COUNT_KEY,
|
||||
DB_META_ZONE_CURSOR_KEY, DB_META_ZONE_SDK_CHECKPOINT_KEY,
|
||||
},
|
||||
@ -245,6 +246,82 @@ pub struct PendingDepositEventRecord {
|
||||
pub metadata: Vec<u8>,
|
||||
}
|
||||
|
||||
/// A cross-zone delivery the watcher has read off a peer block but which is not
|
||||
/// yet known to be irreversibly delivered.
|
||||
///
|
||||
/// The watcher's delivery floor is durable, so once it advances past a peer
|
||||
/// block that block is never re-read. This record is what stands in its place:
|
||||
/// block production drains it every turn, and it survives a restart. Mirrors
|
||||
/// [`PendingDepositEventRecord`], which solves the same problem for deposits,
|
||||
/// and like it carries no "submitted" mark: the record is dropped when the
|
||||
/// delivery itself finalizes, and re-including one meanwhile is harmless
|
||||
/// because the inbox no-ops a replay on chain.
|
||||
#[derive(Debug, Clone, PartialEq, Eq, BorshSerialize, BorshDeserialize)]
|
||||
pub struct PendingCrossZoneDispatchRecord {
|
||||
/// Content-addressed replay key of the delivered message, and this record's
|
||||
/// identity.
|
||||
pub message_key: [u8; 32],
|
||||
/// The borsh-encoded dispatch transaction, so production can re-feed it
|
||||
/// without re-reading the peer channel.
|
||||
pub transaction: Vec<u8>,
|
||||
/// Production attempts that ended in an execution failure.
|
||||
///
|
||||
/// 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.
|
||||
pub failed_attempts: u32,
|
||||
}
|
||||
|
||||
impl PendingCrossZoneDispatchRecord {
|
||||
/// A delivery the watcher has just read: never attempted.
|
||||
#[must_use]
|
||||
pub const fn recorded(message_key: [u8; 32], transaction: Vec<u8>) -> Self {
|
||||
Self {
|
||||
message_key,
|
||||
transaction,
|
||||
failed_attempts: 0,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(BorshDeserialize)]
|
||||
pub struct PendingCrossZoneDispatchesCellOwned(pub Vec<PendingCrossZoneDispatchRecord>);
|
||||
|
||||
impl SimpleStorableCell for PendingCrossZoneDispatchesCellOwned {
|
||||
type KeyParams = ();
|
||||
|
||||
const CELL_NAME: &'static str = DB_META_PENDING_CROSS_ZONE_DISPATCHES_KEY;
|
||||
const CF_NAME: &'static str = CF_META_NAME;
|
||||
}
|
||||
|
||||
impl SimpleReadableCell for PendingCrossZoneDispatchesCellOwned {}
|
||||
|
||||
#[derive(BorshSerialize)]
|
||||
pub struct PendingCrossZoneDispatchesCellRef<'records>(
|
||||
pub &'records [PendingCrossZoneDispatchRecord],
|
||||
);
|
||||
|
||||
impl SimpleStorableCell for PendingCrossZoneDispatchesCellRef<'_> {
|
||||
type KeyParams = ();
|
||||
|
||||
const CELL_NAME: &'static str = DB_META_PENDING_CROSS_ZONE_DISPATCHES_KEY;
|
||||
const CF_NAME: &'static str = CF_META_NAME;
|
||||
}
|
||||
|
||||
impl SimpleWritableCell for PendingCrossZoneDispatchesCellRef<'_> {
|
||||
fn value_constructor(&self) -> DbResult<Vec<u8>> {
|
||||
borsh::to_vec(&self).map_err(|err| {
|
||||
DbError::borsh_cast_message(
|
||||
err,
|
||||
Some("Failed to serialize pending cross-zone dispatches cell".to_owned()),
|
||||
)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(BorshDeserialize)]
|
||||
pub struct PendingDepositEventsCellOwned(pub Vec<PendingDepositEventRecord>);
|
||||
|
||||
@ -278,6 +355,72 @@ impl SimpleWritableCell for PendingDepositEventsCellRef<'_> {
|
||||
}
|
||||
}
|
||||
|
||||
/// Identifies which peer channel a cross-zone watcher cursor belongs to. The
|
||||
/// 32-byte peer channel id doubles as the peer's zone id.
|
||||
pub type PeerZoneKey = [u8; 32];
|
||||
|
||||
/// Opaque bytes for one peer's cross-zone read cursor. As with the zone-sdk
|
||||
/// checkpoint, the caller owns the encoding, since the cursor type derives serde
|
||||
/// rather than borsh.
|
||||
#[derive(BorshDeserialize)]
|
||||
pub struct PeerFloorCellOwned(pub Vec<u8>);
|
||||
|
||||
impl SimpleStorableCell for PeerFloorCellOwned {
|
||||
type KeyParams = PeerZoneKey;
|
||||
|
||||
const CELL_NAME: &'static str = DB_META_CROSS_ZONE_PEER_FLOOR_KEY;
|
||||
const CF_NAME: &'static str = CF_META_NAME;
|
||||
|
||||
/// Folds the peer zone into the key so each peer keeps its own cursor.
|
||||
fn key_constructor(peer_zone: Self::KeyParams) -> DbResult<Vec<u8>> {
|
||||
borsh::to_vec(&(Self::CELL_NAME, peer_zone)).map_err(|err| {
|
||||
DbError::borsh_cast_message(
|
||||
err,
|
||||
Some(format!(
|
||||
"Failed to serialize {:?} key params",
|
||||
Self::CELL_NAME
|
||||
)),
|
||||
)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
impl SimpleReadableCell for PeerFloorCellOwned {}
|
||||
|
||||
#[derive(BorshSerialize)]
|
||||
pub struct PeerFloorCellRef<'bytes>(pub &'bytes [u8]);
|
||||
|
||||
impl SimpleStorableCell for PeerFloorCellRef<'_> {
|
||||
type KeyParams = PeerZoneKey;
|
||||
|
||||
const CELL_NAME: &'static str = DB_META_CROSS_ZONE_PEER_FLOOR_KEY;
|
||||
const CF_NAME: &'static str = CF_META_NAME;
|
||||
|
||||
/// Folds the peer zone into the key so each peer keeps its own cursor.
|
||||
fn key_constructor(peer_zone: Self::KeyParams) -> DbResult<Vec<u8>> {
|
||||
borsh::to_vec(&(Self::CELL_NAME, peer_zone)).map_err(|err| {
|
||||
DbError::borsh_cast_message(
|
||||
err,
|
||||
Some(format!(
|
||||
"Failed to serialize {:?} key params",
|
||||
Self::CELL_NAME
|
||||
)),
|
||||
)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
impl SimpleWritableCell for PeerFloorCellRef<'_> {
|
||||
fn value_constructor(&self) -> DbResult<Vec<u8>> {
|
||||
borsh::to_vec(&self).map_err(|err| {
|
||||
DbError::borsh_cast_message(
|
||||
err,
|
||||
Some("Failed to serialize cross-zone peer floor cell".to_owned()),
|
||||
)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
|
||||
pub struct WithdrawalReconciliationKey {
|
||||
pub amount: u64,
|
||||
|
||||
@ -37,6 +37,17 @@ fn deposit_record(seed: u8) -> PendingDepositEventRecord {
|
||||
}
|
||||
}
|
||||
|
||||
fn dispatch_record(seed: u8) -> PendingCrossZoneDispatchRecord {
|
||||
PendingCrossZoneDispatchRecord::recorded([seed; 32], vec![seed; 4])
|
||||
}
|
||||
|
||||
/// 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];
|
||||
key[..8].copy_from_slice(&u64::try_from(index).expect("test index fits").to_le_bytes());
|
||||
key
|
||||
}
|
||||
|
||||
fn stored_balance(dbio: &RocksDBIO) -> u128 {
|
||||
dbio.get_lee_state()
|
||||
.unwrap()
|
||||
@ -399,6 +410,171 @@ fn finalized_deposit_records_are_removed_by_op_id() {
|
||||
assert_eq!(stored, vec![second]);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn dispatch_records_round_trip_and_dedupe_by_message_key() {
|
||||
let temp_dir = tempdir().unwrap();
|
||||
let (dbio, _genesis) = dbio_with_genesis(temp_dir.path());
|
||||
|
||||
let record = dispatch_record(1);
|
||||
assert_eq!(
|
||||
dbio.add_pending_cross_zone_dispatches(vec![record.clone()])
|
||||
.unwrap(),
|
||||
1
|
||||
);
|
||||
// The watcher re-reads a slot it stalled on, so the same delivery arrives
|
||||
// again; recording it twice would double-count its failed attempts.
|
||||
assert_eq!(
|
||||
dbio.add_pending_cross_zone_dispatches(vec![record.clone(), dispatch_record(2)])
|
||||
.unwrap(),
|
||||
1,
|
||||
"only the delivery not already held is newly recorded"
|
||||
);
|
||||
|
||||
assert_eq!(
|
||||
dbio.get_pending_cross_zone_dispatches().unwrap(),
|
||||
vec![record, dispatch_record(2)]
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn recording_past_the_cap_writes_nothing() {
|
||||
let temp_dir = tempdir().unwrap();
|
||||
let (dbio, _genesis) = dbio_with_genesis(temp_dir.path());
|
||||
|
||||
// What fills this list is chosen by peer zones, so the bound is what stops a
|
||||
// peer deciding how large our store gets. Refusing the whole write leaves
|
||||
// the watcher's floor where it is, so the slot is read again later and
|
||||
// nothing is lost.
|
||||
let full: Vec<_> = (0..MAX_PENDING_CROSS_ZONE_DISPATCHES)
|
||||
.map(|seed| PendingCrossZoneDispatchRecord::recorded(key_from_index(seed), vec![0_u8; 4]))
|
||||
.collect();
|
||||
assert_eq!(
|
||||
dbio.add_pending_cross_zone_dispatches(full).unwrap(),
|
||||
MAX_PENDING_CROSS_ZONE_DISPATCHES
|
||||
);
|
||||
|
||||
let over = PendingCrossZoneDispatchRecord::recorded(
|
||||
key_from_index(MAX_PENDING_CROSS_ZONE_DISPATCHES),
|
||||
vec![0_u8; 4],
|
||||
);
|
||||
assert!(
|
||||
dbio.add_pending_cross_zone_dispatches(vec![over]).is_err(),
|
||||
"recording past the cap must fail so the caller holds its floor"
|
||||
);
|
||||
assert_eq!(
|
||||
dbio.get_pending_cross_zone_dispatches().unwrap().len(),
|
||||
MAX_PENDING_CROSS_ZONE_DISPATCHES,
|
||||
"a refused write must leave the list untouched"
|
||||
);
|
||||
|
||||
// Re-offering only what is already held is not growth, so it still succeeds.
|
||||
assert_eq!(
|
||||
dbio.add_pending_cross_zone_dispatches(vec![PendingCrossZoneDispatchRecord::recorded(
|
||||
key_from_index(0),
|
||||
vec![0_u8; 4]
|
||||
)])
|
||||
.unwrap(),
|
||||
0
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn settled_dispatch_records_are_dropped_outside_an_update() {
|
||||
let temp_dir = tempdir().unwrap();
|
||||
let (dbio, _genesis) = dbio_with_genesis(temp_dir.path());
|
||||
|
||||
// The watcher re-reads a slot it already consumed and re-records a delivery
|
||||
// that settled long ago. Its key will never appear in a future block, so the
|
||||
// store-update path cannot reach it and this is the only thing that does.
|
||||
let first = dispatch_record(1);
|
||||
let second = dispatch_record(2);
|
||||
dbio.add_pending_cross_zone_dispatches(vec![first.clone(), second.clone()])
|
||||
.unwrap();
|
||||
|
||||
assert_eq!(
|
||||
dbio.drop_settled_cross_zone_dispatches(&[first.message_key])
|
||||
.unwrap(),
|
||||
1
|
||||
);
|
||||
assert_eq!(
|
||||
dbio.get_pending_cross_zone_dispatches().unwrap(),
|
||||
vec![second]
|
||||
);
|
||||
|
||||
// Dropping one that is already gone is a no-op, not an error.
|
||||
assert_eq!(
|
||||
dbio.drop_settled_cross_zone_dispatches(&[first.message_key])
|
||||
.unwrap(),
|
||||
0
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn finalized_dispatch_records_are_removed_by_message_key() {
|
||||
let temp_dir = tempdir().unwrap();
|
||||
let (dbio, _genesis) = dbio_with_genesis(temp_dir.path());
|
||||
|
||||
let first = dispatch_record(1);
|
||||
let second = dispatch_record(2);
|
||||
dbio.add_pending_cross_zone_dispatches(vec![first.clone(), second.clone()])
|
||||
.unwrap();
|
||||
|
||||
// Only the finalized delivery's key is dropped. Two deliveries can sit in
|
||||
// the same block, so a record must go by its own identity rather than by
|
||||
// anything about the height its delivery landed at.
|
||||
dbio.store_update(&StoreUpdate {
|
||||
remove_dispatch_records: &[first.message_key],
|
||||
..StoreUpdate::new(&state_with_balance(100))
|
||||
})
|
||||
.unwrap();
|
||||
|
||||
assert_eq!(
|
||||
dbio.get_pending_cross_zone_dispatches().unwrap(),
|
||||
vec![second]
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn record_dispatch_failure_drops_the_record_at_the_limit() {
|
||||
let temp_dir = tempdir().unwrap();
|
||||
let (dbio, _genesis) = dbio_with_genesis(temp_dir.path());
|
||||
|
||||
let record = dispatch_record(1);
|
||||
let key = record.message_key;
|
||||
let survivor = dispatch_record(2);
|
||||
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,
|
||||
"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"
|
||||
);
|
||||
|
||||
// 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.
|
||||
assert_eq!(
|
||||
dbio.get_pending_cross_zone_dispatches().unwrap(),
|
||||
vec![survivor],
|
||||
"giving up on a delivery drops its record 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"
|
||||
);
|
||||
assert_eq!(dbio.get_pending_cross_zone_dispatches().unwrap().len(), 1);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn repeated_withdrawal_key_in_one_update_folds_once_per_occurrence() {
|
||||
let temp_dir = tempdir().unwrap();
|
||||
|
||||
@ -217,7 +217,7 @@ impl Drop for TestContext {
|
||||
temp_wallet_dir: _,
|
||||
} = self;
|
||||
|
||||
let sequencer_handle = sequencer_handle
|
||||
let mut sequencer_handle = sequencer_handle
|
||||
.take()
|
||||
.expect("Sequencer handle should be present in TestContext drop");
|
||||
if !sequencer_handle.is_healthy() {
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user