mirror of
https://github.com/logos-blockchain/lssa.git
synced 2026-07-24 00:23:13 +00:00
1377 lines
52 KiB
Rust
1377 lines
52 KiB
Rust
use std::{
|
|
path::Path,
|
|
sync::{Arc, Mutex},
|
|
time::Instant,
|
|
};
|
|
|
|
use anyhow::{Context as _, Result, anyhow};
|
|
use borsh::BorshDeserialize;
|
|
use chain_state::{
|
|
AcceptOutcome, Anchor, AnchorConsistencyCheck, ChainConsistency, ChainMismatch, ChainState, Tip,
|
|
};
|
|
use common::{
|
|
HashType,
|
|
block::{BedrockStatus, Block, BlockMeta, HashableBlockData},
|
|
transaction::{LeeTransaction, clock_invocation},
|
|
};
|
|
use config::{GenesisAction, SequencerConfig};
|
|
use futures::StreamExt as _;
|
|
use itertools::Itertools as _;
|
|
use lee::{AccountId, PublicTransaction, public_transaction::Message};
|
|
use lee_core::GENESIS_BLOCK_ID;
|
|
use log::{error, info, warn};
|
|
use logos_blockchain_key_management_system_service::keys::{ED25519_SECRET_KEY_SIZE, Ed25519Key};
|
|
use logos_blockchain_zone_sdk::{
|
|
Slot, ZoneMessage,
|
|
sequencer::{DepositInfo, WithdrawArg},
|
|
};
|
|
use mempool::{MemPool, MemPoolHandle};
|
|
#[cfg(feature = "mock")]
|
|
pub use mock::SequencerCoreWithMockClients;
|
|
use num_bigint::BigUint;
|
|
pub use storage::error::DbError;
|
|
use storage::sequencer::{
|
|
RocksDBIO,
|
|
sequencer_cells::{PendingDepositEventRecord, WithdrawalReconciliationKey, ZoneAnchorRecord},
|
|
};
|
|
|
|
use crate::{
|
|
block_publisher::{BlockPublisherTrait, MsgId, ZoneSdkPublisher},
|
|
block_store::SequencerStore,
|
|
};
|
|
|
|
pub mod block_publisher;
|
|
pub mod block_store;
|
|
pub mod config;
|
|
pub mod cross_zone_watcher;
|
|
|
|
#[cfg(feature = "mock")]
|
|
pub mod mock;
|
|
|
|
/// The origin of a transaction.
|
|
#[derive(Clone, Copy)]
|
|
pub enum TransactionOrigin {
|
|
/// Basic transactions submitted by users via RPC.
|
|
User,
|
|
/// Transactions generated by the sequencer itself.
|
|
Sequencer,
|
|
}
|
|
|
|
#[derive(Clone, Debug, BorshDeserialize)]
|
|
struct DepositMetadata {
|
|
recipient_id: lee::AccountId,
|
|
}
|
|
|
|
pub struct SequencerCore<BP: BlockPublisherTrait = ZoneSdkPublisher> {
|
|
/// Two-tier chain state: production builds on its head; the publisher's
|
|
/// `on_follow` sink feeds adopted/orphaned/finalized peer blocks into it.
|
|
chain: Arc<Mutex<ChainState>>,
|
|
store: SequencerStore,
|
|
mempool: MemPool<(TransactionOrigin, LeeTransaction)>,
|
|
sequencer_config: SequencerConfig,
|
|
block_publisher: BP,
|
|
}
|
|
|
|
impl<BP: BlockPublisherTrait> SequencerCore<BP> {
|
|
/// Starts the sequencer using the provided configuration.
|
|
/// If an existing database is found, the sequencer state is loaded from it and
|
|
/// assumed to represent the correct latest state consistent with Bedrock-finalized data.
|
|
/// If no database is found, the sequencer performs a fresh start from genesis,
|
|
/// initializing its state with the accounts defined in the configuration file.
|
|
fn open_or_create_store(config: &SequencerConfig) -> (SequencerStore, lee::V03State) {
|
|
let signing_key = lee::PrivateKey::try_new(config.signing_key).unwrap();
|
|
let db_path = config.home.join("rocksdb");
|
|
|
|
if db_path.exists() {
|
|
let store = SequencerStore::open_db(&db_path, signing_key).unwrap_or_else(|err| {
|
|
panic!(
|
|
"Failed to open database at {} with error: {err}",
|
|
db_path.display()
|
|
)
|
|
});
|
|
let state = store
|
|
.get_lee_state()
|
|
.expect("Failed to read state from store");
|
|
(store, state)
|
|
} else {
|
|
warn!(
|
|
"Database not found at {}, starting from genesis",
|
|
db_path.display()
|
|
);
|
|
|
|
let (genesis_state, genesis_txs) = build_genesis_state(config);
|
|
|
|
let hashable_data = HashableBlockData {
|
|
block_id: GENESIS_BLOCK_ID,
|
|
transactions: genesis_txs,
|
|
prev_block_hash: HashType([0; 32]),
|
|
timestamp: 0,
|
|
};
|
|
let genesis_block = hashable_data.into_pending_block(&signing_key);
|
|
|
|
let store = SequencerStore::create_db_with_genesis(
|
|
&db_path,
|
|
&genesis_block,
|
|
&genesis_state,
|
|
signing_key,
|
|
)
|
|
.expect("Failed to create database with genesis block");
|
|
|
|
(store, genesis_state)
|
|
}
|
|
}
|
|
|
|
/// Rebuilds the two-tier [`ChainState`]: the final tier from the persisted
|
|
/// final snapshot (pre-genesis state when absent), the head tier by replaying
|
|
/// every stored block above it, so a post-restart orphan can still revert.
|
|
fn restore_chain_state(
|
|
config: &SequencerConfig,
|
|
store: &SequencerStore,
|
|
stored_head_state: &lee::V03State,
|
|
) -> ChainState {
|
|
let final_snapshot = store
|
|
.dbio()
|
|
.get_final_snapshot()
|
|
.expect("Failed to read final snapshot from store");
|
|
let (final_state, final_tip) = match final_snapshot {
|
|
Some((state, meta)) => (state, Some(Tip::from(meta))),
|
|
// Nothing finalized yet: replay the whole stored chain.
|
|
None => (build_initial_state(config), None),
|
|
};
|
|
let boundary = final_tip.as_ref().map_or(0, |tip| tip.block_id);
|
|
|
|
let mut head_blocks = store
|
|
.get_all_blocks()
|
|
.filter_ok(|block| block.header.block_id > boundary)
|
|
.collect::<Result<Vec<_>, _>>()
|
|
.expect("Failed to read blocks from store while restoring chain state");
|
|
head_blocks.sort_unstable_by_key(|block| block.header.block_id);
|
|
|
|
let mut chain = ChainState::from_final(final_state, final_tip);
|
|
for block in head_blocks {
|
|
let block_id = block.header.block_id;
|
|
chain.restore_head_block(block).unwrap_or_else(|err| {
|
|
panic!("Stored block {block_id} does not replay while restoring chain state: {err}")
|
|
});
|
|
}
|
|
|
|
// The replayed head must reproduce the persisted state, else store
|
|
// and config disagree (e.g. edited genesis actions).
|
|
assert!(
|
|
chain.head_state() == stored_head_state,
|
|
"Persisted state does not match the replayed chain; reset the store or restore the original config"
|
|
);
|
|
|
|
chain
|
|
}
|
|
|
|
pub async fn start_from_config(
|
|
config: SequencerConfig,
|
|
) -> (Self, MemPoolHandle<(TransactionOrigin, LeeTransaction)>) {
|
|
let bedrock_signing_key =
|
|
load_or_create_signing_key(&config.home.join("bedrock_signing_key"))
|
|
.expect("Failed to load or create bedrock signing key");
|
|
info!(
|
|
"Bedrock signing public key: {}",
|
|
hex::encode(bedrock_signing_key.public_key().to_bytes())
|
|
);
|
|
|
|
let (store, state) = Self::open_or_create_store(&config);
|
|
|
|
let chain = Arc::new(Mutex::new(Self::restore_chain_state(
|
|
&config, &store, &state,
|
|
)));
|
|
|
|
let initial_checkpoint = store
|
|
.get_zone_checkpoint()
|
|
.expect("Failed to load zone-sdk checkpoint");
|
|
let is_fresh_start = initial_checkpoint.is_none();
|
|
|
|
let (mempool, mempool_handle) = MemPool::new(config.mempool_max_size);
|
|
replay_unfulfilled_deposit_events(&store, mempool_handle.clone());
|
|
|
|
let block_publisher = BP::new(
|
|
&config.bedrock_config,
|
|
bedrock_signing_key,
|
|
config.retry_pending_blocks_timeout,
|
|
initial_checkpoint,
|
|
Self::on_checkpoint(store.dbio()),
|
|
Self::on_finalized_block(store.dbio()),
|
|
Self::on_deposit_event(store.dbio(), mempool_handle.clone()),
|
|
Self::on_withdraw_event(store.dbio()),
|
|
Self::on_follow(store.dbio(), Arc::clone(&chain), mempool_handle.clone()),
|
|
)
|
|
.await
|
|
.expect("Failed to initialize Block Publisher");
|
|
|
|
// 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,
|
|
);
|
|
}
|
|
// Before producing, verify our local state still belongs to the chain
|
|
// the channel serves and replay any channel blocks we are missing
|
|
// (e.g. from other sequencers).
|
|
let channel_was_empty =
|
|
Self::verify_and_reconstruct(&block_publisher, &store, &chain, is_fresh_start)
|
|
.await
|
|
.expect("Failed to verify/reconstruct sequencer state from Bedrock");
|
|
|
|
// Publish our blocks only when bootstrapping an empty channel. If the
|
|
// channel already has blocks (another sequencer bootstrapped it), we
|
|
// adopted them during reconstruction instead.
|
|
if is_fresh_start && channel_was_empty {
|
|
let mut pending_blocks = store
|
|
.get_all_blocks()
|
|
.filter_ok(|block| matches!(block.bedrock_status, BedrockStatus::Pending))
|
|
.collect::<Result<Vec<_>, _>>()
|
|
.expect("Failed to read blocks from store while republishing on fresh start");
|
|
pending_blocks.sort_unstable_by_key(|block| block.header.block_id);
|
|
|
|
assert!(
|
|
pending_blocks
|
|
.first()
|
|
.is_none_or(|block| block.header.block_id == GENESIS_BLOCK_ID),
|
|
"First pending block on fresh start should be the genesis block"
|
|
);
|
|
|
|
for block in &pending_blocks {
|
|
block_publisher
|
|
.publish_block(block, vec![])
|
|
.await
|
|
.unwrap_or_else(|err| {
|
|
panic!(
|
|
"Failed to publish block {} on fresh start: {err:#}",
|
|
block.header.block_id
|
|
)
|
|
});
|
|
}
|
|
}
|
|
|
|
let sequencer_core = Self {
|
|
chain,
|
|
store,
|
|
mempool,
|
|
sequencer_config: config,
|
|
block_publisher,
|
|
};
|
|
|
|
(sequencer_core, mempool_handle)
|
|
}
|
|
|
|
/// Verifies the local store still belongs to the chain the connected channel
|
|
/// serves and replays any finalized channel blocks missing locally into
|
|
/// `state`/`store`, recording each block's L1 inscription slot as the new
|
|
/// anchor. Fails (never parks) on any divergence.
|
|
///
|
|
/// Returns whatever channel was empty or not.
|
|
async fn verify_and_reconstruct(
|
|
publisher: &BP,
|
|
store: &SequencerStore,
|
|
chain: &Mutex<ChainState>,
|
|
is_fresh_start: bool,
|
|
) -> Result<bool> {
|
|
let anchor_record = store
|
|
.get_zone_anchor()
|
|
.context("Failed to read zone anchor")?;
|
|
|
|
let after_slot = anchor_record
|
|
.and_then(|record| record.slot.checked_sub(1))
|
|
.map(Slot::from);
|
|
let channel_tip_slot = publisher
|
|
.channel_tip_slot()
|
|
.await
|
|
.context("Failed to read channel tip slot")?;
|
|
|
|
// If this sequencer has already committed blocks to the channel, that
|
|
// channel must still exist. A missing channel then means a wiped/rewound
|
|
// Bedrock or a node pointing at a different chain, so refuse to resume
|
|
// onto a foreign channel.
|
|
//
|
|
// "Committed" requires *both* a non-genesis tip and a checkpoint that was
|
|
// persisted before this startup: the tip alone is set the moment we produce
|
|
// (before the channel confirms it), while a checkpoint alone is written by
|
|
// zone-sdk's cold-start backfill even on a brand-new empty channel before we
|
|
// publish genesis. We must read the checkpoint presence from before `BP::new`
|
|
// ran (`!is_fresh_start`), because its cold-start backfill re-persists a
|
|
// checkpoint by the time we reach here — reading the store now would always
|
|
// see one. Together they mean we produced blocks and zone-sdk processed
|
|
// channel activity in a prior run.
|
|
let local_tip = store
|
|
.latest_block_meta()
|
|
.context("Failed to read latest block meta")?
|
|
.map(|meta| meta.id);
|
|
let had_checkpoint_before_start = !is_fresh_start;
|
|
if let Some(local_tip) = local_tip
|
|
&& had_checkpoint_before_start
|
|
&& channel_tip_slot.is_none()
|
|
{
|
|
return Err(anyhow!(
|
|
"Sequencer holds committed blocks (tip {local_tip}) but the Bedrock channel \
|
|
no longer exists on the connected chain — the channel was wiped or the node \
|
|
points at a different chain. Refusing to resume onto a foreign channel."
|
|
));
|
|
}
|
|
|
|
let divergence_error = |mismatch: &ChainMismatch| {
|
|
anyhow!(
|
|
"Sequencer store diverges from the Bedrock channel ({mismatch}). \
|
|
Delete the sequencer storage directory or point at the correct channel."
|
|
)
|
|
};
|
|
|
|
// With a recorded anchor, probe the channel for positive evidence of a
|
|
// different chain: the frontier upfront (a missing/behind channel serves
|
|
// no messages to scan), then the anchor block as messages stream in.
|
|
let mut consistency_check = anchor_record.map(|record| {
|
|
let anchor = Anchor::new(
|
|
Slot::from(record.slot),
|
|
Some((record.block_id, record.hash)),
|
|
);
|
|
let mut check = AnchorConsistencyCheck::new(anchor);
|
|
check.check_frontier(channel_tip_slot);
|
|
check
|
|
});
|
|
if let Some(ChainConsistency::Inconsistent(mismatch)) = consistency_check
|
|
.as_ref()
|
|
.and_then(AnchorConsistencyCheck::verdict)
|
|
{
|
|
return Err(divergence_error(mismatch));
|
|
}
|
|
|
|
// Verify each message against the anchor and replay the
|
|
// blocks (applying the ones we miss, checking the ones we hold).
|
|
let messages = publisher
|
|
.read_channel_after(after_slot)
|
|
.await
|
|
.context("Failed to read channel history for reconstruction")?;
|
|
let mut messages = std::pin::pin!(messages);
|
|
let mut channel_is_empty = true;
|
|
while let Some((message, slot)) = messages.next().await {
|
|
if let Some(check) = &mut consistency_check
|
|
&& let Some(ChainConsistency::Inconsistent(mismatch)) =
|
|
check.observe(&message, slot)
|
|
{
|
|
return Err(divergence_error(mismatch));
|
|
}
|
|
|
|
let ZoneMessage::Block(zone_block) = message else {
|
|
continue;
|
|
};
|
|
let block: Block = borsh::from_slice(&zone_block.data).map_err(|err| {
|
|
anyhow!(
|
|
"Failed to deserialize channel block at slot {}: {err}",
|
|
slot.into_inner()
|
|
)
|
|
})?;
|
|
channel_is_empty = false;
|
|
// Locked per message (not across the stream `await`): concurrent
|
|
// follow events interleave safely — both paths apply idempotently
|
|
// and persist under this same lock.
|
|
let mut chain = chain.lock().expect("chain state mutex poisoned");
|
|
Self::apply_reconstructed_block(store, &mut chain, &block, slot)?;
|
|
}
|
|
|
|
Ok(channel_is_empty)
|
|
}
|
|
|
|
/// Applies a single channel block during reconstruction: idempotent for
|
|
/// blocks we already hold (verifying their hash), a validated continuation
|
|
/// for new ones. Advances the persisted anchor to the block's slot.
|
|
fn apply_reconstructed_block(
|
|
store: &SequencerStore,
|
|
chain: &mut ChainState,
|
|
block: &Block,
|
|
slot: Slot,
|
|
) -> Result<()> {
|
|
let tip = store
|
|
.latest_block_meta()
|
|
.context("Failed to read latest block meta")?;
|
|
let block_id = block.header.block_id;
|
|
let block_hash = block.header.hash;
|
|
|
|
let record = ZoneAnchorRecord {
|
|
slot: slot.into_inner(),
|
|
block_id,
|
|
hash: block_hash,
|
|
};
|
|
|
|
// A block at/below the tip must match what we already stored, otherwise
|
|
// the channel is a different chain.
|
|
if let Some(tip) = &tip
|
|
&& block_id <= tip.id
|
|
{
|
|
match store
|
|
.get_block_at_id(block_id)
|
|
.context("Failed to read stored block")?
|
|
{
|
|
Some(stored) if stored.header.hash == block_hash => {
|
|
store
|
|
.set_zone_anchor(&record)
|
|
.context("Failed to persist zone anchor")?;
|
|
return Ok(());
|
|
}
|
|
Some(stored) => {
|
|
return Err(anyhow!(
|
|
"Channel block {block_id} hash {block_hash} does not match stored hash {}",
|
|
stored.header.hash
|
|
));
|
|
}
|
|
None => {
|
|
return Err(anyhow!(
|
|
"Channel block {block_id} is at/below local tip {} but is missing locally",
|
|
tip.id
|
|
));
|
|
}
|
|
}
|
|
}
|
|
|
|
// New continuation: channel history is finalized, so it goes through
|
|
// the final tier — validation happens inside `apply_finalized`.
|
|
match chain.apply_finalized(MsgId::from(block.header.hash.0), block, slot) {
|
|
AcceptOutcome::Applied | AcceptOutcome::AlreadyApplied => {}
|
|
AcceptOutcome::Parked(err) | AcceptOutcome::RetryableFailure(err) => {
|
|
return Err(anyhow!(
|
|
"Channel block {block_id} does not extend local tip {:?}: {err}",
|
|
tip.map(|tip| tip.id)
|
|
));
|
|
}
|
|
}
|
|
|
|
// Persist like the follow path: the tip meta stays pinned to the head
|
|
// tip even when the reconstructed block lands below it.
|
|
let head_tip = chain.head_tip().map(|head| BlockMeta::from(&head));
|
|
let final_meta = chain.final_tip().map(|meta| BlockMeta::from(&meta));
|
|
store
|
|
.dbio()
|
|
.store_followed_blocks(
|
|
&[(block, true)],
|
|
head_tip.as_ref(),
|
|
chain.head_state(),
|
|
final_meta.as_ref().map(|meta| (chain.final_state(), meta)),
|
|
)
|
|
.context("Failed to persist reconstructed block")?;
|
|
|
|
// Mark the deposits' pending records submitted so the production-time
|
|
// guard drops the mints cold-start backfill re-queued for them. Withdraw
|
|
// intents are deliberately not counted: backfill already re-delivered and
|
|
// dropped their finalized L1 events, so an increment here would never be
|
|
// consumed and would leave a phantom count.
|
|
let deposit_event_ids: Vec<_> = block
|
|
.body
|
|
.transactions
|
|
.iter()
|
|
.filter_map(extract_bridge_deposit_id)
|
|
.collect();
|
|
store
|
|
.mark_deposit_events_submitted(&deposit_event_ids, block_id)
|
|
.context("Failed to mark reconstructed deposits submitted")?;
|
|
|
|
store
|
|
.set_zone_anchor(&record)
|
|
.context("Failed to persist zone anchor")?;
|
|
Ok(())
|
|
}
|
|
|
|
fn on_checkpoint(dbio: Arc<RocksDBIO>) -> block_publisher::CheckpointSink {
|
|
Box::new(move |cp| {
|
|
let bytes = match serde_json::to_vec(&cp) {
|
|
Ok(b) => b,
|
|
Err(err) => {
|
|
error!("Failed to serialize zone-sdk checkpoint: {err:#}");
|
|
return;
|
|
}
|
|
};
|
|
if let Err(err) = dbio.put_zone_sdk_checkpoint_bytes(&bytes) {
|
|
error!("Failed to persist zone-sdk checkpoint: {err:#}");
|
|
}
|
|
})
|
|
}
|
|
|
|
fn on_finalized_block(dbio: Arc<RocksDBIO>) -> block_publisher::FinalizedBlockSink {
|
|
Box::new(move |block_id| {
|
|
// NOTE: Theoretically Zone SDK may report finalization happening multiple times for the
|
|
// same block. In practice this is very unlikely to happen. For that to
|
|
// happen Sequencer should crash between receiving Finalized and Checkpoint events while
|
|
// these events happen very fast (because Checkpoints are generated by Zone SDK
|
|
// locally).
|
|
|
|
if let Err(err) = dbio.clean_pending_blocks_up_to(block_id) {
|
|
error!("Failed to mark pending blocks finalized up to {block_id}: {err:#}");
|
|
}
|
|
|
|
match dbio.remove_fulfilled_pending_deposit_events_up_to_block(block_id) {
|
|
Ok(0) => {}
|
|
Ok(removed) => {
|
|
info!(
|
|
"Removed {removed} fulfilled pending deposit events up to finalized block {block_id}"
|
|
);
|
|
}
|
|
Err(err) => {
|
|
error!(
|
|
"Failed to remove fulfilled pending deposit events up to block {block_id}: {err:#}"
|
|
);
|
|
}
|
|
}
|
|
})
|
|
}
|
|
|
|
fn on_deposit_event(
|
|
dbio: Arc<RocksDBIO>,
|
|
mempool_handle: MemPoolHandle<(TransactionOrigin, LeeTransaction)>,
|
|
) -> block_publisher::OnDepositEventSink {
|
|
Box::new(move |deposit| {
|
|
// NOTE: Theoretically Zone SDK may report multiple identical deposits. In practice this
|
|
// is very unlikely to happen. For that to happen Sequencer should crash
|
|
// between receiving Deposit and Checkpoint events while these events happen
|
|
// very fast (because Checkpoints are generated by Zone SDK locally).
|
|
|
|
let dbio = Arc::clone(&dbio);
|
|
let mempool_handle = mempool_handle.clone();
|
|
|
|
Box::pin(async move {
|
|
let id_hex = hex::encode(deposit.op_id);
|
|
info!("Observed Bedrock Deposit event with id: {id_hex}");
|
|
|
|
let event_record = pending_deposit_event_record(&deposit);
|
|
|
|
match dbio.add_pending_deposit_event(event_record.clone()) {
|
|
Ok(true) => {}
|
|
Ok(false) => {
|
|
info!(
|
|
"Deposit event {id_hex} already persisted as unfulfilled, skipping duplicate enqueue",
|
|
);
|
|
return;
|
|
}
|
|
Err(err) => {
|
|
error!(
|
|
"Failed to persist unfulfilled deposit event {id_hex} before enqueue: {err:#}. Deposit will be lost.",
|
|
);
|
|
return;
|
|
}
|
|
}
|
|
|
|
let tx = match build_bridge_deposit_tx_from_event(&event_record) {
|
|
Ok(tx) => tx,
|
|
Err(err) => {
|
|
error!(
|
|
"Failed to build transaction from Bedrock deposit event {id_hex}: {err:#}. Deposit will be lost.",
|
|
);
|
|
return;
|
|
}
|
|
};
|
|
|
|
if let Err(err) = mempool_handle
|
|
.push((TransactionOrigin::Sequencer, tx))
|
|
.await
|
|
{
|
|
error!(
|
|
"Failed to queue sequencer transaction built from finalized Bedrock event: {err:#}. Deposit will be lost."
|
|
);
|
|
}
|
|
})
|
|
})
|
|
}
|
|
|
|
fn on_withdraw_event(dbio: Arc<RocksDBIO>) -> block_publisher::OnWithdrawEventSink {
|
|
Box::new(move |withdraw| {
|
|
let dbio = Arc::clone(&dbio);
|
|
Box::pin(async move {
|
|
let hash_encoded = hex::encode(withdraw.tx_hash.as_ref());
|
|
let withdraw_key = match withdraw_event_reconciliation_key(&withdraw.op.outputs) {
|
|
Ok(key) => key,
|
|
Err(err) => {
|
|
error!(
|
|
"Failed to build reconciliation key for Bedrock Withdraw event with tx_hash {hash_encoded}: {err:#}"
|
|
);
|
|
return;
|
|
}
|
|
};
|
|
|
|
match dbio.consume_unseen_withdraw_count(withdraw_key) {
|
|
Ok(true) => {
|
|
info!("Validated Bedrock Withdraw event with tx_hash: {hash_encoded}");
|
|
}
|
|
Ok(false) => warn!(
|
|
"Unexpected Bedrock Withdraw event with tx_hash {hash_encoded}: no matching unseen withdraw found"
|
|
),
|
|
Err(err) => error!(
|
|
"Failed to reconcile Bedrock Withdraw event with tx_hash {hash_encoded}: {err:#}"
|
|
),
|
|
}
|
|
})
|
|
})
|
|
}
|
|
|
|
/// Publisher sink adapter over [`apply_follow_update`].
|
|
fn on_follow(
|
|
dbio: Arc<RocksDBIO>,
|
|
chain: Arc<Mutex<ChainState>>,
|
|
mempool_handle: MemPoolHandle<(TransactionOrigin, LeeTransaction)>,
|
|
) -> block_publisher::OnFollowSink {
|
|
Box::new(move |update: block_publisher::FollowUpdate| {
|
|
apply_follow_update(&dbio, &chain, &mempool_handle, update);
|
|
})
|
|
}
|
|
|
|
/// Produces a new block from mempool transactions and publishes it via zone-sdk.
|
|
pub async fn produce_new_block(&mut self) -> Result<u64> {
|
|
let BlockWithMeta {
|
|
block,
|
|
deposit_event_ids,
|
|
withdrawals,
|
|
} = self
|
|
.build_block_from_mempool()
|
|
.context("Failed to build block from mempool transactions")?;
|
|
|
|
let withdrawal_reconciliation_keys = withdrawals
|
|
.iter()
|
|
.map(|withdraw| withdraw_event_reconciliation_key(&withdraw.outputs))
|
|
.collect::<Result<_>>()
|
|
.context("Failed to build reconciliation keys for block withdrawals")?;
|
|
|
|
let this_msg = self
|
|
.block_publisher
|
|
.publish_block(&block, withdrawals)
|
|
.await
|
|
.context("Failed to publish block to Bedrock")?;
|
|
|
|
self.record_produced_block(
|
|
this_msg,
|
|
&block,
|
|
&deposit_event_ids,
|
|
withdrawal_reconciliation_keys,
|
|
)?;
|
|
|
|
Ok(block.header.block_id)
|
|
}
|
|
|
|
/// Applies our own freshly-published block to the head with the [`MsgId`] the
|
|
/// publish assigned it, so the head advances and the later adopted
|
|
/// redelivery dedups, then persists it.
|
|
///
|
|
/// Persistence is gated on the block actually becoming the head: if a peer
|
|
/// block won this height while we were publishing (`AlreadyApplied`, or
|
|
/// `Parked` when the head reorged to a different parent), the canonical
|
|
/// block is persisted by the follow path instead, and our invalidated
|
|
/// inscription comes back via `orphaned`.
|
|
fn record_produced_block(
|
|
&mut self,
|
|
this_msg: MsgId,
|
|
block: &Block,
|
|
deposit_event_ids: &[HashType],
|
|
withdrawal_reconciliation_keys: Vec<WithdrawalReconciliationKey>,
|
|
) -> Result<()> {
|
|
let mut chain = self.chain.lock().expect("chain state mutex poisoned");
|
|
match chain.apply_produced(this_msg, block) {
|
|
AcceptOutcome::Applied => {
|
|
// Persisted under the lock so disk writes land in apply order
|
|
// with the follow path.
|
|
self.store.update(
|
|
block,
|
|
deposit_event_ids,
|
|
withdrawal_reconciliation_keys,
|
|
chain.head_state(),
|
|
)?;
|
|
}
|
|
AcceptOutcome::AlreadyApplied => {
|
|
warn!(
|
|
"Produced block {} lost a competing-write race, skipping persistence",
|
|
block.header.block_id
|
|
);
|
|
}
|
|
AcceptOutcome::Parked(err) | AcceptOutcome::RetryableFailure(err) => {
|
|
warn!(
|
|
"Produced block {} no longer chains on the head, skipping persistence: {err}",
|
|
block.header.block_id
|
|
);
|
|
}
|
|
}
|
|
|
|
Ok(())
|
|
}
|
|
|
|
/// Validates and applies a single mempool transaction to the current state.
|
|
/// Returns `Ok(true)` if the transaction was valid and applied, `Ok(false)` if
|
|
/// it was skipped due to validation failure.
|
|
#[expect(
|
|
clippy::too_many_arguments,
|
|
reason = "splitting the produce-path accumulators into a struct buys nothing"
|
|
)]
|
|
fn apply_mempool_transaction(
|
|
store: &SequencerStore,
|
|
state: &mut lee::V03State,
|
|
origin: TransactionOrigin,
|
|
tx: &LeeTransaction,
|
|
block_height: u64,
|
|
timestamp: u64,
|
|
deposit_event_ids: &mut Vec<HashType>,
|
|
withdrawals: &mut Vec<WithdrawArg>,
|
|
) -> Result<bool> {
|
|
let tx_hash = tx.hash();
|
|
match origin {
|
|
TransactionOrigin::User => {
|
|
let validated_diff = match tx.validate_on_state(state, block_height, timestamp) {
|
|
Ok(diff) => diff,
|
|
Err(err) => {
|
|
error!(
|
|
"Transaction with hash {tx_hash} failed execution check with error: {err:#?}, skipping it",
|
|
);
|
|
return Ok(false);
|
|
}
|
|
};
|
|
|
|
if let Some(withdraw_data) = extract_bridge_withdraw_data(tx) {
|
|
withdrawals.push(withdraw_data);
|
|
}
|
|
|
|
state.apply_state_diff(validated_diff);
|
|
}
|
|
TransactionOrigin::Sequencer => {
|
|
let LeeTransaction::Public(public_tx) = tx else {
|
|
panic!("Sequencer may only generate Public transactions, found {tx:#?}");
|
|
};
|
|
|
|
if let Some(deposit_op_id) = extract_bridge_deposit_id(tx) {
|
|
if store
|
|
.is_deposit_event_submitted(deposit_op_id)
|
|
.context("Failed to check whether deposit was already submitted")?
|
|
{
|
|
info!("Skipping already-submitted bridge deposit {deposit_op_id}");
|
|
return Ok(false);
|
|
}
|
|
deposit_event_ids.push(deposit_op_id);
|
|
}
|
|
|
|
state
|
|
.transition_from_public_transaction(public_tx, block_height, timestamp)
|
|
.context("Failed to execute sequencer-generated transaction")?;
|
|
}
|
|
}
|
|
|
|
info!("Validated transaction with hash {tx_hash}, including it in block");
|
|
Ok(true)
|
|
}
|
|
|
|
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) = {
|
|
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| {
|
|
head.block_id
|
|
.checked_add(1)
|
|
.expect("block id should not overflow")
|
|
});
|
|
let prev = tip.map_or(HashType([0; 32]), |head| head.hash);
|
|
(prev, height, chain.head_state().clone())
|
|
};
|
|
|
|
let mut valid_transactions = Vec::new();
|
|
let mut deposit_event_ids = Vec::new();
|
|
let mut withdrawals = Vec::new();
|
|
|
|
let max_block_size = usize::try_from(self.sequencer_config.max_block_size.as_u64())
|
|
.expect("`max_block_size` should fit into usize");
|
|
|
|
let new_block_timestamp = u64::try_from(chrono::Utc::now().timestamp_millis())
|
|
.expect("Timestamp must be positive");
|
|
|
|
let clock_tx = clock_invocation(new_block_timestamp);
|
|
let clock_lee_tx = LeeTransaction::Public(clock_tx.clone());
|
|
|
|
while let Some((origin, tx)) = self.mempool.pop() {
|
|
let tx_hash = tx.hash();
|
|
|
|
let temp_valid_transactions = [
|
|
valid_transactions.as_slice(),
|
|
std::slice::from_ref(&tx),
|
|
std::slice::from_ref(&clock_lee_tx),
|
|
]
|
|
.concat();
|
|
let temp_hashable_data = HashableBlockData {
|
|
block_id: new_block_height,
|
|
transactions: temp_valid_transactions,
|
|
prev_block_hash,
|
|
timestamp: new_block_timestamp,
|
|
};
|
|
|
|
let block_size = borsh::to_vec(&temp_hashable_data)
|
|
.context("Failed to serialize block for size check")?
|
|
.len();
|
|
|
|
if block_size > max_block_size {
|
|
warn!(
|
|
"Transaction with hash {tx_hash} deferred to next block: \
|
|
block size {block_size} bytes would exceed limit of {max_block_size} bytes",
|
|
);
|
|
self.mempool.push_front((origin, tx));
|
|
break;
|
|
}
|
|
|
|
if Self::apply_mempool_transaction(
|
|
&self.store,
|
|
&mut working_state,
|
|
origin,
|
|
&tx,
|
|
new_block_height,
|
|
new_block_timestamp,
|
|
&mut deposit_event_ids,
|
|
&mut withdrawals,
|
|
)? {
|
|
valid_transactions.push(tx);
|
|
}
|
|
|
|
if valid_transactions.len() >= self.sequencer_config.max_num_tx_in_block {
|
|
break;
|
|
}
|
|
}
|
|
|
|
working_state
|
|
.transition_from_public_transaction(&clock_tx, new_block_height, new_block_timestamp)
|
|
.context("Clock transaction failed. Aborting block production.")?;
|
|
valid_transactions.push(clock_lee_tx);
|
|
|
|
let hashable_data = HashableBlockData {
|
|
block_id: new_block_height,
|
|
transactions: valid_transactions,
|
|
prev_block_hash,
|
|
timestamp: new_block_timestamp,
|
|
};
|
|
|
|
let block = hashable_data
|
|
.clone()
|
|
.into_pending_block(self.store.signing_key());
|
|
|
|
log::info!(
|
|
"Created block with {} transactions in {} seconds",
|
|
hashable_data.transactions.len(),
|
|
now.elapsed().as_secs()
|
|
);
|
|
|
|
Ok(BlockWithMeta {
|
|
block,
|
|
deposit_event_ids,
|
|
withdrawals,
|
|
})
|
|
}
|
|
|
|
/// Reads the current head state under the lock without cloning it, so callers
|
|
/// reuse `V03State`'s own API (accounts, nonces, proofs) with no whole-state copy.
|
|
pub fn with_state<R>(&self, f: impl FnOnce(&lee::V03State) -> R) -> R {
|
|
f(self
|
|
.chain
|
|
.lock()
|
|
.expect("chain state mutex poisoned")
|
|
.head_state())
|
|
}
|
|
|
|
pub const fn block_store(&self) -> &SequencerStore {
|
|
&self.store
|
|
}
|
|
|
|
#[must_use]
|
|
pub fn chain_height(&self) -> u64 {
|
|
self.chain
|
|
.lock()
|
|
.expect("chain state mutex poisoned")
|
|
.head_tip()
|
|
.map_or(0, |tip| tip.block_id)
|
|
}
|
|
|
|
pub const fn sequencer_config(&self) -> &SequencerConfig {
|
|
&self.sequencer_config
|
|
}
|
|
|
|
/// Marks all pending blocks with `block_id <= last_finalized_block_id` as
|
|
/// finalized. Idempotent. Production callers don't invoke this directly —
|
|
/// it's wired up in `start_from_config` to the publisher's
|
|
/// `on_finalized_block` sink, which fires on `Event::TxsFinalized` /
|
|
/// `Event::FinalizedInscriptions`. Kept on the type for tests.
|
|
// TODO: Delete blocks instead of marking them as finalized. Current
|
|
// approach is used because we still have `GetBlockDataRequest`.
|
|
pub fn clean_finalized_blocks_from_db(&self, last_finalized_block_id: u64) -> Result<()> {
|
|
info!("Clearing pending blocks up to id: {last_finalized_block_id}");
|
|
self.store
|
|
.dbio()
|
|
.clean_pending_blocks_up_to(last_finalized_block_id)?;
|
|
Ok(())
|
|
}
|
|
|
|
/// Returns the list of stored pending blocks.
|
|
pub fn get_pending_blocks(&self) -> Result<Vec<Block>> {
|
|
Ok(self
|
|
.store
|
|
.get_all_blocks()
|
|
.collect::<block_store::DbResult<Vec<Block>>>()?
|
|
.into_iter()
|
|
.filter(|block| matches!(block.bedrock_status, BedrockStatus::Pending))
|
|
.collect())
|
|
}
|
|
|
|
pub const fn block_publisher(&self) -> &BP {
|
|
&self.block_publisher
|
|
}
|
|
|
|
/// Whether this sequencer is currently authorized to write to the channel.
|
|
#[must_use]
|
|
pub fn is_our_turn(&self) -> bool {
|
|
self.block_publisher.is_our_turn()
|
|
}
|
|
|
|
/// Shared handle to the two-tier follow state, for tests to drive the
|
|
/// follow path directly.
|
|
#[cfg(all(test, feature = "mock"))]
|
|
fn chain(&self) -> Arc<Mutex<ChainState>> {
|
|
Arc::clone(&self.chain)
|
|
}
|
|
}
|
|
|
|
struct BlockWithMeta {
|
|
block: Block,
|
|
deposit_event_ids: Vec<HashType>,
|
|
withdrawals: Vec<WithdrawArg>,
|
|
}
|
|
|
|
/// 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
|
|
/// [`SequencerCore::on_follow`]; a free function so tests can drive it directly.
|
|
///
|
|
/// TODO: unlike the indexer's ingest loop, this path does not retry
|
|
/// `is_retryable` (transient) apply failures — a failed block just parks and
|
|
/// relies on a valid successor or a restart. `ChainState` never emits
|
|
/// `AcceptOutcome::RetryableFailure` yet; adding retry parity here is a
|
|
/// follow-up.
|
|
fn apply_follow_update(
|
|
dbio: &RocksDBIO,
|
|
chain: &Mutex<ChainState>,
|
|
mempool_handle: &MemPoolHandle<(TransactionOrigin, LeeTransaction)>,
|
|
update: block_publisher::FollowUpdate,
|
|
) {
|
|
let block_publisher::FollowUpdate {
|
|
adopted,
|
|
orphaned,
|
|
finalized,
|
|
} = update;
|
|
|
|
// The lock is held across the persist below so disk writes land in apply
|
|
// order — the produce path persists under this same lock.
|
|
let resubmit_txs = {
|
|
let mut chain = chain.lock().expect("chain state mutex poisoned");
|
|
|
|
// User txs of orphaned blocks, returned to the mempool below.
|
|
let resubmit_txs: Vec<LeeTransaction> = orphaned
|
|
.iter()
|
|
.flat_map(|(_, block)| resubmittable_txs(block))
|
|
.collect();
|
|
|
|
// Outcomes align with `adopted`.
|
|
let outcomes = chain.apply_channel_update(&orphaned, &adopted);
|
|
let mut to_persist: Vec<(&Block, bool)> = adopted
|
|
.iter()
|
|
.zip(&outcomes)
|
|
.filter(|(_, outcome)| matches!(outcome, AcceptOutcome::Applied))
|
|
.map(|((_, block), _)| (block, false))
|
|
.collect();
|
|
|
|
let mut final_advanced = false;
|
|
for (this_msg, block) in &finalized {
|
|
// FIXME: thread the finalized inscription's L1 slot once the
|
|
// sdk surfaces it; only used for the invalid-finalized stall.
|
|
if matches!(
|
|
chain.apply_finalized(*this_msg, block, Slot::from(0)),
|
|
AcceptOutcome::Applied
|
|
) {
|
|
to_persist.push((block, true));
|
|
final_advanced = true;
|
|
}
|
|
}
|
|
// Snapshot the advanced final tier so a restart re-anchors on it.
|
|
let final_meta = final_advanced.then(|| {
|
|
let tip = chain.final_tip().expect("advanced final tier has a tip");
|
|
BlockMeta::from(&tip)
|
|
});
|
|
let head_tip = chain.head_tip().map(|tip| BlockMeta::from(&tip));
|
|
|
|
// One atomic write for the whole update: blocks, tip meta and the
|
|
// state after the last block land together, so a crash can never
|
|
// leave the stored state ahead of the stored blocks. 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.
|
|
//
|
|
// TODO: the zone-sdk checkpoint is persisted by `on_checkpoint`
|
|
// before this write; a crash in between resumes past these blocks
|
|
// without them ever landing in the store. Full `BlocksProcessed`
|
|
// atomicity (checkpoint + blocks + state in one batch, per the sdk's
|
|
// event contract) is a follow-up.
|
|
dbio.store_followed_blocks(
|
|
&to_persist,
|
|
head_tip.as_ref(),
|
|
chain.head_state(),
|
|
final_meta.as_ref().map(|meta| (chain.final_state(), meta)),
|
|
)
|
|
.unwrap_or_else(|err| panic!("Failed to persist followed blocks: {err:#}"));
|
|
|
|
resubmit_txs
|
|
};
|
|
|
|
// Rebuild orphaned work: return its user txs to the mempool so the
|
|
// next on-turn production re-includes them on the new head.
|
|
//
|
|
// We use [`try_push`] here because this is called from the
|
|
// publisher's drive task, and only the block production drains the mempool.
|
|
// A blocking push on a full mempool would deadlock here.
|
|
for tx in resubmit_txs {
|
|
let tx_hash = tx.hash();
|
|
if let Err(err) = mempool_handle.try_push((TransactionOrigin::User, tx)) {
|
|
warn!("Dropping orphaned transaction {tx_hash} on resubmit: {err}");
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Checks the database for any pending deposit events that have not yet been marked as submitted in
|
|
/// a block, and re-queues them in the mempool in a separate async task for inclusion in the next
|
|
/// block.
|
|
fn replay_unfulfilled_deposit_events(
|
|
store: &SequencerStore,
|
|
mempool_handle: MemPoolHandle<(TransactionOrigin, LeeTransaction)>,
|
|
) {
|
|
let replay_records: Vec<PendingDepositEventRecord> = store
|
|
.get_unfulfilled_deposit_events()
|
|
.expect("Failed to load unfulfilled deposit events")
|
|
.into_iter()
|
|
.filter(|record| record.submitted_in_block_id.is_none())
|
|
.collect();
|
|
|
|
if replay_records.is_empty() {
|
|
return;
|
|
}
|
|
|
|
info!(
|
|
"Found {} unfulfilled deposit events in DB, re-queueing",
|
|
replay_records.len()
|
|
);
|
|
tokio::spawn(async move {
|
|
for record in replay_records {
|
|
let tx = match build_bridge_deposit_tx_from_event(&record) {
|
|
Ok(tx) => tx,
|
|
Err(err) => {
|
|
warn!(
|
|
"Skipping replay of pending deposit event {} due to tx build failure: {err:#}",
|
|
hex::encode(record.deposit_op_id)
|
|
);
|
|
continue;
|
|
}
|
|
};
|
|
|
|
if let Err(err) = mempool_handle
|
|
.push((TransactionOrigin::Sequencer, tx))
|
|
.await
|
|
{
|
|
error!(
|
|
"Failed to re-queue unfulfilled deposit event {} from DB: {err:#}",
|
|
hex::encode(record.deposit_op_id)
|
|
);
|
|
break;
|
|
}
|
|
}
|
|
});
|
|
}
|
|
|
|
/// The pre-genesis state: `testnet_initial_state` plus accounts seeded outside
|
|
/// any transaction (bridge-lock holdings, the cross-zone inbox config).
|
|
fn build_initial_state(config: &SequencerConfig) -> lee::V03State {
|
|
#[cfg(not(feature = "testnet"))]
|
|
let mut state = testnet_initial_state::initial_state();
|
|
|
|
#[cfg(feature = "testnet")]
|
|
let mut state = testnet_initial_state::initial_state_testnet();
|
|
|
|
// Seed bridge-lock holder balances directly: they are not produced by any tx.
|
|
for action in &config.genesis {
|
|
if let GenesisAction::SupplyBridgeLockHolding { holder, amount } = action {
|
|
let (holder_id, account) = cross_zone::build_holding_account(*holder, *amount);
|
|
state.insert_genesis_account(holder_id, account);
|
|
}
|
|
}
|
|
|
|
// Seed this zone's cross-zone inbox config so the inbox guest can authorize
|
|
// inbound peer messages (zone-specific config, not produced by any tx).
|
|
if let Some(cross_zone) = &config.cross_zone {
|
|
let self_zone = *config.bedrock_config.channel_id.as_ref();
|
|
let (config_id, config_account) =
|
|
cross_zone::build_inbox_config_account(self_zone, cross_zone);
|
|
state.insert_genesis_account(config_id, config_account);
|
|
}
|
|
|
|
state
|
|
}
|
|
|
|
/// Builds the initial genesis state from [`build_initial_state`] plus configured
|
|
/// genesis transactions. Returns the final state and the list of
|
|
/// [`LeeTransaction`]s that should be committed to the genesis block so external
|
|
/// observers can replay them.
|
|
fn build_genesis_state(config: &SequencerConfig) -> (lee::V03State, Vec<LeeTransaction>) {
|
|
let mut state = build_initial_state(config);
|
|
|
|
let genesis_txs = config
|
|
.genesis
|
|
.iter()
|
|
.filter_map(|genesis_tx| match genesis_tx {
|
|
GenesisAction::SupplyAccount {
|
|
account_id,
|
|
balance,
|
|
} => Some(build_supply_account_genesis_transaction(
|
|
account_id, *balance,
|
|
)),
|
|
GenesisAction::SupplyBridgeAccount { balance } => {
|
|
Some(build_supply_bridge_account_genesis_transaction(*balance))
|
|
}
|
|
// Force-inserted in `build_initial_state`: bridge_lock has no mint transaction.
|
|
GenesisAction::SupplyBridgeLockHolding { .. } => None,
|
|
})
|
|
.chain(std::iter::once(clock_invocation(0)))
|
|
.inspect(|tx| {
|
|
state
|
|
.transition_from_public_transaction(tx, GENESIS_BLOCK_ID, 0)
|
|
.expect("Failed to execute genesis transaction");
|
|
})
|
|
.map(LeeTransaction::Public)
|
|
.collect();
|
|
|
|
(state, genesis_txs)
|
|
}
|
|
|
|
/// Whether a program may only be invoked by sequencer-origin transactions.
|
|
///
|
|
/// The cross-zone inbox is injected solely by the watcher; a user-submitted call
|
|
/// must be rejected at ingress, since `TransactionOrigin` is not carried in the
|
|
/// block.
|
|
#[must_use]
|
|
pub fn is_sequencer_only_program(program_id: lee::ProgramId) -> bool {
|
|
cross_zone::is_sequencer_only_program(program_id)
|
|
}
|
|
|
|
fn build_supply_account_genesis_transaction(
|
|
account_id: &AccountId,
|
|
balance: u128,
|
|
) -> PublicTransaction {
|
|
let faucet_program_id = programs::faucet().id();
|
|
let vault_program_id = programs::vault().id();
|
|
let recipient_vault_id = vault_core::compute_vault_account_id(vault_program_id, *account_id);
|
|
|
|
let message = Message::try_new(
|
|
faucet_program_id,
|
|
vec![system_accounts::faucet_account_id(), recipient_vault_id],
|
|
vec![],
|
|
faucet_core::Instruction::GenesisTransferVault {
|
|
vault_program_id,
|
|
recipient_id: *account_id,
|
|
amount: balance,
|
|
},
|
|
)
|
|
.expect("Failed to serialize genesis transfer instruction");
|
|
let witness_set = lee::public_transaction::WitnessSet::from_raw_parts(vec![]);
|
|
|
|
PublicTransaction::new(message, witness_set)
|
|
}
|
|
|
|
fn build_supply_bridge_account_genesis_transaction(balance: u128) -> PublicTransaction {
|
|
let faucet_program_id = programs::faucet().id();
|
|
let bridge_account_id = system_accounts::bridge_account_id();
|
|
|
|
let message = Message::try_new(
|
|
faucet_program_id,
|
|
vec![system_accounts::faucet_account_id(), bridge_account_id],
|
|
vec![],
|
|
faucet_core::Instruction::GenesisTransferDirect { amount: balance },
|
|
)
|
|
.expect("Failed to serialize bridge genesis transfer instruction");
|
|
let witness_set = lee::public_transaction::WitnessSet::from_raw_parts(vec![]);
|
|
|
|
PublicTransaction::new(message, witness_set)
|
|
}
|
|
|
|
fn pending_deposit_event_record(deposit: &DepositInfo) -> PendingDepositEventRecord {
|
|
PendingDepositEventRecord {
|
|
deposit_op_id: HashType(deposit.op_id),
|
|
source_tx_hash: HashType(deposit.tx_hash.0),
|
|
amount: deposit.amount,
|
|
metadata: deposit.metadata.clone().into(),
|
|
submitted_in_block_id: None,
|
|
}
|
|
}
|
|
|
|
fn build_bridge_deposit_tx_from_event(event: &PendingDepositEventRecord) -> Result<LeeTransaction> {
|
|
let metadata = DepositMetadata::try_from_slice(&event.metadata)
|
|
.context("Failed to decode finalized Bedrock deposit metadata")?;
|
|
|
|
let bridge_program_id = programs::bridge().id();
|
|
let vault_program_id = programs::vault().id();
|
|
let recipient_vault_id =
|
|
vault_core::compute_vault_account_id(vault_program_id, metadata.recipient_id);
|
|
|
|
let message = Message::try_new(
|
|
bridge_program_id,
|
|
vec![system_accounts::bridge_account_id(), recipient_vault_id],
|
|
vec![],
|
|
bridge_core::Instruction::Deposit {
|
|
l1_deposit_op_id: event.deposit_op_id.0,
|
|
vault_program_id,
|
|
recipient_id: metadata.recipient_id,
|
|
amount: event.amount,
|
|
},
|
|
)
|
|
.context("Failed to build bridge deposit message")?;
|
|
|
|
let witness_set = lee::public_transaction::WitnessSet::from_raw_parts(vec![]);
|
|
Ok(LeeTransaction::Public(PublicTransaction::new(
|
|
message,
|
|
witness_set,
|
|
)))
|
|
}
|
|
|
|
/// User transactions of an orphaned block to return to the mempool: everything
|
|
/// except the trailing clock tx, sequencer-generated bridge deposits (replayed
|
|
/// from their own bedrock events) and sequencer-only cross-zone txs (replayed
|
|
/// by the watcher; the ingress guard rejects them as `User`).
|
|
fn resubmittable_txs(block: &Block) -> Vec<LeeTransaction> {
|
|
let Some((_clock, rest)) = block.body.transactions.split_last() else {
|
|
return Vec::new();
|
|
};
|
|
rest.iter()
|
|
.filter(|tx| extract_bridge_deposit_id(tx).is_none() && !is_sequencer_only_tx(tx))
|
|
.cloned()
|
|
.collect()
|
|
}
|
|
|
|
#[must_use]
|
|
fn is_sequencer_only_tx(tx: &LeeTransaction) -> bool {
|
|
matches!(tx, LeeTransaction::Public(tx)
|
|
if is_sequencer_only_program(tx.message().program_id))
|
|
}
|
|
|
|
#[must_use]
|
|
fn extract_bridge_deposit_id(tx: &LeeTransaction) -> Option<HashType> {
|
|
let LeeTransaction::Public(tx) = tx else {
|
|
return None;
|
|
};
|
|
|
|
let message = tx.message();
|
|
if message.program_id != programs::bridge().id() {
|
|
return None;
|
|
}
|
|
|
|
let instruction =
|
|
risc0_zkvm::serde::from_slice::<bridge_core::Instruction, u32>(&message.instruction_data)
|
|
.ok()?;
|
|
|
|
match instruction {
|
|
bridge_core::Instruction::Deposit {
|
|
l1_deposit_op_id, ..
|
|
} => Some(HashType(l1_deposit_op_id)),
|
|
bridge_core::Instruction::Withdraw { .. } => None,
|
|
}
|
|
}
|
|
|
|
#[must_use]
|
|
fn extract_bridge_withdraw_data(tx: &LeeTransaction) -> Option<WithdrawArg> {
|
|
let LeeTransaction::Public(tx) = tx else {
|
|
return None;
|
|
};
|
|
|
|
let message = tx.message();
|
|
if message.program_id != programs::bridge().id() {
|
|
return None;
|
|
}
|
|
|
|
let instruction =
|
|
risc0_zkvm::serde::from_slice::<bridge_core::Instruction, u32>(&message.instruction_data)
|
|
.ok()?;
|
|
|
|
let bridge_core::Instruction::Withdraw {
|
|
amount,
|
|
bedrock_account_pk,
|
|
} = instruction
|
|
else {
|
|
return None;
|
|
};
|
|
|
|
let recipient_pk = logos_blockchain_key_management_system_service::keys::ZkPublicKey::from(
|
|
BigUint::from_bytes_le(&bedrock_account_pk),
|
|
);
|
|
|
|
Some(WithdrawArg {
|
|
outputs: logos_blockchain_core::mantle::ledger::Outputs::new(
|
|
logos_blockchain_core::mantle::Note::new(amount, recipient_pk),
|
|
),
|
|
})
|
|
}
|
|
|
|
fn withdraw_event_reconciliation_key(
|
|
outputs: &logos_blockchain_core::mantle::ledger::Outputs,
|
|
) -> Result<WithdrawalReconciliationKey> {
|
|
let [note] = outputs.as_ref().as_slice() else {
|
|
return Err(anyhow!(
|
|
"Unsupported withdraw output count for reconciliation: {}",
|
|
outputs.len()
|
|
));
|
|
};
|
|
|
|
// `extract_bridge_withdraw_data` maps [u8;32] LE -> BigUint -> ZkPublicKey.
|
|
// Reconcile by reversing that direction here.
|
|
let mut bedrock_account_pk = BigUint::from(note.pk.into_inner()).to_bytes_le();
|
|
if bedrock_account_pk.len() > 32 {
|
|
return Err(anyhow!(
|
|
"Withdraw recipient public key is too large: {} bytes",
|
|
bedrock_account_pk.len()
|
|
));
|
|
}
|
|
bedrock_account_pk.resize(32, 0);
|
|
|
|
let bedrock_account_pk: [u8; 32] = bedrock_account_pk
|
|
.try_into()
|
|
.expect("Public key bytes were padded/truncated to 32 bytes");
|
|
|
|
Ok(WithdrawalReconciliationKey {
|
|
amount: note.value,
|
|
bedrock_account_pk,
|
|
})
|
|
}
|
|
|
|
/// Load signing key from file or generate a new one if it doesn't exist.
|
|
pub fn load_or_create_signing_key(path: &Path) -> Result<Ed25519Key> {
|
|
if path.exists() {
|
|
let key_bytes = std::fs::read(path)?;
|
|
|
|
let key_array: [u8; ED25519_SECRET_KEY_SIZE] = key_bytes
|
|
.try_into()
|
|
.map_err(|_bytes| anyhow!("Found key with incorrect length"))?;
|
|
|
|
Ok(Ed25519Key::from_bytes(&key_array))
|
|
} else {
|
|
let mut key_bytes = [0_u8; ED25519_SECRET_KEY_SIZE];
|
|
rand::RngCore::fill_bytes(&mut rand::thread_rng(), &mut key_bytes);
|
|
// Create parent directory if it doesn't exist
|
|
if let Some(parent) = path.parent() {
|
|
std::fs::create_dir_all(parent)?;
|
|
}
|
|
std::fs::write(path, key_bytes)?;
|
|
Ok(Ed25519Key::from_bytes(&key_bytes))
|
|
}
|
|
}
|
|
|
|
#[cfg(test)]
|
|
#[cfg(feature = "mock")]
|
|
mod tests;
|