diff --git a/Cargo.lock b/Cargo.lock index b470cdb3..bcc95943 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1193,6 +1193,7 @@ name = "bridge_core" version = "0.1.0" dependencies = [ "lee_core", + "risc0-zkvm", "serde", ] @@ -4155,6 +4156,7 @@ dependencies = [ "storage", "tempfile", "testnet_initial_state", + "thiserror 2.0.18", "tokio", "url", ] diff --git a/artifacts/lez/programs/amm.bin b/artifacts/lez/programs/amm.bin index 4e330974..7d0e45c6 100644 Binary files a/artifacts/lez/programs/amm.bin and b/artifacts/lez/programs/amm.bin differ diff --git a/artifacts/lez/programs/associated_token_account.bin b/artifacts/lez/programs/associated_token_account.bin index 1422fb6c..1cad1bb8 100644 Binary files a/artifacts/lez/programs/associated_token_account.bin and b/artifacts/lez/programs/associated_token_account.bin differ diff --git a/artifacts/lez/programs/authenticated_transfer.bin b/artifacts/lez/programs/authenticated_transfer.bin index 8fbe1100..885f723e 100644 Binary files a/artifacts/lez/programs/authenticated_transfer.bin and b/artifacts/lez/programs/authenticated_transfer.bin differ diff --git a/artifacts/lez/programs/bridge.bin b/artifacts/lez/programs/bridge.bin index 695373c9..4c9c1620 100644 Binary files a/artifacts/lez/programs/bridge.bin and b/artifacts/lez/programs/bridge.bin differ diff --git a/artifacts/lez/programs/bridge_lock.bin b/artifacts/lez/programs/bridge_lock.bin index a8d644e7..2c8f4e16 100644 Binary files a/artifacts/lez/programs/bridge_lock.bin and b/artifacts/lez/programs/bridge_lock.bin differ diff --git a/artifacts/lez/programs/clock.bin b/artifacts/lez/programs/clock.bin index c9d54e1d..e1f02c9f 100644 Binary files a/artifacts/lez/programs/clock.bin and b/artifacts/lez/programs/clock.bin differ diff --git a/artifacts/lez/programs/cross_zone_inbox.bin b/artifacts/lez/programs/cross_zone_inbox.bin index 5d96e3c7..adc32921 100644 Binary files a/artifacts/lez/programs/cross_zone_inbox.bin and b/artifacts/lez/programs/cross_zone_inbox.bin differ diff --git a/artifacts/lez/programs/cross_zone_outbox.bin b/artifacts/lez/programs/cross_zone_outbox.bin index 311ada1f..c01abe3e 100644 Binary files a/artifacts/lez/programs/cross_zone_outbox.bin and b/artifacts/lez/programs/cross_zone_outbox.bin differ diff --git a/artifacts/lez/programs/faucet.bin b/artifacts/lez/programs/faucet.bin index 2d6a9f1a..be8a500c 100644 Binary files a/artifacts/lez/programs/faucet.bin and b/artifacts/lez/programs/faucet.bin differ diff --git a/artifacts/lez/programs/pinata.bin b/artifacts/lez/programs/pinata.bin index 4d93d505..a6876882 100644 Binary files a/artifacts/lez/programs/pinata.bin and b/artifacts/lez/programs/pinata.bin differ diff --git a/artifacts/lez/programs/pinata_token.bin b/artifacts/lez/programs/pinata_token.bin index 0af077e1..1876bfcf 100644 Binary files a/artifacts/lez/programs/pinata_token.bin and b/artifacts/lez/programs/pinata_token.bin differ diff --git a/artifacts/lez/programs/ping_receiver.bin b/artifacts/lez/programs/ping_receiver.bin index e6daf6e4..5fd53651 100644 Binary files a/artifacts/lez/programs/ping_receiver.bin and b/artifacts/lez/programs/ping_receiver.bin differ diff --git a/artifacts/lez/programs/ping_sender.bin b/artifacts/lez/programs/ping_sender.bin index bdc15bdd..4cb16580 100644 Binary files a/artifacts/lez/programs/ping_sender.bin and b/artifacts/lez/programs/ping_sender.bin differ diff --git a/artifacts/lez/programs/token.bin b/artifacts/lez/programs/token.bin index 345f14b4..8f5fc900 100644 Binary files a/artifacts/lez/programs/token.bin and b/artifacts/lez/programs/token.bin differ diff --git a/artifacts/lez/programs/vault.bin b/artifacts/lez/programs/vault.bin index f0408f69..16828c6e 100644 Binary files a/artifacts/lez/programs/vault.bin and b/artifacts/lez/programs/vault.bin differ diff --git a/artifacts/lez/programs/wrapped_token.bin b/artifacts/lez/programs/wrapped_token.bin index 55f32217..6dc2b3d0 100644 Binary files a/artifacts/lez/programs/wrapped_token.bin and b/artifacts/lez/programs/wrapped_token.bin differ diff --git a/integration_tests/tests/bridge.rs b/integration_tests/tests/bridge.rs index f3fe87c5..3ff3027a 100644 --- a/integration_tests/tests/bridge.rs +++ b/integration_tests/tests/bridge.rs @@ -47,10 +47,11 @@ async fn public_bridge_deposit_invocation_is_dropped() -> anyhow::Result<()> { let bridge_account_id = system_accounts::bridge_account_id(); let vault_program_id = programs::vault().id(); let recipient_vault_id = vault_core::compute_vault_account_id(vault_program_id, recipient_id); + let receipt_id = bridge_core::deposit_receipt_account_id(programs::bridge().id(), [0_u8; 32]); let message = public_transaction::Message::try_new( programs::bridge().id(), - vec![bridge_account_id, recipient_vault_id], + vec![bridge_account_id, recipient_vault_id, receipt_id], vec![], bridge_core::Instruction::Deposit { l1_deposit_op_id: [0_u8; 32], @@ -95,10 +96,11 @@ async fn public_bridge_deposit_with_zero_amount_is_rejected() -> anyhow::Result< let bridge_account_id = system_accounts::bridge_account_id(); let vault_program_id = programs::vault().id(); let recipient_vault_id = vault_core::compute_vault_account_id(vault_program_id, recipient_id); + let receipt_id = bridge_core::deposit_receipt_account_id(programs::bridge().id(), [0_u8; 32]); let message = public_transaction::Message::try_new( programs::bridge().id(), - vec![bridge_account_id, recipient_vault_id], + vec![bridge_account_id, recipient_vault_id, receipt_id], vec![], bridge_core::Instruction::Deposit { l1_deposit_op_id: [0_u8; 32], @@ -155,8 +157,10 @@ async fn private_bridge_deposit_invocation_is_dropped() -> anyhow::Result<()> { let bridge_account_id = system_accounts::bridge_account_id(); let vault_program_id = programs::vault().id(); let recipient_vault_id = vault_core::compute_vault_account_id(vault_program_id, recipient_id); + let receipt_id = bridge_core::deposit_receipt_account_id(programs::bridge().id(), [0_u8; 32]); - // Get pre-state of bridge and vault accounts + // Get pre-state of bridge and vault accounts; the receipt is unminted (a + // default account), so the program would create it on a first mint. let bridge_pre = AccountWithMetadata::new( get_account(&ctx, bridge_account_id).await?, false, @@ -167,6 +171,8 @@ async fn private_bridge_deposit_invocation_is_dropped() -> anyhow::Result<()> { false, recipient_vault_id, ); + let receipt_pre = + AccountWithMetadata::new(lee_core::account::Account::default(), false, receipt_id); // Create program with dependencies let program_with_deps = @@ -193,16 +199,24 @@ async fn private_bridge_deposit_invocation_is_dropped() -> anyhow::Result<()> { // Execute and prove the bridge deposit let (output, proof) = execute_and_prove( - vec![bridge_pre.clone(), vault_pre.clone()], + vec![bridge_pre.clone(), vault_pre.clone(), receipt_pre.clone()], instruction, - vec![InputAccountIdentity::Public, InputAccountIdentity::Public], + vec![ + InputAccountIdentity::Public, + InputAccountIdentity::Public, + InputAccountIdentity::Public, + ], &program_with_deps, ) .context("Failed to execute/prove bridge deposit")?; // Create privacy-preserving transaction from circuit output let message = privacy_preserving_transaction::Message::from_circuit_output( - vec![bridge_pre.account.nonce, vault_pre.account.nonce], + vec![ + bridge_pre.account.nonce, + vault_pre.account.nonce, + receipt_pre.account.nonce, + ], output, ); diff --git a/lee/state_machine/src/state/mod.rs b/lee/state_machine/src/state/mod.rs index 8d278ea5..8b91b698 100644 --- a/lee/state_machine/src/state/mod.rs +++ b/lee/state_machine/src/state/mod.rs @@ -270,6 +270,12 @@ impl V03State { .unwrap_or_else(Account::default) } + /// Borrowing counterpart of [`Self::get_account_by_id`]. + #[must_use] + pub fn get_account_by_id_ref(&self, account_id: AccountId) -> Option<&Account> { + self.public_state.get(&account_id) + } + #[must_use] pub fn get_proof_for_commitment(&self, commitment: &Commitment) -> Option { self.private_state.0.get_proof_for(commitment) diff --git a/lez/indexer/core/Cargo.toml b/lez/indexer/core/Cargo.toml index afba6f29..c8c8590f 100644 --- a/lez/indexer/core/Cargo.toml +++ b/lez/indexer/core/Cargo.toml @@ -37,6 +37,7 @@ async-stream.workspace = true tokio.workspace = true risc0-zkvm.workspace = true hex.workspace = true +thiserror.workspace = true [dev-dependencies] tempfile.workspace = true diff --git a/lez/indexer/core/src/cross_zone_verifier.rs b/lez/indexer/core/src/cross_zone_verifier.rs index 5d95df4e..da568ab8 100644 --- a/lez/indexer/core/src/cross_zone_verifier.rs +++ b/lez/indexer/core/src/cross_zone_verifier.rs @@ -4,18 +4,18 @@ use std::{ time::Duration, }; -use anyhow::{Result, bail}; +use anyhow::anyhow; use common::{block::Block, transaction::LeeTransaction}; use cross_zone::{build_dispatch_from_emission, extract_emission}; use cross_zone_inbox_core::{ CrossZoneMessage, Instruction as InboxInstruction, MessageKey, ZoneId, message_key, }; -use futures::StreamExt as _; -use lee::PublicKey; +use futures::{Stream, StreamExt as _}; +use lee::{GENESIS_BLOCK_ID, PublicKey}; use log::{debug, error, info}; use logos_blockchain_core::mantle::ops::channel::ChannelId; use logos_blockchain_zone_sdk::{ - CommonHttpClient, ZoneMessage, adapter::NodeHttpClient, indexer::ZoneIndexer, + CommonHttpClient, Slot, ZoneMessage, adapter::NodeHttpClient, indexer::ZoneIndexer, }; use tokio::sync::RwLock; @@ -25,39 +25,157 @@ use crate::config::IndexerConfig; /// so a stuck wait is observable without rejecting a legitimate message. const LAG_LOG_INTERVAL: Duration = Duration::from_secs(30); +/// How long to wait for a referenced peer block before giving up on this pass. +/// Generous, since ordinary L1 finality lag delays a peer block by minutes. +/// Expiry is not a rejection: the caller retries the same block, so the cost of a +/// premature expiry is one repeated pass. +const PEER_BLOCK_WAIT_TIMEOUT: Duration = Duration::from_secs(300); + +/// How long each wait iteration sleeps. Also the unit the elapsed counter is +/// advanced by, so `waited` counts sleeps rather than wall time and, since a +/// sleep can overshoot, understates it. +const PEER_BLOCK_POLL_INTERVAL: Duration = Duration::from_secs(1); + +/// Consecutive passes a peer reader re-reads the same undecodable slot before +/// giving up and reading past it. +const DECODE_RETRY_LIMIT: u32 = 3; + +/// Why a cross-zone dispatch could not be verified. +/// +/// A forgery is terminal and must stop the block applying; an unavailable peer +/// block is transient and must be retried, or a lagging peer reader would +/// permanently halt ingestion. +#[derive(Debug, thiserror::Error)] +#[non_exhaustive] +pub enum CrossZoneVerifyError { + /// The dispatch does not match the peer's finalized chain. + #[error("{0:#}")] + Forged(anyhow::Error), + /// The referenced peer block has not been read yet. + #[error( + "peer zone {} block {block_id} still unavailable after {waited:?}", + hex::encode(zone) + )] + PeerUnavailable { + zone: ZoneId, + block_id: u64, + waited: Duration, + }, +} + +/// One peer zone's cached blocks, plus how far this reader has read them as an +/// unbroken hash-linked run from the peer's genesis. +#[derive(Default)] +struct PeerChain { + blocks: HashMap, + /// Highest id such that every block from [`GENESIS_BLOCK_ID`] up to it has + /// been read and each links to its predecessor. `None` until genesis is read. + /// + /// This, not `max(blocks.keys())`, is what the forgery test gates on: a peer + /// picks its own `block_id`s, and an id that does not continue the run + /// cannot advance the run. It bounds how far this reader has read, not that + /// the chain is authentic: the link is self-asserted, since `header.hash` is + /// not recomputed on decode and the reader does not apply the pinned-key + /// check that [`crate::cross_zone_verifier::CrossZoneVerifier::rederive`] + /// applies to the block a dispatch actually names. + verified_prefix: Option, +} + +impl PeerChain { + /// The id that would extend the verified run. + const fn next_expected(&self) -> u64 { + match self.verified_prefix { + Some(prefix) => prefix.saturating_add(1), + None => GENESIS_BLOCK_ID, + } + } + + /// Extends the verified run as far as the cached blocks allow. + fn extend_prefix(&mut self) { + while let Some(next) = self.blocks.get(&self.next_expected()) { + let links = match self.verified_prefix { + Some(prefix) => self + .blocks + .get(&prefix) + .is_some_and(|prev| prev.header.hash == next.header.prev_block_hash), + // Genesis has no predecessor to link to. + None => true, + }; + if !links { + return; + } + self.verified_prefix = Some(next.header.block_id); + } + } +} + +/// What one consistent look at the peer cache says about a referenced block. +enum PeerLookup { + Cached(Box), + /// Inside the verified run but not held, so it is not on the peer chain. + InsideRun, + /// The reader has not verified this far yet. + Behind, +} + /// Cache of finalized peer-zone blocks, filled by per-peer reader tasks and read /// by the verifier to re-derive cross-zone dispatch transactions. #[derive(Clone, Default)] struct PeerBlocks { - chains: Arc>>>, + chains: Arc>>, } impl PeerBlocks { async fn insert(&self, zone: ZoneId, block: Block) { - self.chains - .write() - .await - .entry(zone) - .or_default() - .insert(block.header.block_id, block); + let mut chains = self.chains.write().await; + let chain = chains.entry(zone).or_default(); + chain.blocks.insert(block.header.block_id, block); + chain.extend_prefix(); } + /// Resolves `block_id` under a single read lock. + /// + /// Answering "is it cached?" and "is it inside the verified run?" under two + /// separate locks races with the peer reader: an insert landing between them + /// reads as absent-and-inside-the-run, which is the forgery signal, for a + /// block that is in fact cached. That is the normal steady state, a waiting + /// verifier and the block it waits for arriving, so it must be one look. + async fn resolve(&self, zone: ZoneId, block_id: u64) -> PeerLookup { + let chains = self.chains.read().await; + let Some(chain) = chains.get(&zone) else { + return PeerLookup::Behind; + }; + if let Some(block) = chain.blocks.get(&block_id) { + return PeerLookup::Cached(Box::new(block.clone())); + } + if chain + .verified_prefix + .is_some_and(|prefix| prefix >= block_id) + { + PeerLookup::InsideRun + } else { + PeerLookup::Behind + } + } + + #[cfg(test)] async fn get(&self, zone: ZoneId, block_id: u64) -> Option { self.chains .read() .await .get(&zone) - .and_then(|chain| chain.get(&block_id).cloned()) + .and_then(|chain| chain.blocks.get(&block_id).cloned()) } - /// The highest block id this reader has finalized for `zone`, or `None` if it - /// has read nothing yet. - async fn highest_seen(&self, zone: ZoneId) -> Option { + /// How far this reader has read `zone` as an unbroken run from genesis, or + /// `None` if it has not read the peer's genesis block yet. + #[cfg(test)] + async fn verified_prefix(&self, zone: ZoneId) -> Option { self.chains .read() .await .get(&zone) - .and_then(|chain| chain.keys().copied().max()) + .and_then(|chain| chain.verified_prefix) } } @@ -115,8 +233,12 @@ impl CrossZoneVerifier { }) } - /// Verifies every cross-zone dispatch in a block, returning `Err` on the first - /// forged dispatch (the caller halts ingestion) or the keys to mark seen. + /// Verifies every cross-zone dispatch in a block, returning the keys to mark + /// seen, or the reason verification could not complete. + /// + /// [`CrossZoneVerifyError::Forged`] means the caller must halt ingestion; + /// [`CrossZoneVerifyError::PeerUnavailable`] means the caller must hold its + /// read cursor and retry, since the block is not yet judged either way. /// /// The caller MUST record the returned keys via [`Self::record_seen`] only /// after the block applies, so the seen-set mirrors the inbox's on-chain @@ -124,7 +246,10 @@ impl CrossZoneVerifier { /// forged dispatch reuse it to skip re-derivation while the inbox delivers the /// forgery. A key already seen is a replay the inbox no-ops, so it is accepted /// without re-derivation rather than halting on a legitimate re-delivery. - pub async fn verify_block(&self, block: &Block) -> Result> { + pub async fn verify_block( + &self, + block: &Block, + ) -> Result, CrossZoneVerifyError> { let mut verified = Vec::new(); for tx in &block.body.transactions { let Some(msg) = Self::decode_dispatch(tx) else { @@ -144,12 +269,12 @@ impl CrossZoneVerifier { let expected = self.rederive(&msg).await?; if LeeTransaction::Public(expected) != *tx { - bail!( + return Err(CrossZoneVerifyError::Forged(anyhow!( "forged cross-zone dispatch from zone {} block {} tx {}: re-derivation mismatch", hex::encode(msg.src_zone), msg.src_block_id, msg.src_tx_index - ); + ))); } info!( @@ -194,7 +319,10 @@ impl CrossZoneVerifier { /// Re-derives the dispatch transaction the watcher should have injected for /// `msg`, reading the source emission from the peer's finalized block. - async fn rederive(&self, msg: &CrossZoneMessage) -> Result { + async fn rederive( + &self, + msg: &CrossZoneMessage, + ) -> Result { let peer_block = self .wait_for_peer_block(msg.src_zone, msg.src_block_id) .await?; @@ -204,35 +332,43 @@ impl CrossZoneVerifier { if let Some(expected) = self.peer_pubkeys.get(&msg.src_zone) && !peer_block.is_signed_by(expected) { - bail!( + return Err(CrossZoneVerifyError::Forged(anyhow!( "forged cross-zone dispatch: peer zone {} block {} is not signed by the pinned block-signing key", hex::encode(msg.src_zone), msg.src_block_id - ); + ))); } + // Everything below is a property of the peer block just read, so a + // mismatch is the dispatch lying about it, not a transient condition. let emission_tx = peer_block .body .transactions .get(usize::try_from(msg.src_tx_index).expect("u32 index fits in usize")) .ok_or_else(|| { - anyhow::anyhow!( + CrossZoneVerifyError::Forged(anyhow!( "src_tx_index {} out of range in peer block", msg.src_tx_index - ) + )) })?; let LeeTransaction::Public(emission_tx) = emission_tx else { - bail!("peer emission transaction is not public"); + return Err(CrossZoneVerifyError::Forged(anyhow!( + "peer emission transaction is not public" + ))); }; let message = emission_tx.message(); let emission = extract_emission(message.program_id, &message.instruction_data).ok_or_else(|| { - anyhow::anyhow!("peer transaction at src_tx_index is not a recognized emitter") + CrossZoneVerifyError::Forged(anyhow!( + "peer transaction at src_tx_index is not a recognized emitter" + )) })?; if emission.target_zone != self.self_zone { - bail!("peer emission targets a different zone"); + return Err(CrossZoneVerifyError::Forged(anyhow!( + "peer emission targets a different zone" + ))); } Ok(build_dispatch_from_emission( @@ -248,26 +384,42 @@ impl CrossZoneVerifier { /// Resolves the referenced peer block, distinguishing forgery from lag. /// - /// If the block is cached, return it. If our peer reader has already - /// finalized past `block_id` and we still do not have it, the reference is to - /// a block that does not exist on the peer chain, a forgery, so reject now. - async fn wait_for_peer_block(&self, zone: ZoneId, block_id: u64) -> Result { + /// A `block_id` inside the run verified from the peer's genesis (see + /// [`PeerChain::verified_prefix`]) that we do not hold does not exist on the + /// peer chain, so reject it. Otherwise the reader has not reached it, so + /// wait, and give up after [`PEER_BLOCK_WAIT_TIMEOUT`] rather than block + /// ingestion forever. A reference to a block the peer will never produce + /// stalls rather than being rejected, since that is indistinguishable from a + /// peer that has not produced it yet; either way it is never applied. + async fn wait_for_peer_block( + &self, + zone: ZoneId, + block_id: u64, + ) -> Result { let mut waited = Duration::ZERO; loop { - if let Some(block) = self.peers.get(zone, block_id).await { - return Ok(block); + match self.peers.resolve(zone, block_id).await { + PeerLookup::Cached(block) => return Ok(*block), + // A backstop, not the live path: every id inside the run is + // cached by construction. Bounding the cache must preserve that + // or track a floor alongside the prefix, since an evicted block + // is not a forged one and reporting it as forged would halt a + // legitimate dispatch. + PeerLookup::InsideRun => { + return Err(CrossZoneVerifyError::Forged(anyhow!( + "forged cross-zone reference: peer zone {} chain is verified past block {} but it is absent", + hex::encode(zone), + block_id + ))); + } + PeerLookup::Behind => {} } - if self - .peers - .highest_seen(zone) - .await - .is_some_and(|h| h >= block_id) - { - bail!( - "forged cross-zone reference: peer zone {} finalized past block {} but it is absent", - hex::encode(zone), - block_id - ); + if waited >= PEER_BLOCK_WAIT_TIMEOUT { + return Err(CrossZoneVerifyError::PeerUnavailable { + zone, + block_id, + waited, + }); } if !waited.is_zero() && waited.as_secs().is_multiple_of(LAG_LOG_INTERVAL.as_secs()) { info!( @@ -277,12 +429,21 @@ impl CrossZoneVerifier { waited.as_secs() ); } - tokio::time::sleep(Duration::from_secs(1)).await; - waited = waited.saturating_add(Duration::from_secs(1)); + tokio::time::sleep(PEER_BLOCK_POLL_INTERVAL).await; + waited = waited.saturating_add(PEER_BLOCK_POLL_INTERVAL); } } } +/// The outcome of one pass over a peer's message stream. +#[derive(Debug, PartialEq, Eq)] +struct PeerPass { + /// Where the next pass resumes from. + cursor: Option, + /// Set when the pass ended early on a message that would not decode. + stalled_at: Option, +} + /// Reads a peer zone's finalized blocks from Bedrock into the shared cache. #[expect( clippy::infinite_loop, @@ -300,41 +461,124 @@ async fn read_peer( ); let mut cursor = None; + // The slot the reader is stuck on and how many passes it has spent there. + // Keyed by slot: the retry budget is per slot, so a failure at a new slot + // must not inherit an older slot's count and be skipped on its first try. + let mut stalled: Option<(Slot, u32)> = None; + let mut skip_slot = None; loop { - let stream = match zone_indexer.next_messages(cursor).await { - Ok(stream) => stream, - Err(err) => { - error!( - "Peer reader next_messages failed for {}: {err}", - hex::encode(peer_zone) - ); - tokio::time::sleep(poll_interval).await; - continue; - } - }; - let mut stream = std::pin::pin!(stream); - - while let Some((msg, slot)) = stream.next().await { - if let ZoneMessage::Block(zone_block) = msg { - match borsh::from_slice::(&zone_block.data) { - Ok(block) => peers.insert(peer_zone, block).await, - Err(err) => error!("Peer reader failed to deserialize block: {err}"), + match zone_indexer.next_messages(cursor).await { + Ok(stream) => { + let pass = consume_peer_stream(stream, peer_zone, &peers, cursor, skip_slot).await; + cursor = pass.cursor; + if let Some(slot) = pass.stalled_at { + let attempts = match stalled { + Some((prev, attempts)) if prev == slot => attempts.saturating_add(1), + _ => 1, + }; + if attempts >= DECODE_RETRY_LIMIT { + // Reading on leaves a hole: dispatches referencing the + // skipped block can no longer be verified, but every + // later block stays readable. + error!( + "Peer reader for {} could not decode slot {slot:?} after {attempts} attempts; reading past it.", + hex::encode(peer_zone) + ); + skip_slot = Some(slot); + stalled = None; + } else { + stalled = Some((slot, attempts)); + } + } else { + stalled = None; + skip_slot = None; } } - cursor = Some(slot); + Err(err) => error!( + "Peer reader next_messages failed for {}: {err}", + hex::encode(peer_zone) + ), } tokio::time::sleep(poll_interval).await; } } +/// Caches the finalized peer blocks carried by `stream`. +/// +/// A block that fails to deserialize ends the pass and holds the cursor at the +/// last fully-consumed slot, so the next poll re-reads it and a transient +/// failure heals itself. `skip_slot` names a slot the caller gave up on after +/// [`DECODE_RETRY_LIMIT`] attempts, which is read past instead so a permanently +/// undecodable inscription cannot wedge the reader. Skipping only leaves a hole, +/// which cannot advance [`PeerChain::verified_prefix`] past itself. +/// +/// The cursor advances only on a slot boundary, since one slot can carry several +/// messages and resuming mid-slot would skip the ones after the failure. This +/// relies on the stream never truncating mid-slot, which holds because the +/// zone-sdk materializes a whole batch before yielding and batches end on slot +/// boundaries. +async fn consume_peer_stream( + stream: S, + peer_zone: ZoneId, + peers: &PeerBlocks, + resume_from: Option, + skip_slot: Option, +) -> PeerPass +where + S: Stream, +{ + let mut stream = std::pin::pin!(stream); + let mut cursor = resume_from; + // The slot being consumed: cached so far, but there may be more to come. + let mut in_progress: Option = None; + + while let Some((msg, slot)) = stream.next().await { + if in_progress != Some(slot) { + cursor = in_progress.or(cursor); + in_progress = Some(slot); + } + + let ZoneMessage::Block(zone_block) = msg else { + continue; + }; + match borsh::from_slice::(&zone_block.data) { + Ok(block) => peers.insert(peer_zone, block).await, + Err(err) if skip_slot == Some(slot) => { + debug!( + "Peer reader skipping undecodable block from {} at slot {slot:?}: {err}", + hex::encode(peer_zone) + ); + } + Err(err) => { + error!( + "Peer reader failed to deserialize block from {} at slot {slot:?}: {err}. Holding the cursor and retrying.", + hex::encode(peer_zone) + ); + return PeerPass { + cursor, + stalled_at: Some(slot), + }; + } + } + } + + PeerPass { + cursor: in_progress.or(cursor), + stalled_at: None, + } +} + #[cfg(test)] mod tests { use common::test_utils::produce_dummy_block; + use futures::stream; use lee::{ PrivateKey, PublicKey, PublicTransaction, public_transaction::{Message, WitnessSet}, }; + use logos_blockchain_core::mantle::ops::channel::{MsgId, inscribe::Inscription}; + use logos_blockchain_zone_sdk::ZoneBlock; use ping_core::{SenderInstruction, ping_record_pda}; use super::*; @@ -375,6 +619,44 @@ mod tests { )) } + /// A peer-stream item inscribing `data` at `slot`. + fn peer_msg(data: Vec, slot: u64) -> (ZoneMessage, Slot) { + ( + ZoneMessage::Block(ZoneBlock { + id: MsgId::from([0; 32]), + data: Inscription::try_from(data).expect("test inscription is within bounds"), + }), + Slot::from(slot), + ) + } + + /// A hash-linked chain of `len` blocks from genesis, each carrying a `b"hi"` + /// emission. Only a chain built this way advances the verified prefix. + fn linked_chain(len: u64) -> Vec { + let mut prev = None; + let mut blocks = Vec::new(); + for offset in 0..len { + let block = produce_dummy_block( + GENESIS_BLOCK_ID.saturating_add(offset), + prev, + vec![emission(b"hi")], + ); + prev = Some(block.header.hash); + blocks.push(block); + } + blocks + } + + /// A peer-stream item carrying `block`. + fn peer_block_msg(block: &Block, slot: u64) -> (ZoneMessage, Slot) { + peer_msg(borsh::to_vec(block).expect("block serializes"), slot) + } + + /// A peer-stream item whose inscription is not a decodable block. + fn undecodable_msg(slot: u64) -> (ZoneMessage, Slot) { + peer_msg(b"not a block".to_vec(), slot) + } + /// The dispatch a watcher would inject for a `PEER_BLOCK_ID` emission of `payload`. fn dispatch(payload: &[u8]) -> LeeTransaction { let receiver_id = programs::ping_receiver().id(); @@ -472,28 +754,6 @@ mod tests { ); } - #[tokio::test] - async fn rejects_reference_to_a_block_the_peer_never_finalized() { - let verifier = verifier(); - // The reader has finalized past PEER_BLOCK_ID (it holds a later block) but - // never saw PEER_BLOCK_ID itself, so a dispatch referencing it is a forgery - // and must be rejected rather than waited on forever. - verifier - .peers - .insert( - PEER_ZONE, - produce_dummy_block(PEER_BLOCK_ID + 1, None, vec![emission(b"hi")]), - ) - .await; - - let block = produce_dummy_block(9, None, vec![dispatch(b"hi")]); - let err = verifier.verify_block(&block).await.unwrap_err(); - assert!( - err.to_string().contains("forged"), - "unexpected error: {err}" - ); - } - #[tokio::test] async fn accepts_replayed_dispatch_as_noop() { let verifier = verifier(); @@ -565,4 +825,208 @@ mod tests { "unexpected error: {err}" ); } + + #[tokio::test] + async fn peer_reader_advances_over_a_fully_decoded_stream() { + let peers = PeerBlocks::default(); + let chain = linked_chain(2); + let stream = stream::iter(vec![ + peer_block_msg(&chain[0], 0), + peer_block_msg(&chain[1], 1), + ]); + + let pass = consume_peer_stream(stream, PEER_ZONE, &peers, None, None).await; + + assert_eq!(pass.cursor, Some(Slot::from(1))); + assert_eq!(pass.stalled_at, None); + assert_eq!(peers.verified_prefix(PEER_ZONE).await, Some(2)); + } + + #[tokio::test] + async fn a_block_that_does_not_link_does_not_extend_the_verified_run() { + let peers = PeerBlocks::default(); + let genesis = linked_chain(1); + peers.insert(PEER_ZONE, genesis[0].clone()).await; + // Claims the next id, but not the predecessor it would have to follow. + peers + .insert( + PEER_ZONE, + produce_dummy_block(GENESIS_BLOCK_ID + 1, None, vec![emission(b"hi")]), + ) + .await; + + assert_eq!( + peers.verified_prefix(PEER_ZONE).await, + Some(GENESIS_BLOCK_ID) + ); + } + + #[tokio::test] + async fn a_block_arriving_between_the_two_halves_of_a_lookup_is_not_forged() { + // `resolve` answers "cached?" and "inside the verified run?" under one + // lock. Split across two, an insert landing between them reads as + // absent-and-inside-the-run, the forgery signal, for a cached block. + let peers = PeerBlocks::default(); + for block in linked_chain(2) { + peers.insert(PEER_ZONE, block).await; + } + + assert!(matches!( + peers.resolve(PEER_ZONE, 2).await, + PeerLookup::Cached(_) + )); + assert!(matches!( + peers.resolve(PEER_ZONE, 3).await, + PeerLookup::Behind + )); + } + + #[tokio::test] + async fn peer_reader_holds_its_cursor_on_an_undecodable_block() { + let peers = PeerBlocks::default(); + let chain = linked_chain(3); + let stream = stream::iter(vec![ + peer_block_msg(&chain[0], 0), + undecodable_msg(1), + peer_block_msg(&chain[2], 2), + ]); + + let pass = consume_peer_stream(stream, PEER_ZONE, &peers, None, None).await; + + assert_eq!(pass.cursor, Some(Slot::from(0))); + assert_eq!(pass.stalled_at, Some(Slot::from(1))); + assert_eq!(peers.verified_prefix(PEER_ZONE).await, Some(1)); + assert!(peers.get(PEER_ZONE, 3).await.is_none()); + } + + #[tokio::test] + async fn peer_reader_does_not_resume_inside_a_partially_failed_slot() { + let peers = PeerBlocks::default(); + let chain = linked_chain(1); + // One slot can carry several messages; the second one fails. + let stream = stream::iter(vec![peer_block_msg(&chain[0], 7), undecodable_msg(7)]); + + let pass = consume_peer_stream(stream, PEER_ZONE, &peers, Some(Slot::from(6)), None).await; + + // Slot 7 is re-read whole next pass, not resumed past the failure. + assert_eq!(pass.cursor, Some(Slot::from(6))); + } + + #[tokio::test] + async fn peer_reader_reads_past_a_slot_it_has_given_up_on() { + // After DECODE_RETRY_LIMIT attempts the caller nominates the slot to + // skip, so a permanently undecodable inscription cannot wedge the reader. + let peers = PeerBlocks::default(); + let chain = linked_chain(3); + let stream = stream::iter(vec![ + peer_block_msg(&chain[0], 0), + undecodable_msg(1), + peer_block_msg(&chain[2], 2), + ]); + + let pass = consume_peer_stream(stream, PEER_ZONE, &peers, None, Some(Slot::from(1))).await; + + assert_eq!(pass.cursor, Some(Slot::from(2)), "the pass drains"); + assert_eq!(pass.stalled_at, None); + // Block 3 is cached and servable, so dispatches referencing it verify. + assert!(peers.get(PEER_ZONE, 3).await.is_some()); + // But the hole stops the verified run, so block 2 is never called forged. + assert_eq!(peers.verified_prefix(PEER_ZONE).await, Some(1)); + } + + #[tokio::test(start_paused = true)] + async fn undecodable_peer_block_does_not_make_a_later_dispatch_look_forged() { + let verifier = verifier(); + let chain = linked_chain(3); + let stream = stream::iter(vec![ + peer_block_msg(&chain[0], 0), + undecodable_msg(1), + peer_block_msg(&chain[2], 2), + ]); + consume_peer_stream( + stream, + PEER_ZONE, + &verifier.peers, + None, + Some(Slot::from(1)), + ) + .await; + + // Regression: the reader cached block 3, so the old `max(cached ids)` + // high-water mark reached 3 and a dispatch referencing block 2 was + // rejected as forged, halting ingestion permanently. + let err = verifier + .wait_for_peer_block(PEER_ZONE, 2) + .await + .expect_err("block 2 was never read, so it cannot be resolved"); + assert!( + matches!(err, CrossZoneVerifyError::PeerUnavailable { .. }), + "a block outside the verified run is lag, not forgery: {err}" + ); + } + + #[tokio::test(start_paused = true)] + async fn a_high_block_id_cannot_poison_the_forgery_test() { + // A peer picks its own block ids, so one inscribed block claiming a huge + // id would drive a `max(cached ids)` high-water mark past every real id + // and make each later dispatch look forged. It cannot extend the + // verified run, so it is inert. + let verifier = verifier(); + let chain = linked_chain(2); + verifier.peers.insert(PEER_ZONE, chain[0].clone()).await; + verifier.peers.insert(PEER_ZONE, chain[1].clone()).await; + verifier + .peers + .insert( + PEER_ZONE, + produce_dummy_block(u64::MAX, None, vec![emission(b"hi")]), + ) + .await; + + assert_eq!(verifier.peers.verified_prefix(PEER_ZONE).await, Some(2)); + let err = verifier + .wait_for_peer_block(PEER_ZONE, 3) + .await + .expect_err("block 3 has not been read yet"); + assert!( + matches!(err, CrossZoneVerifyError::PeerUnavailable { .. }), + "a block beyond the verified run is lag, not forgery: {err}" + ); + } + + #[tokio::test] + async fn a_transient_decode_failure_heals_on_the_next_pass() { + let verifier = verifier(); + let chain = linked_chain(PEER_BLOCK_ID); + + let pass = consume_peer_stream( + stream::iter(vec![undecodable_msg(0)]), + PEER_ZONE, + &verifier.peers, + None, + None, + ) + .await; + assert_eq!(pass.cursor, None, "the failed slot is not skipped"); + assert_eq!(pass.stalled_at, Some(Slot::from(0))); + + // The next pass re-reads the same slot, which now decodes. + let pass = consume_peer_stream( + stream::iter(chain.iter().enumerate().map(|(index, block)| { + peer_block_msg(block, u64::try_from(index).expect("test index fits in u64")) + })), + PEER_ZONE, + &verifier.peers, + pass.cursor, + None, + ) + .await; + assert_eq!(pass.stalled_at, None); + + let block = produce_dummy_block(9, None, vec![dispatch(b"hi")]); + verifier + .verify_block(&block) + .await + .expect("the dispatch verifies once the peer block has been read"); + } } diff --git a/lez/indexer/core/src/lib.rs b/lez/indexer/core/src/lib.rs index 4cf7d58f..ccf8dbe7 100644 --- a/lez/indexer/core/src/lib.rs +++ b/lez/indexer/core/src/lib.rs @@ -16,7 +16,7 @@ use retry::ApplyRetryGate; use crate::{ block_store::IndexerStore, config::IndexerConfig, - cross_zone_verifier::CrossZoneVerifier, + cross_zone_verifier::{CrossZoneVerifier, CrossZoneVerifyError}, status::{IndexerStatus, IndexerSyncStatus}, }; @@ -29,6 +29,16 @@ pub mod status; /// Consecutive failed apply attempts of the same block before parking. const APPLY_RETRY_LIMIT: u32 = 3; +/// Which slot the ingest loop is currently inside, so the read cursor only ever +/// moves on a slot boundary. +/// +/// One L1 slot can carry several L2 blocks, and the channel stream resumes +/// *after* the stored slot. Advancing the cursor as each block is handled would +/// therefore put a later block in the same slot beyond the cursor whenever a +/// pass ends early, and nothing would ever read it again. +#[derive(Default)] +struct SlotProgress(Option); + #[derive(Clone)] pub struct IndexerCore { pub zone_indexer: Arc>, @@ -42,6 +52,23 @@ pub struct IndexerCore { pub verifier: Option, } +impl SlotProgress { + /// Records that a message from `slot` is being handled, returning the slot + /// that just completed, if this message begins a new one. + fn enter(&mut self, slot: Slot) -> Option { + if self.0 == Some(slot) { + return None; + } + self.0.replace(slot) + } + + /// The slot in progress when the stream drained cleanly, which is therefore + /// complete. Not called when a pass ends early: that slot must be re-read. + const fn drained(self) -> Option { + self.0 + } +} + impl IndexerCore { /// Builds the core, then verifies the stored chain matches the channel's by /// re-reading the channel at the stored tip's position. @@ -264,8 +291,20 @@ impl IndexerCore { let mut announced_syncing = false; let mut had_cycle_error = false; + // The slot being consumed: every message of it seen so far is + // handled, but another may follow, so the cursor may not move + // onto it yet. One L1 slot can carry several L2 blocks, and the + // stream resumes *after* the stored slot, so advancing inside a + // slot would put a later message in it beyond the cursor + // for ever if this pass ends early. + let mut in_progress = SlotProgress::default(); while let Some((msg, slot)) = stream.next().await { + // A message from a later slot means the previous one is complete. + if let Some(done) = in_progress.enter(slot) { + self.advance_cursor(&mut cursor, done); + } + if !announced_syncing { self.set_status(IndexerSyncStatus::syncing()); announced_syncing = true; @@ -286,30 +325,43 @@ impl IndexerCore { break; } // L1 proceeds regardless - self.advance_cursor(&mut cursor, slot); continue; } }; - // Option B: re-derive and verify every cross-zone dispatch - // before applying the block. A forged dispatch halts ingestion - // rather than persisting an invalid state; a replay is accepted - // since the inbox no-ops it on chain. The verified keys are - // marked seen only once the block applies (below), so a block - // that does not apply cannot poison the seen-set. + // Re-derive and verify every cross-zone dispatch the block + // carries before applying it, so the destination never trusts + // a dispatch just because a sequencer signed the block: a + // forged one halts ingestion rather than persisting invalid + // state, while a replay is accepted since the inbox no-ops it + // on chain. The verified keys are marked seen only once the + // block applies (below), so a block that does not apply + // cannot poison the seen-set. let verified_keys = match &self.verifier { Some(verifier) => match verifier.verify_block(&block).await { Ok(keys) => keys, - Err(err) => { + Err(err @ CrossZoneVerifyError::Forged(_)) => { error!( - "Cross-zone verification failed for block {}: {err:#}. Halting indexer ingestion.", + "Cross-zone verification failed for block {}: {err}. Halting indexer ingestion.", block.header.block_id ); self.set_status(IndexerSyncStatus::error(format!( - "cross-zone verification failed: {err:#}" + "cross-zone verification failed: {err}" ))); return; } + // Not judged either way yet, so retry rather than halt. + Err(err @ CrossZoneVerifyError::PeerUnavailable { .. }) => { + error!( + "Cross-zone verification of block {} stalled: {err}. Holding the cursor and retrying.", + block.header.block_id + ); + self.set_status(IndexerSyncStatus::error(format!( + "cross-zone peer unavailable: {err}" + ))); + had_cycle_error = true; + break; + } }, None => Vec::new(), }; @@ -322,7 +374,6 @@ impl IndexerCore { retry_gate.reset(); info!("Indexed L2 block {}", block.header.block_id); self.set_status(IndexerSyncStatus::syncing()); - self.advance_cursor(&mut cursor, slot); yield Ok(block); } Ok(AcceptOutcome::AlreadyApplied) => { @@ -330,7 +381,6 @@ impl IndexerCore { "Skipping already-applied block {}", block.header.block_id ); - self.advance_cursor(&mut cursor, slot); } Ok(AcceptOutcome::Parked(ingest_err)) => { error!( @@ -339,7 +389,6 @@ impl IndexerCore { ); self.set_status(IndexerSyncStatus::stalled(ingest_err.to_string())); // L1 proceeds regardless - self.advance_cursor(&mut cursor, slot); } Ok(AcceptOutcome::RetryableFailure(ingest_err)) => { let attempts = retry_gate.register_failure(block.header.block_id); @@ -365,7 +414,6 @@ impl IndexerCore { break; } self.set_status(IndexerSyncStatus::stalled(ingest_err.to_string())); - self.advance_cursor(&mut cursor, slot); retry_gate.reset(); } else { error!( @@ -396,10 +444,17 @@ impl IndexerCore { } if had_cycle_error { + // The slot in progress is not finished, so the cursor stays + // below it and the next pass re-reads it whole. tokio::time::sleep(poll_interval).await; continue; } + // The stream drained cleanly, so the slot in progress completed too. + if let Some(done) = in_progress.drained() { + self.advance_cursor(&mut cursor, done); + } + // Stream drained. Stay Stalled if parked; otherwise we are caught up. // A store error here must not be collapsed to "no stall recorded": // that would wrongly flip us to caught-up, so we log and hold state. @@ -426,6 +481,51 @@ mod tests { use super::*; use crate::config::{ChannelId, ClientConfig, IndexerConfig}; + /// The cursor must not move while more of the same slot may still arrive. + /// + /// Two L2 blocks in one L1 slot: the first applies, the second stalls on an + /// unavailable peer and the pass retries. If handling the first had advanced + /// the cursor onto the slot, the retry would resume past it and the second + /// block would never be read again, silently losing whatever it carried. + #[test] + fn a_slot_is_only_left_behind_once_it_is_finished() { + let mut progress = SlotProgress::default(); + let slot = Slot::from(7); + + assert_eq!( + progress.enter(slot), + None, + "nothing precedes the first slot" + ); + assert_eq!( + progress.enter(slot), + None, + "a second message in the same slot must not release it" + ); + + // The pass ends early here, so `drained` is never called and the cursor + // is still below slot 7: the next pass re-reads it whole. + } + + #[test] + fn a_completed_slot_is_released_when_the_next_one_starts() { + let mut progress = SlotProgress::default(); + + assert_eq!(progress.enter(Slot::from(3)), None); + assert_eq!(progress.enter(Slot::from(3)), None); + assert_eq!( + progress.enter(Slot::from(4)), + Some(Slot::from(3)), + "slot 3 is complete once a message from slot 4 arrives" + ); + assert_eq!(progress.drained(), Some(Slot::from(4))); + } + + #[test] + fn draining_an_untouched_stream_releases_nothing() { + assert_eq!(SlotProgress::default().drained(), None); + } + fn unreachable_core(dir: &std::path::Path) -> IndexerCore { let config = IndexerConfig { consensus_info_polling_interval: Duration::from_secs(1), diff --git a/lez/programs/bridge/core/Cargo.toml b/lez/programs/bridge/core/Cargo.toml index 201e899f..d08f2cd6 100644 --- a/lez/programs/bridge/core/Cargo.toml +++ b/lez/programs/bridge/core/Cargo.toml @@ -9,4 +9,5 @@ workspace = true [dependencies] lee_core.workspace = true +risc0-zkvm.workspace = true serde = { workspace = true, default-features = false } diff --git a/lez/programs/bridge/core/src/lib.rs b/lez/programs/bridge/core/src/lib.rs index c9666f27..129276dc 100644 --- a/lez/programs/bridge/core/src/lib.rs +++ b/lez/programs/bridge/core/src/lib.rs @@ -3,14 +3,19 @@ use lee_core::{account::AccountId, program::ProgramId}; use serde::{Deserialize, Serialize}; const BRIDGE_SEED_DOMAIN_SEPARATOR: [u8; 32] = *b"/LEZ/v0.3/BridgeSeed/0000000000/"; +const DEPOSIT_RECEIPT_SEED_DOMAIN: [u8; 32] = *b"/LEZ/v0.3/BridgeDepositReceipt/0"; #[derive(Serialize, Deserialize)] pub enum Instruction { - /// Transfers native tokens from the bridge PDA account to a recipient vault. + /// Transfers native tokens from the bridge PDA account to a recipient vault, + /// exactly once per `l1_deposit_op_id`. /// - /// Required accounts (2): + /// Required accounts (3): /// - Bridge PDA account /// - Recipient vault PDA account + /// - Deposit-receipt PDA account, derived from `l1_deposit_op_id`. Its existence records that + /// this op id was already minted; a second application of the same op id finds it present and + /// transfers nothing. Deposit { /// Deposit OP ID from L1, stored here to pin each [`Deposit`](Instruction::Deposit) to a /// Deposit Event on L1. @@ -43,3 +48,58 @@ pub const fn compute_bridge_seed() -> PdaSeed { pub fn compute_bridge_account_id(bridge_program_id: ProgramId) -> AccountId { AccountId::for_public_pda(&bridge_program_id, &compute_bridge_seed()) } + +/// Seed of the deposit-receipt PDA for `l1_deposit_op_id`, exposed so the guest +/// can claim the account. Domain-separated from [`compute_bridge_seed`]. +#[must_use] +pub fn deposit_receipt_seed(l1_deposit_op_id: [u8; 32]) -> PdaSeed { + use risc0_zkvm::sha::{Impl, Sha256 as _}; + + let mut bytes = [0_u8; 64]; + bytes[..32].copy_from_slice(&DEPOSIT_RECEIPT_SEED_DOMAIN); + bytes[32..].copy_from_slice(&l1_deposit_op_id); + + let seed: [u8; 32] = Impl::hash_bytes(&bytes) + .as_bytes() + .try_into() + .unwrap_or_else(|_| unreachable!()); + PdaSeed::new(seed) +} + +/// The deposit-receipt PDA whose existence marks `l1_deposit_op_id` as minted. +#[must_use] +pub fn deposit_receipt_account_id( + bridge_program_id: ProgramId, + l1_deposit_op_id: [u8; 32], +) -> AccountId { + AccountId::for_public_pda(&bridge_program_id, &deposit_receipt_seed(l1_deposit_op_id)) +} + +#[cfg(test)] +mod tests { + use super::*; + + const BRIDGE_ID: ProgramId = [7; 8]; + + #[test] + fn receipt_id_is_deterministic_per_op_id() { + let op = [3_u8; 32]; + assert_eq!( + deposit_receipt_account_id(BRIDGE_ID, op), + deposit_receipt_account_id(BRIDGE_ID, op) + ); + } + + #[test] + fn distinct_op_ids_and_domains_do_not_collide() { + let a = deposit_receipt_account_id(BRIDGE_ID, [1; 32]); + let b = deposit_receipt_account_id(BRIDGE_ID, [2; 32]); + assert_ne!(a, b, "different op ids must derive different receipts"); + // The op-id-derived seed must not alias the plain bridge PDA, even if an + // op id ever equals the bridge seed's raw bytes. + assert_ne!( + deposit_receipt_account_id(BRIDGE_ID, *compute_bridge_seed().as_bytes()), + compute_bridge_account_id(BRIDGE_ID) + ); + } +} diff --git a/lez/programs/bridge/src/main.rs b/lez/programs/bridge/src/main.rs index 19d6509c..c42149da 100644 --- a/lez/programs/bridge/src/main.rs +++ b/lez/programs/bridge/src/main.rs @@ -1,6 +1,7 @@ use bridge_core::Instruction; -use lee_core::program::{ - AccountPostState, ChainedCall, ProgramInput, ProgramOutput, read_lee_inputs, +use lee_core::{ + account::Account, + program::{AccountPostState, ChainedCall, Claim, ProgramInput, ProgramOutput, read_lee_inputs}, }; fn unchanged_post_states( @@ -29,18 +30,17 @@ fn main() { ); let pre_states_clone = pre_states.clone(); - let post_states = unchanged_post_states(&pre_states_clone); - let chained_calls = match instruction { + let (post_states, chained_calls) = match instruction { Instruction::Deposit { - l1_deposit_op_id: _, + l1_deposit_op_id, vault_program_id, recipient_id, amount, } => { - let [bridge, recipient_vault] = pre_states + let [bridge, recipient_vault, receipt] = pre_states .try_into() - .expect("Deposit requires exactly 2 accounts"); + .expect("Deposit requires exactly 3 accounts"); assert_eq!( bridge.account_id, @@ -54,20 +54,54 @@ fn main() { "Second account must be recipient vault PDA" ); - let mut bridge_for_vault = bridge; - bridge_for_vault.is_authorized = true; + assert_eq!( + receipt.account_id, + bridge_core::deposit_receipt_account_id(self_program_id, l1_deposit_op_id), + "Third account must be the deposit-receipt PDA" + ); - vec![ - ChainedCall::new( - vault_program_id, - vec![bridge_for_vault, recipient_vault], - &vault_core::Instruction::Transfer { - recipient_id, - amount: u128::from(amount), - }, - ) - .with_pda_seeds(vec![bridge_core::compute_bridge_seed()]), - ] + // Replay protection: the receipt PDA exists iff this op id was + // already minted. On replay it is non-default and the whole + // instruction is a no-op. + // + // Observability note: a no-op replay and a real first mint are both + // successful txs, so an indexer cannot tell "credited here" from + // "already credited by a peer" without deriving the receipt id and + // checking whether it existed before this block — the receipt claim + // is the only on-chain signal. Relevant once the explorer surfaces + // deposits. + if receipt.account != Account::default() { + (unchanged_post_states(&pre_states_clone), vec![]) + } else { + // First mint: claim the receipt — its existence is the record, + // the account's contents are never read — and chain the vault + // transfer. + let receipt_post = AccountPostState::new_claimed_if_default( + receipt.account, + Claim::Pda(bridge_core::deposit_receipt_seed(l1_deposit_op_id)), + ); + + let post_states = vec![ + AccountPostState::new(bridge.account.clone()), + AccountPostState::new(recipient_vault.account.clone()), + receipt_post, + ]; + + let mut bridge_for_vault = bridge; + bridge_for_vault.is_authorized = true; + let chained_calls = vec![ + ChainedCall::new( + vault_program_id, + vec![bridge_for_vault, recipient_vault], + &vault_core::Instruction::Transfer { + recipient_id, + amount: u128::from(amount), + }, + ) + .with_pda_seeds(vec![bridge_core::compute_bridge_seed()]), + ]; + (post_states, chained_calls) + } } Instruction::Withdraw { amount, @@ -89,13 +123,14 @@ fn main() { "Sender account must be owned by the authenticated transfer program" ); - vec![ChainedCall::new( + let chained_calls = vec![ChainedCall::new( auth_transfer_program_id, vec![sender, bridge], &authenticated_transfer_core::Instruction::Transfer { amount: u128::from(amount), }, - )] + )]; + (unchanged_post_states(&pre_states_clone), chained_calls) } }; diff --git a/lez/sequencer/core/src/block_publisher.rs b/lez/sequencer/core/src/block_publisher.rs index 8f2f2031..86d9dc0c 100644 --- a/lez/sequencer/core/src/block_publisher.rs +++ b/lez/sequencer/core/src/block_publisher.rs @@ -1,4 +1,4 @@ -use std::{pin::Pin, sync::Arc, time::Duration}; +use std::{sync::Arc, time::Duration}; use anyhow::{Context as _, Result, anyhow, ensure}; use common::block::Block; @@ -47,26 +47,15 @@ use crate::config::BedrockConfig; /// backpressure if the drive task stalls (reconnect, long backfill). const PUBLISH_INBOX_CAPACITY: usize = 32; -/// Sink for `Event::Published` checkpoints emitted by the drive task. -/// Caller is responsible for persistence (e.g. writing to rocksdb). -pub type CheckpointSink = Box; - -/// Sink for finalized L2 block ids derived from `Event::TxsFinalized` and -/// `Event::FinalizedInscriptions`. Caller is responsible for cleanup -/// (e.g. marking pending blocks as finalized in storage). -pub type FinalizedBlockSink = Box; - -/// Sink for finalized Bedrock deposit events. -pub type OnDepositEventSink = - Box Pin + Send>> + Send + 'static>; - -/// Sink for finalized Bedrock withdraw events. -pub type OnWithdrawEventSink = - Box Pin + Send>> + Send + 'static>; - -/// The channel delta the follow path consumes from one `Event::BlocksProcessed`, -/// with inscription payloads decoded into `(MsgId, Block)` pairs. +/// Everything one `Event::BlocksProcessed` carries, with inscription payloads +/// decoded into `(MsgId, Block)` pairs. +/// +/// One struct rather than a sink per effect, because the `checkpoint` and +/// everything it covers must reach the store in a single write. pub struct FollowUpdate { + /// Resume cursor for this event. Persist only together with the effects + /// below, never ahead of them. + pub checkpoint: SequencerCheckpoint, /// Inscriptions newly on the followed L1 branch, in channel order: they /// extend (or, after a reorg, replace part of) the `head` tier. pub adopted: Vec<(MsgId, Block)>, @@ -76,19 +65,24 @@ pub struct FollowUpdate { /// Inscriptions whose containing L1 block reached finality: their blocks /// move into the irreversible `final` tier. pub finalized: Vec<(MsgId, Block)>, + /// Finalized Bedrock deposit events, to record and mint on L2. + pub deposits: Vec, + /// Finalized Bedrock withdraw events, to reconcile against local intents. + pub withdrawals: Vec, } -/// Sink for the follow path: apply adopted/finalized blocks to chain state and -/// revert orphaned ones. +/// Sink for the follow path: apply the channel delta to chain state and +/// persist the whole event in one write. pub type OnFollowSink = Box; /// Commands the drive task executes with `&mut sequencer`. enum Command { - /// Publish an inscription (+ atomic withdrawals); responds with the assigned `MsgId`. + /// Publish an inscription (+ atomic withdrawals); responds with the assigned + /// `MsgId` and the checkpoint that now includes it as pending. Publish { inscription: Inscription, withdrawals: Vec, - resp: oneshot::Sender>, + resp: oneshot::Sender>, }, } @@ -96,25 +90,25 @@ type CommandSender = mpsc::Sender; #[expect(async_fn_in_trait, reason = "We don't care about Send/Sync here")] pub trait BlockPublisherTrait: Sized { - #[expect( - clippy::too_many_arguments, - reason = "Looks better than bundling all those callbacks into a struct" - )] async fn new( config: &BedrockConfig, bedrock_signing_key: Ed25519Key, resubmit_interval: Duration, initial_checkpoint: Option, - on_checkpoint: CheckpointSink, - on_finalized_block: FinalizedBlockSink, - on_deposit_event: OnDepositEventSink, - on_withdraw_event: OnWithdrawEventSink, on_follow: OnFollowSink, ) -> Result; - /// Publish a block and return the `MsgId` zone-sdk assigned its inscription. - /// Zone-sdk drives the actual submission and retries internally. - async fn publish_block(&self, block: &Block, withdrawals: Vec) -> Result; + /// Publish a block and return the `MsgId` zone-sdk assigned its inscription + /// together with the checkpoint that now holds it as pending. Zone-sdk + /// drives the actual submission and retries internally. + /// + /// The checkpoint must be persisted with the block — restoring an older one + /// drops the inscription from the pending set, and it is never resubmitted. + async fn publish_block( + &self, + block: &Block, + withdrawals: Vec, + ) -> Result<(MsgId, SequencerCheckpoint)>; fn channel_id(&self) -> ChannelId; @@ -168,10 +162,6 @@ impl BlockPublisherTrait for ZoneSdkPublisher { bedrock_signing_key: Ed25519Key, resubmit_interval: Duration, initial_checkpoint: Option, - on_checkpoint: CheckpointSink, - on_finalized_block: FinalizedBlockSink, - on_deposit_event: OnDepositEventSink, - on_withdraw_event: OnWithdrawEventSink, on_follow: OnFollowSink, ) -> Result { let basic_auth = config.auth.clone().map(Into::into); @@ -229,7 +219,7 @@ impl BlockPublisherTrait for ZoneSdkPublisher { }; let msg_result = published - .map(|(result, _checkpoint)| result.tx.inscription().this_msg); + .map(|(result, checkpoint)| (result.tx.inscription().this_msg, checkpoint)); match &msg_result { Ok(_) if withdraw_count == 0 => { info!("Published block with the size of {data_byte_size} bytes"); @@ -254,8 +244,6 @@ impl BlockPublisherTrait for ZoneSdkPublisher { channel_update, finalized, } => { - on_checkpoint(checkpoint); - let adopted = channel_update .adopted .iter() @@ -269,29 +257,35 @@ impl BlockPublisherTrait for ZoneSdkPublisher { .collect(); let mut finalized_blocks = Vec::new(); + let mut deposits = Vec::new(); + let mut withdrawals = Vec::new(); for op in finalized.into_iter().flat_map(|item| item.ops) { match op { FinalizedOp::Inscription(inscription) => { - if let Some((msg, block)) = + if let Some(entry) = block_from_inscription(&inscription) { - on_finalized_block(block.header.block_id); - finalized_blocks.push((msg, block)); + finalized_blocks.push(entry); } } - FinalizedOp::Deposit(deposit) => { - on_deposit_event(deposit).await; - } + FinalizedOp::Deposit(deposit) => deposits.push(deposit), FinalizedOp::Withdraw(withdraw) => { - on_withdraw_event(withdraw).await; + withdrawals.push(withdraw); } } } + // Nothing is awaited here: an await in this + // arm blocks the same task `publish_block` + // needs, and a non-turn sequencer never + // drains what it would be waiting on. on_follow(FollowUpdate { + checkpoint, adopted, orphaned, finalized: finalized_blocks, + deposits, + withdrawals, }); } Event::Ready => {} @@ -328,7 +322,11 @@ impl BlockPublisherTrait for ZoneSdkPublisher { }) } - async fn publish_block(&self, block: &Block, withdrawals: Vec) -> Result { + async fn publish_block( + &self, + block: &Block, + withdrawals: Vec, + ) -> Result<(MsgId, SequencerCheckpoint)> { let data = borsh::to_vec(block).context("Failed to serialize block")?; let data_bounded: Inscription = data .try_into() diff --git a/lez/sequencer/core/src/block_store.rs b/lez/sequencer/core/src/block_store.rs index 37394352..57c94990 100644 --- a/lez/sequencer/core/src/block_store.rs +++ b/lez/sequencer/core/src/block_store.rs @@ -155,13 +155,13 @@ impl SequencerStore { pub(crate) fn update( &mut self, block: &Block, - deposit_event_ids: &[HashType], - withdrawals: Vec, + withdrawals: &[WithdrawalReconciliationKey], state: &V03State, + checkpoint: Option<&[u8]>, ) -> DbResult<()> { let new_transactions_map = block_to_transactions_map(block); self.dbio - .atomic_update(block, deposit_event_ids, withdrawals, state)?; + .atomic_update(block, withdrawals, state, checkpoint)?; self.tx_hash_to_block_map.extend(new_transactions_map); Ok(()) } @@ -194,10 +194,12 @@ impl SequencerStore { Ok(Some(checkpoint)) } + /// Persists `checkpoint` on its own. Only valid when the effects it covers + /// are already durable — otherwise it must ride in the same write as them, + /// via [`storage::sequencer::StoreUpdate`]. pub fn set_zone_checkpoint(&self, checkpoint: &SequencerCheckpoint) -> Result<()> { - let bytes = - serde_json::to_vec(checkpoint).context("Failed to serialize zone-sdk checkpoint")?; - self.dbio.put_zone_sdk_checkpoint_bytes(&bytes)?; + self.dbio + .put_zone_sdk_checkpoint_bytes(&checkpoint_bytes(checkpoint)?)?; Ok(()) } @@ -211,23 +213,15 @@ impl SequencerStore { self.dbio.put_zone_anchor(anchor) } - pub fn get_unfulfilled_deposit_events(&self) -> DbResult> { + pub fn get_pending_deposit_events(&self) -> DbResult> { self.dbio.get_pending_deposit_events() } +} - pub fn is_deposit_event_submitted(&self, deposit_op_id: HashType) -> DbResult { - self.dbio.is_deposit_event_submitted(deposit_op_id) - } - - /// Marks the given deposit events submitted in `block_id`, in one write. - pub fn mark_deposit_events_submitted( - &self, - deposit_op_ids: &[HashType], - submitted_block_id: u64, - ) -> DbResult<()> { - self.dbio - .mark_deposit_events_submitted(deposit_op_ids, submitted_block_id) - } +/// The checkpoint's on-disk encoding. `serde_json` because `SequencerCheckpoint` +/// derives serde but not borsh; paired with `get_zone_checkpoint`'s decode. +pub(crate) fn checkpoint_bytes(checkpoint: &SequencerCheckpoint) -> Result> { + serde_json::to_vec(checkpoint).context("Failed to serialize zone-sdk checkpoint") } pub(crate) fn block_to_transactions_map(block: &Block) -> HashMap { @@ -278,9 +272,7 @@ mod tests { assert_eq!(None, retrieved_tx); // Add the block with the transaction let dummy_state = V03State::new(); - node_store - .update(&block, &[], vec![], &dummy_state) - .unwrap(); + node_store.update(&block, &[], &dummy_state, None).unwrap(); // Try again let output = node_store.get_transaction_by_hash(tx.hash()); assert_eq!(Some((tx, 1)), output); @@ -345,9 +337,7 @@ mod tests { let block_hash = block.header.hash; let dummy_state = V03State::new(); - node_store - .update(&block, &[], vec![], &dummy_state) - .unwrap(); + node_store.update(&block, &[], &dummy_state, None).unwrap(); // Verify that the latest block meta now equals the new block's hash let latest_meta = node_store.latest_block_meta().unwrap().unwrap(); @@ -383,9 +373,7 @@ mod tests { let block_id = block.header.block_id; let dummy_state = V03State::new(); - node_store - .update(&block, &[], vec![], &dummy_state) - .unwrap(); + node_store.update(&block, &[], &dummy_state, None).unwrap(); // Verify initial status is Pending let retrieved_block = node_store.get_block_at_id(block_id).unwrap().unwrap(); @@ -434,7 +422,7 @@ mod tests { // Add a new block let block = common::test_utils::produce_dummy_block(1, None, vec![tx.clone()]); node_store - .update(&block, &[], vec![], &V03State::new()) + .update(&block, &[], &V03State::new(), None) .unwrap(); } diff --git a/lez/sequencer/core/src/lib.rs b/lez/sequencer/core/src/lib.rs index f2507430..5ed0e50c 100644 --- a/lez/sequencer/core/src/lib.rs +++ b/lez/sequencer/core/src/lib.rs @@ -1,4 +1,5 @@ use std::{ + collections::VecDeque, path::Path, sync::{Arc, Mutex}, time::Instant, @@ -31,7 +32,7 @@ pub use mock::SequencerCoreWithMockClients; use num_bigint::BigUint; pub use storage::error::DbError; use storage::sequencer::{ - RocksDBIO, + RocksDBIO, StoreUpdate, sequencer_cells::{PendingDepositEventRecord, WithdrawalReconciliationKey, ZoneAnchorRecord}, }; @@ -188,17 +189,12 @@ impl SequencerCore { 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 @@ -241,8 +237,9 @@ impl SequencerCore { "First pending block on fresh start should be the genesis block" ); + let mut last_checkpoint = None; for block in &pending_blocks { - block_publisher + let (_msg, checkpoint) = block_publisher .publish_block(block, vec![]) .await .unwrap_or_else(|err| { @@ -251,6 +248,16 @@ impl SequencerCore { block.header.block_id ) }); + last_checkpoint = Some(checkpoint); + } + + // These blocks are already stored, so only the sdk's pending set + // moved. Checkpoints are cumulative — persisting just the last one + // is both sufficient and the only way to keep this loop linear. + if let Some(checkpoint) = last_checkpoint { + store + .set_zone_checkpoint(&checkpoint) + .expect("Failed to persist checkpoint after republishing on fresh start"); } } @@ -448,171 +455,36 @@ impl SequencerCore { } } - // 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 + // A reconstructed block is finalized, so any deposit it mints is + // permanently reflected in state (its receipt PDA); drop the pending + // record backfill may have re-delivered, so the drain stops re-minting. + let finalized_deposit_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")?; + // 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 + // itself landed. + let head_tip = chain.head_tip().map(|head| BlockMeta::from(&head)); + let final_meta = chain.final_tip().map(|meta| BlockMeta::from(&meta)); store - .set_zone_anchor(&record) - .context("Failed to persist zone anchor")?; + .dbio() + .store_update(&StoreUpdate { + blocks: &[(block, true)], + head_tip: head_tip.as_ref(), + final_snapshot: final_meta.as_ref().map(|meta| (chain.final_state(), meta)), + remove_deposit_records: &finalized_deposit_ids, + zone_anchor: Some(&record), + ..StoreUpdate::new(chain.head_state()) + }) + .context("Failed to persist reconstructed block")?; + Ok(()) } - fn on_checkpoint(dbio: Arc) -> 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) -> 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, - 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) -> 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, @@ -626,21 +498,17 @@ impl SequencerCore { /// Produces a new block from mempool transactions and publishes it via zone-sdk. pub async fn produce_new_block(&mut self) -> Result { - let BlockWithMeta { - block, - deposit_event_ids, - withdrawals, - } = self + let BlockWithMeta { block, withdrawals } = self .build_block_from_mempool() .context("Failed to build block from mempool transactions")?; - let withdrawal_reconciliation_keys = withdrawals + let withdrawal_reconciliation_keys: Vec<_> = withdrawals .iter() .map(|withdraw| withdraw_event_reconciliation_key(&withdraw.outputs)) - .collect::>() + .collect::>>() .context("Failed to build reconciliation keys for block withdrawals")?; - let this_msg = self + let (this_msg, checkpoint) = self .block_publisher .publish_block(&block, withdrawals) .await @@ -649,8 +517,8 @@ impl SequencerCore { self.record_produced_block( this_msg, &block, - &deposit_event_ids, - withdrawal_reconciliation_keys, + &withdrawal_reconciliation_keys, + &checkpoint, )?; Ok(block.header.block_id) @@ -669,9 +537,11 @@ impl SequencerCore { &mut self, this_msg: MsgId, block: &Block, - deposit_event_ids: &[HashType], - withdrawal_reconciliation_keys: Vec, + withdrawal_reconciliation_keys: &[WithdrawalReconciliationKey], + checkpoint: &block_publisher::SequencerCheckpoint, ) -> Result<()> { + let checkpoint_bytes = block_store::checkpoint_bytes(checkpoint)?; + let mut chain = self.chain.lock().expect("chain state mutex poisoned"); match chain.apply_produced(this_msg, block) { AcceptOutcome::Applied => { @@ -679,11 +549,14 @@ impl SequencerCore { // with the follow path. self.store.update( block, - deposit_event_ids, withdrawal_reconciliation_keys, chain.head_state(), + Some(&checkpoint_bytes), )?; } + // Neither branch persists anything, checkpoint included: the + // inscription it holds as pending belongs to a block that is not + // ours to keep. AcceptOutcome::AlreadyApplied => { warn!( "Produced block {} lost a competing-write race, skipping persistence", @@ -704,20 +577,14 @@ impl SequencerCore { /// 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, withdrawals: &mut Vec, - ) -> Result { + ) -> bool { let tx_hash = tx.hash(); match origin { TransactionOrigin::User => { @@ -727,7 +594,7 @@ impl SequencerCore { error!( "Transaction with hash {tx_hash} failed execution check with error: {err:#?}, skipping it", ); - return Ok(false); + return false; } }; @@ -742,25 +609,30 @@ impl SequencerCore { 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); + // Bridge deposits are deduped by their receipt PDA in chain + // state (drained only when unminted, no-op on replay), so no + // node-local guard is needed here. + // + // Skip-and-log rather than propagate: a drained deposit is + // re-fed from the store every turn and only finality removes it, + // so a `?` here would let a single unexecutable mint (e.g. a + // bridge escrow under-funded relative to the L1 deposit, which + // every sequencer hits identically) abort production on all of + // them forever. Skipping keeps the record queued for retry + // without halting the node. + if let Err(err) = + state.transition_from_public_transaction(public_tx, block_height, timestamp) + { + error!( + "Sequencer-generated transaction {tx_hash} failed execution: {err:#?}, skipping it", + ); + return false; } - - 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) + true } fn build_block_from_mempool(&mut self) -> Result { @@ -780,9 +652,36 @@ impl SequencerCore { }; let mut valid_transactions = Vec::new(); - let mut deposit_event_ids = Vec::new(); let mut withdrawals = Vec::new(); + // Bridge deposit mints are drained from the store, not the mempool: the + // follow path records the event durably but cannot enqueue the mint + // itself (it runs on the publisher's drive task, where an await stalls + // the very task production needs). Draining here also subsumes the old + // startup replay. + // + // Skip any deposit whose receipt PDA already exists in the state we + // 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 = self + .store + .get_pending_deposit_events() + .context("Failed to load pending deposit events")? + .into_iter() + .filter(|record| !deposit_already_minted(&working_state, record.deposit_op_id)) + .filter_map(|record| { + build_bridge_deposit_tx_from_event(&record) + .inspect_err(|err| { + warn!( + "Skipping pending deposit event {} due to tx build failure: {err:#}", + hex::encode(record.deposit_op_id) + ); + }) + .ok() + }) + .collect(); + let max_block_size = usize::try_from(self.sequencer_config.max_block_size.as_u64()) .expect("`max_block_size` should fit into usize"); @@ -792,7 +691,14 @@ impl SequencerCore { 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() { + // 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 + .pop_front() + .map(|tx| (TransactionOrigin::Sequencer, tx, true)) + .or_else(|| self.mempool.pop().map(|(origin, tx)| (origin, tx, false))) + { let tx_hash = tx.hash(); let temp_valid_transactions = [ @@ -817,20 +723,22 @@ impl SequencerCore { "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)); + // A deposit mint needs no requeue: its record stays unfulfilled + // in the store and is drained again on the next turn. + if !from_store { + 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); } @@ -861,11 +769,7 @@ impl SequencerCore { now.elapsed().as_secs() ); - Ok(BlockWithMeta { - block, - deposit_event_ids, - withdrawals, - }) + Ok(BlockWithMeta { block, withdrawals }) } /// Reads the current head state under the lock without cloning it, so callers @@ -896,10 +800,9 @@ impl SequencerCore { } /// 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. + /// finalized. Idempotent. Production no longer calls this: finalization + /// flips now ride the follow path's atomic write via + /// [`StoreUpdate::finalized_up_to`]. 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<()> { @@ -941,15 +844,26 @@ impl SequencerCore { struct BlockWithMeta { block: Block, - deposit_event_ids: Vec, withdrawals: Vec, } +/// Whether `deposit_op_id`'s mint is already reflected in `state` — its receipt +/// PDA exists. The receipt is the exactly-once ledger the bridge program keeps. +fn deposit_already_minted(state: &lee::V03State, deposit_op_id: HashType) -> bool { + let receipt_id = + bridge_core::deposit_receipt_account_id(programs::bridge().id(), deposit_op_id.0); + state + .get_account_by_id_ref(receipt_id) + .is_some_and(|receipt| *receipt != lee::Account::default()) +} + /// 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. /// +/// Everything the event produced lands in one write — see [`StoreUpdate`]. +/// /// 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 @@ -962,14 +876,42 @@ fn apply_follow_update( update: block_publisher::FollowUpdate, ) { let block_publisher::FollowUpdate { + checkpoint, adopted, orphaned, finalized, + deposits, + withdrawals, } = update; + let checkpoint_bytes = block_store::checkpoint_bytes(&checkpoint) + .unwrap_or_else(|err| panic!("Failed to serialize zone-sdk checkpoint: {err:#}")); + + // NOTE: Theoretically Zone SDK may re-deliver an already seen deposit or + // finalization. Both are idempotent here: a deposit already on record is + // not re-appended, and a finalization only ever moves the tier forward. + let deposit_records: Vec = + deposits.iter().map(pending_deposit_event_record).collect(); + + // A withdraw whose outputs we cannot read has no counter to reconcile + // against; log and drop it rather than fail the whole update. + let consumed_withdrawals: Vec = withdrawals + .iter() + .filter_map(|withdraw| { + withdraw_event_reconciliation_key(&withdraw.op.outputs) + .inspect_err(|err| { + error!( + "Failed to build reconciliation key for Bedrock Withdraw event with tx_hash {}: {err:#}", + hex::encode(withdraw.tx_hash.as_ref()) + ); + }) + .ok() + }) + .collect(); + // 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 (resubmit_txs, outcome) = { let mut chain = chain.lock().expect("chain state mutex poisoned"); // User txs of orphaned blocks, returned to the mempool below. @@ -987,16 +929,28 @@ fn apply_follow_update( .map(|((_, block), _)| (block, false)) .collect(); + // Only blocks the final tier holds drive the bookkeeping below: a parked + // one never became irreversible, so marking blocks finalized through it + // or dropping its deposit records would lose them for good. + let mut irreversible: Vec<&Block> = Vec::new(); 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; + // FIXME: thread the finalized inscription's L1 slot instead of + // `Slot::from(0)`; only used for the invalid-finalized stall. + // logos-blockchain PR #3147 surfaces it as `FinalizedTx.l1_slot` — + // wire it through `FollowUpdate::finalized` once the zone-sdk pin is + // bumped past that (a separate PR). + match chain.apply_finalized(*this_msg, block, Slot::from(0)) { + AcceptOutcome::Applied => { + to_persist.push((block, true)); + irreversible.push(block); + final_advanced = true; + } + // A re-delivery of a block the final tier already holds: no new + // payload and the tier does not move, but it is irreversible all + // the same, so it still settles its deposits. + AcceptOutcome::AlreadyApplied => irreversible.push(block), + AcceptOutcome::Parked(_) | AcceptOutcome::RetryableFailure(_) => {} } } // Snapshot the advanced final tier so a restart re-anchors on it. @@ -1006,36 +960,65 @@ fn apply_follow_update( }); 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:#}")); + // Every block at or below the highest finalized one is irreversible, so + // stored blocks there can be marked finalized. + let last_finalized = irreversible.iter().map(|block| block.header.block_id).max(); - resubmit_txs + // A deposit observed in a finalized block is permanently minted (its + // receipt is now in the irreversible tier), so its pending record can be + // dropped. Keyed by op id, not block id: a record only goes once its own + // deposit finalizes, never because some other block finalized at its + // height. + let finalized_deposit_ids: Vec = irreversible + .iter() + .flat_map(|block| block.body.transactions.iter()) + .filter_map(extract_bridge_deposit_id) + .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. + let outcome = dbio + .store_update(&StoreUpdate { + checkpoint: Some(&checkpoint_bytes), + blocks: &to_persist, + head_tip: head_tip.as_ref(), + final_snapshot: final_meta.as_ref().map(|meta| (chain.final_state(), meta)), + finalized_up_to: last_finalized, + new_deposit_events: &deposit_records, + remove_deposit_records: &finalized_deposit_ids, + consumed_withdrawals: &consumed_withdrawals, + ..StoreUpdate::new(chain.head_state()) + }) + .unwrap_or_else(|err| panic!("Failed to persist follow update: {err:#}")); + + (resubmit_txs, outcome) }; + if outcome.accepted_deposits > 0 { + info!( + "Recorded {} Bedrock Deposit event(s); their mints are drained from the store on our next turn", + outcome.accepted_deposits + ); + } + for withdrawal in &outcome.unmatched_withdrawals { + warn!( + "Unexpected Bedrock Withdraw event of {} to {}: no matching unseen withdraw found", + withdrawal.amount, + hex::encode(withdrawal.bedrock_account_pk) + ); + } + // 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. + // We use [`try_push`] here because this is called from the publisher's + // drive task, and only block production drains the mempool. A blocking + // push would stall the drive task, and a sequencer that is not on turn + // never produces — so nothing would ever drain it again. + // + // TODO: a full mempool still drops the transaction; a durable resubmit + // queue is a follow-up. for tx in resubmit_txs { let tx_hash = tx.hash(); if let Err(err) = mempool_handle.try_push((TransactionOrigin::User, tx)) { @@ -1044,55 +1027,6 @@ fn apply_follow_update( } } -/// 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 = 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 the bridge-lock holdings, /// the only accounts seeded outside any transaction. Cross-zone config is seeded /// by genesis `InitConfig` transactions and reconstructed by replaying them. @@ -1232,7 +1166,6 @@ fn pending_deposit_event_record(deposit: &DepositInfo) -> PendingDepositEventRec source_tx_hash: HashType(deposit.tx_hash.0), amount: deposit.amount, metadata: deposit.metadata.clone().into(), - submitted_in_block_id: None, } } @@ -1244,10 +1177,18 @@ fn build_bridge_deposit_tx_from_event(event: &PendingDepositEventRecord) -> Resu let vault_program_id = programs::vault().id(); let recipient_vault_id = vault_core::compute_vault_account_id(vault_program_id, metadata.recipient_id); + // The receipt PDA carries the exactly-once check: the program reads it to + // detect a replay, so it must be in the tx's account list. + let receipt_id = + bridge_core::deposit_receipt_account_id(bridge_program_id, event.deposit_op_id.0); let message = Message::try_new( bridge_program_id, - vec![system_accounts::bridge_account_id(), recipient_vault_id], + vec![ + system_accounts::bridge_account_id(), + recipient_vault_id, + receipt_id, + ], vec![], bridge_core::Instruction::Deposit { l1_deposit_op_id: event.deposit_op_id.0, diff --git a/lez/sequencer/core/src/mock.rs b/lez/sequencer/core/src/mock.rs index 4f2552b7..5b126f46 100644 --- a/lez/sequencer/core/src/mock.rs +++ b/lez/sequencer/core/src/mock.rs @@ -3,16 +3,16 @@ use std::time::Duration; use anyhow::Result; use common::block::Block; use futures::Stream; -use logos_blockchain_core::mantle::ops::channel::{ChannelId, MsgId}; +use logos_blockchain_core::{ + header::HeaderId, + mantle::ops::channel::{ChannelId, MsgId}, +}; use logos_blockchain_key_management_system_service::keys::Ed25519Key; use logos_blockchain_zone_sdk::{Slot, ZoneMessage, sequencer::WithdrawArg}; use tokio_util::sync::CancellationToken; use crate::{ - block_publisher::{ - BlockPublisherTrait, CheckpointSink, FinalizedBlockSink, OnDepositEventSink, OnFollowSink, - OnWithdrawEventSink, SequencerCheckpoint, - }, + block_publisher::{BlockPublisherTrait, OnFollowSink, SequencerCheckpoint}, config::BedrockConfig, }; @@ -54,16 +54,15 @@ impl BlockPublisherTrait for MockBlockPublisher { _bedrock_signing_key: Ed25519Key, _resubmit_interval: Duration, _initial_checkpoint: Option, - _on_checkpoint: CheckpointSink, - _on_finalized_block: FinalizedBlockSink, - _on_deposit_event: OnDepositEventSink, - _on_withdraw_event: OnWithdrawEventSink, _on_follow: OnFollowSink, ) -> Result { Ok(Self { channel_id: config.channel_id, driver_cancellation: CancellationToken::new(), - tip_slot: None, + // An existing but empty channel: `None` means *missing*, which the + // startup guard reads as a wiped Bedrock. Tests that want that say + // so via [`Self::with_canned_channel`]. + tip_slot: Some(Slot::from(0)), messages: Vec::new(), }) } @@ -72,11 +71,11 @@ impl BlockPublisherTrait for MockBlockPublisher { &self, block: &Block, _bridge_withdrawals: Vec, - ) -> Result { + ) -> Result<(MsgId, SequencerCheckpoint)> { // Deterministic per-block id so head dedup behaves in tests. // // TODO: should we allow more "mockability" here? - Ok(MsgId::from(block.header.hash.0)) + Ok((MsgId::from(block.header.hash.0), mock_checkpoint())) } fn channel_id(&self) -> ChannelId { @@ -108,3 +107,16 @@ impl BlockPublisherTrait for MockBlockPublisher { Ok(futures::stream::iter(messages)) } } + +/// A zeroed checkpoint, for [`MockBlockPublisher::publish_block`] and for tests +/// building a [`crate::block_publisher::FollowUpdate`]. Tests only assert *that* +/// a checkpoint was persisted alongside its effects, never what is in it. +#[must_use] +pub(crate) fn mock_checkpoint() -> SequencerCheckpoint { + SequencerCheckpoint { + last_msg_id: MsgId::from([0; 32]), + pending_txs: Vec::new(), + lib: HeaderId::from([0; 32]), + lib_slot: Slot::from(0), + } +} diff --git a/lez/sequencer/core/src/tests.rs b/lez/sequencer/core/src/tests.rs index 63e44678..36136c55 100644 --- a/lez/sequencer/core/src/tests.rs +++ b/lez/sequencer/core/src/tests.rs @@ -22,7 +22,12 @@ use lee_core::{ account::{AccountWithMetadata, Nonce}, program::PdaSeed, }; -use logos_blockchain_core::mantle::ops::channel::{ChannelId, MsgId}; +use logos_blockchain_core::mantle::{ + ledger::Inputs, + ops::channel::{ChannelId, MsgId, deposit::Metadata}, + tx::TxHash, +}; +use logos_blockchain_zone_sdk::sequencer::DepositInfo; use mempool::MemPoolHandle; use storage::sequencer::sequencer_cells::PendingDepositEventRecord; use tempfile::tempdir; @@ -33,9 +38,9 @@ use crate::{ block_publisher::FollowUpdate, block_store::SequencerStore, build_bridge_deposit_tx_from_event, build_genesis_state, - config::{BedrockConfig, SequencerConfig}, - is_sequencer_only_program, - mock::SequencerCoreWithMockClients, + config::{BedrockConfig, GenesisAction, SequencerConfig}, + deposit_already_minted, is_sequencer_only_program, + mock::{SequencerCoreWithMockClients, mock_checkpoint}, resubmittable_txs, }; @@ -46,6 +51,19 @@ struct DepositMetadataForEncoding { recipient_id: lee::AccountId, } +/// A follow update carrying nothing, to fill in the fields a test does not +/// exercise via `..empty_follow_update()`. +fn empty_follow_update() -> FollowUpdate { + FollowUpdate { + checkpoint: mock_checkpoint(), + adopted: Vec::new(), + orphaned: Vec::new(), + finalized: Vec::new(), + deposits: Vec::new(), + withdrawals: Vec::new(), + } +} + fn setup_sequencer_config() -> SequencerConfig { let tempdir = tempfile::tempdir().unwrap(); let home = tempdir.path().to_path_buf(); @@ -211,8 +229,10 @@ async fn start_from_config_panics_when_db_open_returns_non_not_found_error() { } #[tokio::test] -async fn start_from_config_replays_unfulfilled_deposit_events_from_db() { - let config = setup_sequencer_config(); +async fn unfulfilled_deposit_events_are_drained_from_the_store_on_production() { + let mut config = setup_sequencer_config(); + // The mint moves funds out of the bridge account, so it has to hold some. + config.genesis = vec![GenesisAction::SupplyBridgeAccount { balance: 1_000_000 }]; let deposit_op_id = [13_u8; 32]; let expected_amount = 1_u64; let recipient_id = initial_public_user_accounts()[0].account_id; @@ -227,7 +247,6 @@ async fn start_from_config_replays_unfulfilled_deposit_events_from_db() { source_tx_hash: HashType([7_u8; 32]), amount: expected_amount, metadata: borsh::to_vec(&DepositMetadataForEncoding { recipient_id }).unwrap(), - submitted_in_block_id: None, }; { @@ -241,36 +260,221 @@ async fn start_from_config_replays_unfulfilled_deposit_events_from_db() { assert!(inserted); } + // The mint never goes through the mempool: the record is the queue, and + // production drains it. That is what makes a restart — or a follow event + // arriving while a full mempool would have dropped the push — lossless. let (mut sequencer, _mempool_handle) = SequencerCoreWithMockClients::start_from_config(config).await; + assert!( + sequencer.mempool.pop().is_none(), + "deposit mints are drained from the store, never queued in the mempool" + ); - let (origin, tx) = tokio::time::timeout(Duration::from_secs(5), async { - loop { - if let Some((origin, tx)) = sequencer.mempool.pop() { - return (origin, tx); - } + 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!( + block + .body + .transactions + .iter() + .any(|tx| tx_is_bridge_deposit(tx, deposit_op_id, expected_amount)), + "the drained deposit mint should be included in the produced block" + ); - tokio::time::sleep(Duration::from_millis(100)).await; - } + // The record stays until its deposit finalizes; exactly-once is enforced by + // the receipt PDA now in head state, not by any marker on the record. + assert!( + sequencer + .store + .get_pending_deposit_events() + .unwrap() + .iter() + .any(|event| event.deposit_op_id == HashType(deposit_op_id)), + "the record remains until the deposit finalizes" + ); + assert!( + sequencer.with_state(|state| deposit_already_minted(state, HashType(deposit_op_id))), + "the deposit's receipt PDA marks it minted in head state" + ); +} + +#[tokio::test] +async fn a_drained_deposit_is_not_minted_twice_across_turns() { + let mut config = setup_sequencer_config(); + config.genesis = vec![GenesisAction::SupplyBridgeAccount { balance: 1_000_000 }]; + let deposit_op_id = [17_u8; 32]; + let recipient_id = initial_public_user_accounts()[0].account_id; + + let (mut sequencer, _mempool_handle) = + SequencerCoreWithMockClients::start_from_config(config).await; + sequencer + .store + .dbio() + .add_pending_deposit_event(PendingDepositEventRecord { + deposit_op_id: HashType(deposit_op_id), + source_tx_hash: HashType([7_u8; 32]), + amount: 1, + metadata: borsh::to_vec(&DepositMetadataForEncoding { recipient_id }).unwrap(), + }) + .unwrap(); + + let first = sequencer.produce_new_block().await.unwrap(); + let second = sequencer.produce_new_block().await.unwrap(); + + let minted_in = |block_id: u64| { + sequencer + .store + .get_block_at_id(block_id) + .unwrap() + .expect("produced block is stored") + .body + .transactions + .iter() + .filter(|tx| tx_is_bridge_deposit(tx, deposit_op_id, 1)) + .count() + }; + + assert_eq!(minted_in(first), 1); + assert_eq!( + minted_in(second), + 0, + "the receipt PDA from the first mint must keep the drain from re-minting" + ); +} + +#[tokio::test] +async fn an_orphaned_deposit_is_reminted_exactly_once_in_the_replacement() { + // Manifestation 2 from #639: a deposit-carrying block is orphaned. Recovery + // rests entirely on the receipt PDA reverting with the block — no requeue, + // no bookkeeping of our own — so the still-pending record is drained again + // on the next turn and the vault is credited exactly once across the reorg. + let mut config = setup_sequencer_config(); + config.genesis = vec![GenesisAction::SupplyBridgeAccount { balance: 1_000_000 }]; + let recipient_id = initial_public_user_accounts()[0].account_id; + let deposit_op_id = [0x2c_u8; 32]; + let amount = 500_u64; + + let (mut sequencer, mempool_handle) = + SequencerCoreWithMockClients::start_from_config(config).await; + sequencer + .store + .dbio() + .add_pending_deposit_event(PendingDepositEventRecord { + deposit_op_id: HashType(deposit_op_id), + source_tx_hash: HashType([7_u8; 32]), + amount, + metadata: borsh::to_vec(&DepositMetadataForEncoding { recipient_id }).unwrap(), + }) + .unwrap(); + + // Produce the block that mints the deposit; its receipt marks it minted. + sequencer.produce_new_block().await.unwrap(); + let minted_block = sequencer.store.get_block_at_id(2).unwrap().unwrap(); + assert!( + sequencer.with_state(|s| deposit_already_minted(s, HashType(deposit_op_id))), + "the first mint claims the receipt in head state" + ); + + // Orphan that block. The receipt reverts with it — nothing else tracks the + // mint — so the deposit reads as unminted again. + apply_follow_update( + &sequencer.store.dbio(), + &sequencer.chain(), + &mempool_handle, + FollowUpdate { + adopted: vec![], + orphaned: vec![(MsgId::from(minted_block.header.hash.0), minted_block)], + ..empty_follow_update() + }, + ); + assert_eq!(sequencer.chain_height(), 1, "the minting block is orphaned"); + assert!( + !sequencer.with_state(|s| deposit_already_minted(s, HashType(deposit_op_id))), + "the receipt reverts with the orphaned block" + ); + + // Next turn: the still-pending record is drained and re-minted on the new + // head, exactly once. + let replacement = sequencer.produce_new_block().await.unwrap(); + let mints = sequencer + .store + .get_block_at_id(replacement) + .unwrap() + .expect("replacement block is stored") + .body + .transactions + .iter() + .filter(|tx| tx_is_bridge_deposit(tx, deposit_op_id, amount)) + .count(); + assert_eq!( + mints, 1, + "the deposit is re-minted exactly once after the orphan" + ); + let vault_id = vault_core::compute_vault_account_id(programs::vault().id(), recipient_id); + assert_eq!( + sequencer.with_state(|s| s.get_account_by_id(vault_id).balance), + u128::from(amount), + "the vault is credited exactly once across the reorg" + ); +} + +#[tokio::test] +async fn a_replayed_deposit_mint_no_ops_in_the_guest() { + // Runs the bridge guest directly with a pre-existing receipt — the replay + // no-op branch the exactly-once guarantee rests on. The store drain filters + // duplicates out before the program executes, so this is the only test that + // reaches that branch; applying the same mint twice asserts the second is a + // no-op (credited once) rather than an error. + let mut config = setup_sequencer_config(); + config.genesis = vec![GenesisAction::SupplyBridgeAccount { balance: 1_000_000 }]; + let recipient_id = initial_public_user_accounts()[0].account_id; + let deposit_op_id = [0x5a_u8; 32]; + let amount = 500_u64; + + let (sequencer, _mempool_handle) = + SequencerCoreWithMockClients::start_from_config(config).await; + + let deposit_tx = build_bridge_deposit_tx_from_event(&PendingDepositEventRecord { + deposit_op_id: HashType(deposit_op_id), + source_tx_hash: HashType([7_u8; 32]), + amount, + metadata: borsh::to_vec(&DepositMetadataForEncoding { recipient_id }).unwrap(), }) - .await - .expect("Timed out waiting for pending deposit event to be replayed into mempool"); + .unwrap(); + let LeeTransaction::Public(public_tx) = &deposit_tx else { + panic!("bridge deposit tx is public"); + }; - match origin { - TransactionOrigin::Sequencer => {} - TransactionOrigin::User => { - panic!("Unexpected user transaction in empty mempool replay test") - } - } + let vault_id = vault_core::compute_vault_account_id(programs::vault().id(), recipient_id); + let mut state = sequencer.chain().lock().unwrap().head_state().clone(); - assert!(tx_is_bridge_deposit(&tx, deposit_op_id, expected_amount)); + // First mint: claims the receipt and credits the recipient vault. + state + .transition_from_public_transaction(public_tx, 1, 0) + .expect("first mint executes"); + assert_eq!( + state.get_account_by_id(vault_id).balance, + u128::from(amount) + ); + assert!( + deposit_already_minted(&state, HashType(deposit_op_id)), + "the first mint claims the receipt PDA" + ); - let pending_events = sequencer.store.get_unfulfilled_deposit_events().unwrap(); - let replayed_event = pending_events - .into_iter() - .find(|event| event.deposit_op_id == HashType(deposit_op_id)) - .expect("Pending deposit event should remain in DB until included in a block"); - assert!(replayed_event.submitted_in_block_id.is_none()); + // Replay the identical mint. The guest sees the receipt already exists and + // no-ops instead of failing, so the vault is credited exactly once. + state + .transition_from_public_transaction(public_tx, 2, 0) + .expect("a replayed deposit is a no-op, not an error"); + assert_eq!( + state.get_account_by_id(vault_id).balance, + u128::from(amount), + "a replayed deposit must not re-credit the vault" + ); } #[test] @@ -1287,7 +1491,6 @@ fn resubmittable_txs_drops_clock_and_bridge_deposits() { recipient_id: initial_public_user_accounts()[0].account_id, }) .unwrap(), - submitted_in_block_id: None, }) .unwrap(); let withdraw_tx = { @@ -1334,6 +1537,67 @@ fn resubmittable_txs_of_blocks_without_user_txs_is_empty() { assert!(resubmittable_txs(&clock_only).is_empty()); } +#[tokio::test] +async fn follow_update_persists_the_checkpoint_with_its_effects() { + let config = setup_sequencer_config(); + let (sequencer, mempool_handle) = SequencerCoreWithMockClients::start_from_config(config).await; + let genesis_meta = sequencer + .store + .latest_block_meta() + .unwrap() + .expect("genesis meta is set"); + + let peer_block = common::test_utils::produce_dummy_block(2, Some(genesis_meta.hash), vec![]); + apply_follow_update( + &sequencer.store.dbio(), + &sequencer.chain(), + &mempool_handle, + FollowUpdate { + adopted: vec![(MsgId::from([1; 32]), peer_block)], + ..empty_follow_update() + }, + ); + + // The checkpoint is the sdk resume cursor; landing it without the block + // would let a restart stream past a block the store never got. + assert!( + sequencer.store.get_zone_checkpoint().unwrap().is_some(), + "the event's checkpoint must be persisted alongside the block it covers" + ); + assert!(sequencer.store.get_block_at_id(2).unwrap().is_some()); +} + +#[tokio::test] +async fn follow_update_records_deposits_for_the_production_drain() { + let config = setup_sequencer_config(); + let (sequencer, mempool_handle) = SequencerCoreWithMockClients::start_from_config(config).await; + + let recipient_id = initial_public_user_accounts()[0].account_id; + let metadata = borsh::to_vec(&DepositMetadataForEncoding { recipient_id }).unwrap(); + let deposit = DepositInfo { + op_id: [21; 32], + tx_hash: TxHash::from([9; 32]), + channel_id: ChannelId::from([0; 32]), + inputs: Inputs::empty(), + amount: 5, + metadata: Metadata::try_from(metadata).expect("deposit metadata fits"), + }; + + apply_follow_update( + &sequencer.store.dbio(), + &sequencer.chain(), + &mempool_handle, + FollowUpdate { + deposits: vec![deposit], + ..empty_follow_update() + }, + ); + + let pending = sequencer.store.get_pending_deposit_events().unwrap(); + assert_eq!(pending.len(), 1); + assert_eq!(pending[0].deposit_op_id, HashType([21; 32])); +} + #[tokio::test] async fn follow_adopted_peer_block_applies_and_persists() { let config = setup_sequencer_config(); @@ -1361,8 +1625,7 @@ async fn follow_adopted_peer_block_applies_and_persists() { &mempool_handle, FollowUpdate { adopted: vec![(MsgId::from([1; 32]), peer_block.clone())], - orphaned: vec![], - finalized: vec![], + ..empty_follow_update() }, ); @@ -1409,8 +1672,7 @@ async fn follow_redelivery_of_own_block_is_deduped() { &mempool_handle, FollowUpdate { adopted: vec![(MsgId::from(block2.header.hash.0), block2)], - orphaned: vec![], - finalized: vec![], + ..empty_follow_update() }, ); @@ -1451,7 +1713,7 @@ async fn follow_orphan_reverts_head_and_requeues_user_txs() { FollowUpdate { adopted: vec![], orphaned: vec![(MsgId::from(block2.header.hash.0), block2)], - finalized: vec![], + ..empty_follow_update() }, ); @@ -1495,6 +1757,7 @@ async fn follow_finalized_own_block_moves_final_tier_and_marks_store() { adopted: vec![], orphaned: vec![], finalized: vec![(MsgId::from(block2.header.hash.0), block2)], + ..empty_follow_update() }, ); @@ -1532,6 +1795,7 @@ async fn follow_finalized_backfill_block_is_applied_and_marked_finalized() { adopted: vec![], orphaned: vec![], finalized: vec![(MsgId::from([2; 32]), peer_block.clone())], + ..empty_follow_update() }, ); @@ -1549,6 +1813,76 @@ async fn follow_finalized_backfill_block_is_applied_and_marked_finalized() { assert!(matches!(stored.bedrock_status, BedrockStatus::Finalized)); } +#[tokio::test] +async fn parked_finalized_block_neither_sweeps_the_store_nor_drops_its_deposit_record() { + let config = setup_sequencer_config(); + let (mut sequencer, mempool_handle) = + SequencerCoreWithMockClients::start_from_config(config).await; + + // A produced block at head, still pending on the channel. + let tx = common::test_utils::produce_dummy_empty_transaction(); + mempool_handle + .push((TransactionOrigin::User, tx)) + .await + .unwrap(); + sequencer.produce_new_block().await.unwrap(); + + let deposit_op_id = HashType([21; 32]); + let record = PendingDepositEventRecord { + deposit_op_id, + source_tx_hash: HashType([22; 32]), + amount: 5, + metadata: borsh::to_vec(&DepositMetadataForEncoding { + recipient_id: initial_public_user_accounts()[0].account_id, + }) + .unwrap(), + }; + let deposit_tx = build_bridge_deposit_tx_from_event(&record).unwrap(); + assert!( + sequencer + .store + .dbio() + .add_pending_deposit_event(record) + .unwrap() + ); + + // Skip-ahead block carrying that deposit: 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![deposit_tx]); + + apply_follow_update( + &sequencer.store.dbio(), + &sequencer.chain(), + &mempool_handle, + FollowUpdate { + adopted: vec![], + orphaned: vec![], + finalized: vec![(MsgId::from([9; 32]), parked)], + ..empty_follow_update() + }, + ); + + // Nothing became irreversible, so the store must not be swept through the + // parked block's height. + let stored = sequencer.store.get_block_at_id(2).unwrap().unwrap(); + assert!( + matches!(stored.bedrock_status, BedrockStatus::Pending), + "a parked finalized block must not mark earlier blocks finalized" + ); + // And its deposit is not minted anywhere, so dropping the record would lose + // the deposit for good once the stall clears. + assert!( + sequencer + .store + .get_pending_deposit_events() + .unwrap() + .iter() + .any(|event| event.deposit_op_id == deposit_op_id), + "a parked finalized block must not drop its deposit record" + ); +} + #[tokio::test] async fn restart_restores_head_tier_and_recovers_from_orphan() { let config = setup_sequencer_config(); @@ -1592,7 +1926,7 @@ async fn restart_restores_head_tier_and_recovers_from_orphan() { FollowUpdate { adopted: vec![(MsgId::from([21; 32]), block2_prime.clone())], orphaned: vec![(MsgId::from([20; 32]), block2)], - finalized: vec![], + ..empty_follow_update() }, ); @@ -1645,6 +1979,7 @@ async fn restart_reanchors_on_the_persisted_final_snapshot() { adopted: vec![], orphaned: vec![], finalized: vec![(MsgId::from(block2.header.hash.0), block2)], + ..empty_follow_update() }, ); } @@ -1694,7 +2029,7 @@ async fn record_produced_block_skips_persistence_on_lost_race() { MsgId::from(our_block.header.hash.0), &our_block, &[], - vec![], + &mock_checkpoint(), ) .unwrap(); @@ -1718,7 +2053,12 @@ async fn record_produced_block_skips_persistence_when_block_no_longer_chains() { // The head reorged under us: our block's parent is no longer the tip. let stale = common::test_utils::produce_dummy_block(2, Some(HashType([9; 32])), vec![]); sequencer - .record_produced_block(MsgId::from(stale.header.hash.0), &stale, &[], vec![]) + .record_produced_block( + MsgId::from(stale.header.hash.0), + &stale, + &[], + &mock_checkpoint(), + ) .unwrap(); assert!(sequencer.store.get_block_at_id(2).unwrap().is_none()); @@ -1759,6 +2099,7 @@ async fn follow_update_persists_blocks_meta_and_state_atomically() { ], orphaned: vec![], finalized: vec![(MsgId::from([2; 32]), block2)], + ..empty_follow_update() }, ); diff --git a/lez/sequencer/core/src/tests/reconstruction.rs b/lez/sequencer/core/src/tests/reconstruction.rs index e52c3364..47d84c06 100644 --- a/lez/sequencer/core/src/tests/reconstruction.rs +++ b/lez/sequencer/core/src/tests/reconstruction.rs @@ -295,7 +295,6 @@ fn deposit_event_record( recipient_id: recipient, }) .unwrap(), - submitted_in_block_id: None, } } @@ -321,11 +320,11 @@ fn build_public_withdraw_tx( LeeTransaction::Public(lee::PublicTransaction::new(message, witness_set)) } -/// The cold-start backfill re-delivers an already-finalized deposit into the -/// mempool before reconstruction applies the same deposit block, and the queued -/// mint cannot be pulled back out. Since the bridge program does not dedup on -/// `l1_deposit_op_id`, block production must skip the already-submitted deposit -/// so the vault is minted exactly once. +/// Cold-start backfill re-records an already-finalized deposit event as a +/// pending record before reconstruction replays the same deposit block. +/// Reconstruction must drop that record — its mint is permanently reflected in +/// the reconstructed state (the receipt PDA) — so the next production neither +/// re-mints the vault nor emits a stray deposit tx. #[tokio::test] async fn reconstructed_deposit_is_not_reminted_after_backfill_redelivery() { let recipient = initial_public_user_accounts()[0].account_id; @@ -343,7 +342,7 @@ async fn reconstructed_deposit_is_not_reminted_after_backfill_redelivery() { let deposit_tx = crate::build_bridge_deposit_tx_from_event(&deposit_record).expect("build deposit tx"); mempool_a - .push((TransactionOrigin::Sequencer, deposit_tx.clone())) + .push((TransactionOrigin::Sequencer, deposit_tx)) .await .unwrap(); seq_a.produce_new_block().await.unwrap(); @@ -367,10 +366,11 @@ async fn reconstructed_deposit_is_not_reminted_after_backfill_redelivery() { let channel_id = config_a.bedrock_config.channel_id; let config_b = bridge_funded_config(); - let (mut seq_b, mempool_b) = SequencerCoreWithMockClients::start_from_config(config_b).await; + let (mut seq_b, _mempool_b) = SequencerCoreWithMockClients::start_from_config(config_b).await; - // Backfill re-delivery: persist the pending record and queue the mint, as - // `on_deposit_event` does, before reconstruction runs. + // Backfill re-delivery: the deposit event is re-recorded as a pending record + // before reconstruction runs. The mint no longer flows through the mempool + // (that sink was removed); the store drain is the only source. assert!( seq_b .block_store() @@ -378,10 +378,6 @@ async fn reconstructed_deposit_is_not_reminted_after_backfill_redelivery() { .add_pending_deposit_event(deposit_record.clone()) .unwrap() ); - mempool_b - .push((TransactionOrigin::Sequencer, deposit_tx)) - .await - .unwrap(); let mock_b = MockBlockPublisher::with_canned_channel(channel_id, Some(tip_slot), messages); SequencerCore::::verify_and_reconstruct( @@ -397,6 +393,20 @@ async fn reconstructed_deposit_is_not_reminted_after_backfill_redelivery() { assert_eq!(tip_b.id, tip_a.id); assert_eq!(tip_b.hash, tip_a.hash); + // Reconstruction replays the finalized deposit block, minting the receipt + // into state and dropping the re-recorded pending event — so the drain has + // nothing left to re-mint. This is the mechanism that protects against the + // re-delivery, in place of the removed mempool sink. + assert!( + seq_b + .block_store() + .dbio() + .get_pending_deposit_events() + .unwrap() + .is_empty(), + "reconstruction must drop the re-delivered pending deposit record" + ); + seq_b.produce_new_block().await.unwrap(); let vault_id = vault_core::compute_vault_account_id(programs::vault().id(), recipient); @@ -525,7 +535,6 @@ async fn reconstruction_reconciles_already_finished_deposit() { .await .unwrap(); seq_a.produce_new_block().await.unwrap(); - let deposit_block_id = seq_a.block_store().latest_block_meta().unwrap().unwrap().id; let messages = channel_from_store(seq_a.block_store(), 10); let tip_slot = messages.last().unwrap().1; @@ -561,18 +570,19 @@ async fn reconstruction_reconciles_already_finished_deposit() { "already-finished deposit must be applied exactly once" ); - // The pending event is now marked submitted in the reconstructed block, so the - // startup replay would not re-queue it — no double mint on restart. - let record = store_b - .get_unfulfilled_deposit_events() - .unwrap() - .into_iter() - .find(|event| event.deposit_op_id == HashType(deposit_op_id)) - .expect("pending deposit event should still be recorded"); - assert_eq!( - record.submitted_in_block_id, - Some(deposit_block_id), - "reconstruction must reconcile the already-finished deposit against its channel block" + // The mint's receipt PDA is in the reconstructed state, and reconstruction + // dropped the pending record backfill had re-delivered — so the production + // drain sees the deposit as minted and never re-emits it. + assert!( + crate::deposit_already_minted( + chain_b.lock().unwrap().head_state(), + HashType(deposit_op_id) + ), + "the reconstructed deposit's receipt marks it minted" + ); + assert!( + store_b.get_pending_deposit_events().unwrap().is_empty(), + "reconstruction drops the finalized deposit's pending record" ); } diff --git a/lez/storage/src/lib.rs b/lez/storage/src/lib.rs index 8fd485d3..8ce473f2 100644 --- a/lez/storage/src/lib.rs +++ b/lez/storage/src/lib.rs @@ -67,6 +67,18 @@ pub trait DBIO { cell.put_batch(self.db(), params, write_batch) } + /// Stage a cell deletion into `write_batch`, the counterpart of + /// [`Self::put_batch`]. Deleting an absent key is a no-op (rocksdb + /// semantics). + fn del_batch( + &self, + params: T::KeyParams, + write_batch: &mut WriteBatch, + ) -> DbResult<()> { + write_batch.delete_cf(&T::column_ref(self.db()), T::key_constructor(params)?); + Ok(()) + } + /// Delete a cell. Deleting an absent key is a no-op (rocksdb semantics). fn del(&self, params: T::KeyParams) -> DbResult<()> { let cf_ref = T::column_ref(self.db()); diff --git a/lez/storage/src/sequencer/mod.rs b/lez/storage/src/sequencer/mod.rs index 0150a99f..28a9f65e 100644 --- a/lez/storage/src/sequencer/mod.rs +++ b/lez/storage/src/sequencer/mod.rs @@ -1,4 +1,4 @@ -use std::{path::Path, sync::Arc}; +use std::{collections::BTreeMap, path::Path, sync::Arc}; use borsh::{BorshDeserialize, BorshSerialize}; use common::{ @@ -13,10 +13,7 @@ use rocksdb::{ use crate::{ CF_BLOCK_NAME, CF_META_NAME, DB_META_FIRST_BLOCK_IN_DB_KEY, DBIO, DbResult, - cells::{ - SimpleStorableCell, - shared_cells::{BlockCell, FirstBlockCell, FirstBlockSetCell, LastBlockCell}, - }, + cells::shared_cells::{BlockCell, FirstBlockCell, FirstBlockSetCell, LastBlockCell}, error::DbError, sequencer::sequencer_cells::{ FinalBlockMetaCellOwned, FinalBlockMetaCellRef, FinalLeeStateCellOwned, @@ -97,6 +94,77 @@ impl DbDump { } } +/// Everything one sequencer event writes, staged into a single [`WriteBatch`] +/// by [`RocksDBIO::store_update`]. +/// +/// The point of the struct is the `checkpoint`: it is the zone-sdk's resume +/// cursor, so it must land in the *same* write as the effects it covers. +/// Persisted ahead of them, a crash in between resumes the stream past blocks +/// that never reached the store — a gap the node cannot backfill. +pub struct StoreUpdate<'update> { + /// Serialized zone-sdk checkpoint for this event. + pub checkpoint: Option<&'update [u8]>, + + /// `(block, finalized)` payloads to write. + pub blocks: &'update [(&'update Block, bool)], + + /// Head tip to pin the stored chain to; `None` only for an empty chain. + pub head_tip: Option<&'update BlockMeta>, + /// State after the last applied block. + pub head_state: &'update V03State, + + /// `(state, meta)` of the final tier, when it advanced. + pub final_snapshot: Option<(&'update V03State, &'update BlockMeta)>, + /// Highest block id this event made irreversible: stored blocks at or below + /// it become [`BedrockStatus::Finalized`]. + pub finalized_up_to: Option, + + /// Deposit events observed on L1, recorded unless already pending. + pub new_deposit_events: &'update [PendingDepositEventRecord], + /// Deposit op ids whose mint finalized: their pending records are dropped. + pub remove_deposit_records: &'update [HashType], + /// 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. + pub new_withdraw_intents: &'update [WithdrawalReconciliationKey], + + /// Advance the channel-read anchor. + pub zone_anchor: Option<&'update ZoneAnchorRecord>, +} + +impl<'update> StoreUpdate<'update> { + /// An update that writes nothing but the caller's head `state`, to be + /// filled in with `..StoreUpdate::new(state)`. + #[must_use] + pub const fn new(head_state: &'update V03State) -> Self { + Self { + checkpoint: None, + blocks: &[], + head_tip: None, + head_state, + final_snapshot: None, + finalized_up_to: None, + new_deposit_events: &[], + remove_deposit_records: &[], + consumed_withdrawals: &[], + new_withdraw_intents: &[], + zone_anchor: None, + } + } +} + +/// What [`RocksDBIO::store_update`] observed while staging, for the caller to +/// act on *after* the write committed. +#[derive(Debug, Default)] +pub struct StoreUpdateOutcome { + /// How many deposit events were newly recorded; the rest were already + /// pending, and so already owed. + pub accepted_deposits: usize, + /// Withdraw events with no matching local unseen counter, one entry per + /// unmatched occurrence. + pub unmatched_withdrawals: Vec, +} + pub struct RocksDBIO { pub db: DBWithThreadMode, } @@ -384,10 +452,6 @@ impl RocksDBIO { .map_or_else(Vec::new, |cell| cell.0)) } - fn put_pending_deposit_events(&self, records: &[PendingDepositEventRecord]) -> DbResult<()> { - self.put(&PendingDepositEventsCellRef(records), ()) - } - fn put_pending_deposit_events_batch( &self, records: &[PendingDepositEventRecord], @@ -396,135 +460,218 @@ impl RocksDBIO { self.put_batch(&PendingDepositEventsCellRef(records), (), batch) } + /// Records a single deposit event, returning whether it was new. + /// One-shot form of [`RocksDBIO::store_update`]'s `new_deposit_events`. pub fn add_pending_deposit_event(&self, event: PendingDepositEventRecord) -> DbResult { - let mut records = self.get_pending_deposit_events()?; - if records - .iter() - .any(|record| record.deposit_op_id == event.deposit_op_id) - { + let mut batch = WriteBatch::default(); + let accepted = self.stage_pending_deposit_events(&[event], &[], &mut batch)?; + // A re-delivery of an already-pending deposit — the steady state — stages + // nothing; skip the write rather than sync an empty batch. + if batch.is_empty() { return Ok(false); } - records.push(event); - self.put_pending_deposit_events(&records)?; - Ok(true) - } - - /// Marks the given deposit events submitted in `block_id`, in one write. - pub fn mark_deposit_events_submitted( - &self, - deposit_op_ids: &[HashType], - submitted_block_id: u64, - ) -> DbResult<()> { - if deposit_op_ids.is_empty() { - return Ok(()); - } - let mut batch = WriteBatch::default(); - self.mark_pending_deposit_events_submitted(deposit_op_ids, submitted_block_id, &mut batch)?; self.db.write(batch).map_err(|rerr| { DbError::rocksdb_cast_message( rerr, - Some("Failed to mark deposit events submitted".to_owned()), + Some("Failed to add pending deposit event".to_owned()), ) - }) + })?; + Ok(accepted > 0) } - fn mark_pending_deposit_events_submitted( + /// Stages every mutation of the pending-deposit records into `batch`, + /// returning how many were newly appended. + /// + /// The records live in a *single* whole-vector cell, so each mutation kind + /// cannot re-read it from disk and stage its own `put`: a later read would + /// not see the earlier staged write and would silently drop it. Everything + /// is folded in memory here instead, and written exactly once. + fn stage_pending_deposit_events( &self, - deposit_op_ids: &[HashType], - submitted_block_id: u64, + new_events: &[PendingDepositEventRecord], + remove_op_ids: &[HashType], batch: &mut WriteBatch, ) -> DbResult { - let mut records = self.get_pending_deposit_events()?; - let mut updated: usize = 0; - - for record in records - .iter_mut() - .filter(|record| deposit_op_ids.contains(&record.deposit_op_id)) - { - record.submitted_in_block_id = Some(submitted_block_id); - updated = updated.saturating_add(1); + if new_events.is_empty() && remove_op_ids.is_empty() { + return Ok(0); } - if updated > 0 { + // A set for the membership test: a backfill can finalize many deposits + // against many still-pending records at once, and a linear `contains` + // per record would be quadratic. + let to_remove: std::collections::HashSet<&HashType> = remove_op_ids.iter().collect(); + + let mut records = self.get_pending_deposit_events()?; + let before_append = records.len(); + + // `accepted` is the count of records that will actually be drained on a + // future turn, so an op id both observed and finalized in this same + // event (backfill can deliver both at once) is neither appended nor + // counted — its mint already happened, and counting it would log an + // incoming mint that never comes. It is a length delta of the appends + // alone; the retain below only touches pre-existing records. + for event in new_events { + if to_remove.contains(&event.deposit_op_id) + || records + .iter() + .any(|record| record.deposit_op_id == event.deposit_op_id) + { + continue; + } + records.push(event.clone()); + } + let accepted = records.len().saturating_sub(before_append); + + let removed = if remove_op_ids.is_empty() { + 0 + } else { + let before_retain = records.len(); + records.retain(|record| !to_remove.contains(&record.deposit_op_id)); + before_retain.saturating_sub(records.len()) + }; + + // Guard on both counts: the common finalizing event appends nothing yet + // still mutates the cell, and a pure re-delivery mutates neither and + // must not rewrite it. + if accepted > 0 || removed > 0 { self.put_pending_deposit_events_batch(&records, batch)?; } - - Ok(updated) + Ok(accepted) } - pub fn remove_fulfilled_pending_deposit_events_up_to_block( + /// Stages the unseen-withdraw decrements for one update into `batch`, + /// returning one entry per occurrence that matched no local counter. + /// + /// Occurrences are folded per key for the same reason as the deposit + /// records: two withdrawals in one update can share a reconciliation key, + /// and a per-occurrence disk read would miss the staged decrement. + fn stage_consumed_withdrawals( &self, - finalized_block_id: u64, - ) -> DbResult { - let mut records = self.get_pending_deposit_events()?; - let before = records.len(); - records.retain(|record| { - record - .submitted_in_block_id - .is_none_or(|submitted_id| submitted_id > finalized_block_id) - }); - - let removed = before.saturating_sub(records.len()); - if removed > 0 { - self.put_pending_deposit_events(&records)?; + withdrawals: &[WithdrawalReconciliationKey], + batch: &mut WriteBatch, + ) -> DbResult> { + let mut unmatched = Vec::new(); + if withdrawals.is_empty() { + return Ok(unmatched); } - Ok(removed) + // A `Vec` rather than a map: the per-update count is tiny, and it keeps + // the staging order deterministic. + let mut occurrences: Vec<(WithdrawalReconciliationKey, u64)> = Vec::new(); + for withdrawal in withdrawals { + match occurrences.iter_mut().find(|(key, _)| key == withdrawal) { + Some((_, times)) => *times = times.saturating_add(1), + None => occurrences.push((*withdrawal, 1)), + } + } + + for (withdrawal, times) in occurrences { + let stored = self + .get_opt::(withdrawal)? + .map(|cell| cell.0); + + // A stored `count` satisfies `count + 1` occurrences: the last one + // consumes the key by deleting it. Matches the one-shot + // [`Self::consume_unseen_withdraw_count`]. + let matched = times.min(stored.map_or(0, |count| count.saturating_add(1))); + unmatched.extend(std::iter::repeat_n( + withdrawal, + usize::try_from(times.saturating_sub(matched)) + .expect("unmatched withdrawal count fits usize"), + )); + + match stored.and_then(|count| count.checked_sub(times)) { + Some(count) => { + self.put_batch(&UnseenWithdrawCountCell(count), withdrawal, batch)?; + } + // Only stage a delete for a key that was actually there, so a + // fully unmatched update leaves the batch empty. + None if stored.is_some() => { + self.del_batch::(withdrawal, batch)?; + } + None => {} + } + } + + Ok(unmatched) } - /// Whether a bridge deposit for `deposit_op_id` is already recorded as - /// included in a block (its pending record is marked submitted). - pub fn is_deposit_event_submitted(&self, deposit_op_id: HashType) -> DbResult { - Ok(self.get_pending_deposit_events()?.iter().any(|record| { - record.deposit_op_id == deposit_op_id && record.submitted_in_block_id.is_some() - })) + /// Collects the [`BedrockStatus::Finalized`] flip for every stored pending + /// block at or below `last_finalized` into `to_write`. + /// + /// Reads from disk, so blocks the caller is writing itself are already in + /// `to_write` and keep their own version — one `put` per block id, no + /// reliance on the order writes are staged in. + fn collect_finalized_up_to(&self, last_finalized: u64, to_write: &mut BTreeMap) { + let newly_finalized: Vec = self + .get_all_blocks() + .filter_map(Result::ok) + .filter(|block| { + matches!(block.bedrock_status, BedrockStatus::Pending) + && block.header.block_id <= last_finalized + }) + .collect(); + + for mut block in newly_finalized { + block.bedrock_status = BedrockStatus::Finalized; + to_write.entry(block.header.block_id).or_insert(block); + } } - fn increment_unseen_withdraw_count( + /// Stages the unseen-withdraw increments for one update into `batch`. + /// + /// Occurrences are folded per key for the same reason as + /// [`Self::stage_consumed_withdrawals`]: two intents in one update can share + /// a reconciliation key, and a per-occurrence disk read would miss the + /// staged increment and count the pair once. + fn stage_new_withdraw_intents( &self, - withdrawal: WithdrawalReconciliationKey, + withdrawals: &[WithdrawalReconciliationKey], batch: &mut WriteBatch, - ) -> DbResult { - let current = self - .get_opt::(withdrawal)? - .map_or(0, |cell| cell.0); + ) -> DbResult<()> { + if withdrawals.is_empty() { + return Ok(()); + } - let next = current.checked_add(1).ok_or_else(|| { - DbError::db_interaction_error("Unseen withdraw counter overflow".to_owned()) - })?; + let mut occurrences: Vec<(WithdrawalReconciliationKey, u64)> = Vec::new(); + for withdrawal in withdrawals { + match occurrences.iter_mut().find(|(key, _)| key == withdrawal) { + Some((_, times)) => *times = times.saturating_add(1), + None => occurrences.push((*withdrawal, 1)), + } + } - self.put_batch(&UnseenWithdrawCountCell(next), withdrawal, batch)?; + for (withdrawal, times) in occurrences { + let current = self + .get_opt::(withdrawal)? + .map_or(0, |cell| cell.0); - Ok(next) + let next = current.checked_add(times).ok_or_else(|| { + DbError::db_interaction_error("Unseen withdraw counter overflow".to_owned()) + })?; + + self.put_batch(&UnseenWithdrawCountCell(next), withdrawal, batch)?; + } + + Ok(()) } + /// Reconciles a single L1 withdraw event, returning whether it matched a + /// local intent. One-shot form of [`RocksDBIO::store_update`]'s + /// `consumed_withdrawals`. pub fn consume_unseen_withdraw_count( &self, withdrawal: WithdrawalReconciliationKey, ) -> DbResult { - let Some(current) = self - .get_opt::(withdrawal)? - .map(|cell| cell.0) - else { - return Ok(false); - }; - - if let Some(next) = current.checked_sub(1) { - self.put(&UnseenWithdrawCountCell(next), withdrawal)?; - } else { - let cf_meta = self.meta_column(); - let db_key = - ::key_constructor(withdrawal)?; - - self.db.delete_cf(&cf_meta, db_key).map_err(|rerr| { - DbError::rocksdb_cast_message( - rerr, - Some("Failed to delete unseen withdraw count".to_owned()), - ) - })?; - } - - Ok(true) + let mut batch = WriteBatch::default(); + let unmatched = self.stage_consumed_withdrawals(&[withdrawal], &mut batch)?; + self.db.write(batch).map_err(|rerr| { + DbError::rocksdb_cast_message( + rerr, + Some("Failed to consume unseen withdraw count".to_owned()), + ) + })?; + Ok(unmatched.is_empty()) } pub fn put_block(&self, block: &Block, first: bool, batch: &mut WriteBatch) -> DbResult<()> { @@ -623,20 +770,23 @@ impl RocksDBIO { Ok(()) } - /// Mark every pending block with `block_id <= last_finalized` as finalized. - /// Idempotent — already-finalized blocks are skipped. + /// Mark every pending block with `block_id <= last_finalized` as finalized, + /// in one atomic write. Idempotent — already-finalized blocks are skipped. + /// One-shot form of [`RocksDBIO::store_update`]'s `finalized_up_to`. pub fn clean_pending_blocks_up_to(&self, last_finalized: u64) -> DbResult<()> { - let pending_ids: Vec = self - .get_all_blocks() - .filter_map(Result::ok) - .filter(|b| matches!(b.bedrock_status, BedrockStatus::Pending)) - .map(|b| b.header.block_id) - .filter(|id| *id <= last_finalized) - .collect(); - for id in pending_ids { - self.mark_block_as_finalized(id)?; + let mut to_write = BTreeMap::new(); + self.collect_finalized_up_to(last_finalized, &mut to_write); + + let mut batch = WriteBatch::default(); + for block in to_write.values() { + self.put_block_payload(block, &mut batch)?; } - Ok(()) + self.db.write(batch).map_err(|rerr| { + DbError::rocksdb_cast_message( + rerr, + Some("Failed to mark pending blocks finalized".to_owned()), + ) + }) } pub fn mark_block_as_finalized(&self, block_id: u64) -> DbResult<()> { @@ -691,8 +841,8 @@ impl RocksDBIO { Ok(()) } - /// One-block form of [`Self::store_followed_blocks`], with the block as the - /// head tip and no final snapshot. Production always uses the batch form. + /// One-block form of [`Self::store_update`], with the block as the head tip + /// and no final snapshot. Production always uses the batch form. #[cfg(test)] fn store_followed_block( &self, @@ -700,16 +850,17 @@ impl RocksDBIO { state: &V03State, finalized: bool, ) -> DbResult<()> { - self.store_followed_blocks( - &[(block, finalized)], - Some(&BlockMeta::from(block)), - state, - None, - ) + self.store_update(&StoreUpdate { + blocks: &[(block, finalized)], + head_tip: Some(&BlockMeta::from(block)), + ..StoreUpdate::new(state) + }) + .map(|_outcome| ()) } - /// Persists a batch of followed blocks, the caller's head-tip `state`, and - /// the optional final-tier snapshot in one atomic write. + /// Persists everything one sequencer event produced — checkpoint, blocks, + /// tip meta, head state, final snapshot, deposit and withdraw bookkeeping + /// and the channel anchor — in one atomic write. /// /// The tip meta is pinned to `head_tip`, and blocks stored above it (left /// behind by a net-shortening reorg) are deleted in the same write, so @@ -717,69 +868,115 @@ impl RocksDBIO { /// /// Per block: skips the payload write when the store already holds it (by /// id and hash), unless `finalized` is set, which rewrites it with the - /// finalized status. A no-op update (nothing to write, tip unchanged) - /// writes nothing. + /// finalized status. /// - /// TODO: the zone-sdk checkpoint is persisted by `on_checkpoint` *before* - /// this write. Full `BlocksProcessed` atomicity (checkpoint, blocks, state - /// and orphan reverts in one batch) is a follow-up. - pub fn store_followed_blocks( - &self, - blocks: &[(&Block, bool)], - head_tip: Option<&BlockMeta>, - state: &V03State, - final_snapshot: Option<(&V03State, &BlockMeta)>, - ) -> DbResult<()> { + /// The head state and tip meta are only rewritten when the chain actually + /// moved. A checkpoint alone (the common case — every follow event carries + /// one, most carry nothing else) must not drag a full state serialization + /// with it. + pub fn store_update(&self, update: &StoreUpdate<'_>) -> DbResult { + let StoreUpdate { + checkpoint, + blocks, + head_tip, + head_state, + final_snapshot, + finalized_up_to, + new_deposit_events, + remove_deposit_records, + consumed_withdrawals, + new_withdraw_intents, + zone_anchor, + } = *update; + let last_block_in_db = self.get_meta_last_block_in_db()?; let mut batch = WriteBatch::default(); + if let Some(bytes) = checkpoint { + self.put_batch(&ZoneSdkCheckpointCellRef(bytes), (), &mut batch)?; + } + if let Some(anchor) = zone_anchor { + self.put_batch(&ZoneAnchorCell(*anchor), (), &mut batch)?; + } + + // Every block payload this update writes, keyed by id so a block that + // is both explicitly written and swept by `finalized_up_to` is written + // once, with the caller's version. + let mut to_write: BTreeMap = BTreeMap::new(); + + // Whether the stored chain moved, and with it the head state. A + // shrink-only update (orphans without adopted replacements) writes no + // payloads but still rewinds the tip, or the stored state tears + // against the stale disk head on the next produce. + let mut chain_changed = + final_snapshot.is_some() || head_tip.is_some_and(|tip| tip.id != last_block_in_db); + for (block, finalized) in blocks { let already_stored = self .get_block(block.header.block_id)? .filter(|stored| stored.header.hash == block.header.hash); - let mut to_write = match already_stored { + let mut block_to_write = match already_stored { Some(_) if !finalized => continue, Some(stored) => stored, None => (*block).clone(), }; if *finalized { - to_write.bedrock_status = BedrockStatus::Finalized; + block_to_write.bedrock_status = BedrockStatus::Finalized; } - self.put_block_payload(&to_write, &mut batch)?; + to_write.insert(block_to_write.header.block_id, block_to_write); + chain_changed = true; } + if let Some(last_finalized) = finalized_up_to { + self.collect_finalized_up_to(last_finalized, &mut to_write); + } + for block in to_write.values() { + self.put_block_payload(block, &mut batch)?; + } + + let accepted_deposits = self.stage_pending_deposit_events( + new_deposit_events, + remove_deposit_records, + &mut batch, + )?; + let unmatched_withdrawals = + self.stage_consumed_withdrawals(consumed_withdrawals, &mut batch)?; + self.stage_new_withdraw_intents(new_withdraw_intents, &mut batch)?; + // `head_tip` is `None` only for a chain holding no blocks at all, which - // implies nothing was applied — and the store, created with genesis, - // cannot represent it. No tip to pin, nothing to persist. - let Some(tip) = head_tip else { - debug_assert!(batch.is_empty() && final_snapshot.is_none()); - return Ok(()); + // the store — created with genesis — cannot represent. Nothing to pin. + if chain_changed && let Some(tip) = head_tip { + // `last_block_in_db` predates this batch, so on its own it misses + // payloads staged above the pinned tip — a finalized block landing + // below an adopted one rewinds the tip under blocks this same update + // wrote. Leaving one there fails the restart replay. The deletes are + // staged after the puts, so the batch order resolves the overlap. + let highest_staged = to_write.last_key_value().map_or(0, |(id, _)| *id); + for stale_id in tip.id.saturating_add(1)..=last_block_in_db.max(highest_staged) { + self.delete_block_payload(stale_id, &mut batch)?; + } + self.put_meta_last_block_in_db_batch(tip.id, &mut batch)?; + self.put_meta_latest_block_meta_batch(tip, &mut batch)?; + self.put_lee_state_in_db_batch(head_state, &mut batch)?; + if let Some((final_state, final_meta)) = final_snapshot { + self.put_final_snapshot_batch(final_state, final_meta, &mut batch)?; + } + } + + let outcome = StoreUpdateOutcome { + accepted_deposits, + unmatched_withdrawals, }; - // A shrink-only update (orphans without adopted replacements) has no - // payloads to write but must still rewind the tip meta, or the stored - // state tears against the stale disk head on the next produce. - if batch.is_empty() && final_snapshot.is_none() && tip.id == last_block_in_db { - return Ok(()); - } - - for stale_id in tip.id.saturating_add(1)..=last_block_in_db { - self.delete_block_payload(stale_id, &mut batch)?; - } - self.put_meta_last_block_in_db_batch(tip.id, &mut batch)?; - self.put_meta_latest_block_meta_batch(tip, &mut batch)?; - self.put_lee_state_in_db_batch(state, &mut batch)?; - if let Some((final_state, final_meta)) = final_snapshot { - self.put_final_snapshot_batch(final_state, final_meta, &mut batch)?; + if batch.is_empty() { + return Ok(outcome); } self.db.write(batch).map_err(|rerr| { - DbError::rocksdb_cast_message( - rerr, - Some("Failed to write followed blocks batch".to_owned()), - ) - }) + DbError::rocksdb_cast_message(rerr, Some("Failed to write store update".to_owned())) + })?; + Ok(outcome) } pub fn get_all_blocks(&self) -> impl Iterator> { @@ -803,32 +1000,30 @@ impl RocksDBIO { }) } + /// Persists a block we produced, its withdraw intents, the resulting state + /// and the publish `checkpoint` in one atomic write. + /// + /// The produce path is [`Self::store_update`] with a single block that is + /// the new tip; the checkpoint belongs in the same write for the same + /// reason it does there — it carries the sdk's `pending_txs`, so a + /// checkpoint persisted without this block would restore a pending set + /// that no longer contains the inscription we just published, and the sdk + /// would never resubmit it. pub fn atomic_update( &self, block: &Block, - deposit_op_ids: &[HashType], - withdrawals: Vec, + withdrawals: &[WithdrawalReconciliationKey], state: &V03State, + checkpoint: Option<&[u8]>, ) -> DbResult<()> { - let block_id = block.header.block_id; - let mut batch = WriteBatch::default(); - - self.put_block(block, false, &mut batch)?; - - self.mark_pending_deposit_events_submitted(deposit_op_ids, block_id, &mut batch)?; - - for withdrawal in withdrawals { - self.increment_unseen_withdraw_count(withdrawal, &mut batch)?; - } - - self.put_lee_state_in_db_batch(state, &mut batch)?; - - self.db.write(batch).map_err(|rerr| { - DbError::rocksdb_cast_message( - rerr, - Some(format!("Failed to udpate db with block {block_id}")), - ) + self.store_update(&StoreUpdate { + checkpoint, + blocks: &[(block, false)], + head_tip: Some(&BlockMeta::from(block)), + new_withdraw_intents: withdrawals, + ..StoreUpdate::new(state) }) + .map(|_outcome| ()) } } diff --git a/lez/storage/src/sequencer/sequencer_cells.rs b/lez/storage/src/sequencer/sequencer_cells.rs index c10b3d30..8a4223c0 100644 --- a/lez/storage/src/sequencer/sequencer_cells.rs +++ b/lez/storage/src/sequencer/sequencer_cells.rs @@ -232,14 +232,17 @@ impl SimpleWritableCell for ZoneAnchorCell { } } +/// An L1 deposit event observed but not yet seen finalized. +/// +/// Purely a liveness queue: whether to actually emit a mint is decided against +/// chain state (the deposit-receipt PDA), and the record is dropped once its +/// mint finalizes. #[derive(Debug, Clone, PartialEq, Eq, BorshSerialize, BorshDeserialize)] pub struct PendingDepositEventRecord { pub deposit_op_id: HashType, pub source_tx_hash: HashType, pub amount: u64, pub metadata: Vec, - /// Set when block containing the deposit event is submitted, but not necessarily finalized. - pub submitted_in_block_id: Option, } #[derive(BorshDeserialize)] @@ -275,7 +278,7 @@ impl SimpleWritableCell for PendingDepositEventsCellRef<'_> { } } -#[derive(Debug, Clone, Copy)] +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] pub struct WithdrawalReconciliationKey { pub amount: u64, pub bedrock_account_pk: [u8; 32], diff --git a/lez/storage/src/sequencer/tests.rs b/lez/storage/src/sequencer/tests.rs index 0b6ce862..b50a5591 100644 --- a/lez/storage/src/sequencer/tests.rs +++ b/lez/storage/src/sequencer/tests.rs @@ -28,6 +28,15 @@ fn dbio_with_genesis(path: &Path) -> (RocksDBIO, Block) { (dbio, genesis) } +fn deposit_record(seed: u8) -> PendingDepositEventRecord { + PendingDepositEventRecord { + deposit_op_id: HashType([seed; 32]), + source_tx_hash: HashType([seed; 32]), + amount: u64::from(seed), + metadata: vec![seed], + } +} + fn stored_balance(dbio: &RocksDBIO) -> u128 { dbio.get_lee_state() .unwrap() @@ -106,12 +115,11 @@ fn store_followed_blocks_batch_lands_meta_and_state_on_last_block() { id: 3, hash: block3.header.hash, }; - dbio.store_followed_blocks( - &[(&block2, true), (&block3, false)], - Some(&head_tip), - &state_with_balance(300), - None, - ) + dbio.store_update(&StoreUpdate { + blocks: &[(&block2, true), (&block3, false)], + head_tip: Some(&head_tip), + ..StoreUpdate::new(&state_with_balance(300)) + }) .unwrap(); let stored2 = dbio.get_block(2).unwrap().expect("block 2 is stored"); @@ -140,12 +148,12 @@ fn final_snapshot_round_trips_and_is_absent_on_fresh_store() { id: 2, hash: block2.header.hash, }; - dbio.store_followed_blocks( - &[(&block2, true)], - Some(&final_meta), - &state_with_balance(300), - Some((&state_with_balance(200), &final_meta)), - ) + dbio.store_update(&StoreUpdate { + blocks: &[(&block2, true)], + head_tip: Some(&final_meta), + final_snapshot: Some((&state_with_balance(200), &final_meta)), + ..StoreUpdate::new(&state_with_balance(300)) + }) .unwrap(); let (final_state, meta) = dbio @@ -204,12 +212,11 @@ fn net_shortening_reorg_drops_stale_blocks() { id: 2, hash: block2b.header.hash, }; - dbio.store_followed_blocks( - &[(&block2b, false)], - Some(&head_tip), - &state_with_balance(400), - None, - ) + dbio.store_update(&StoreUpdate { + blocks: &[(&block2b, false)], + head_tip: Some(&head_tip), + ..StoreUpdate::new(&state_with_balance(400)) + }) .unwrap(); let stored2 = dbio.get_block(2).unwrap().expect("block 2 is stored"); @@ -241,8 +248,11 @@ fn shrink_only_reorg_rewinds_tip_meta() { id: 2, hash: block2.header.hash, }; - dbio.store_followed_blocks(&[], Some(&head_tip), &state_with_balance(200), None) - .unwrap(); + dbio.store_update(&StoreUpdate { + head_tip: Some(&head_tip), + ..StoreUpdate::new(&state_with_balance(200)) + }) + .unwrap(); assert!( dbio.get_block(3).unwrap().is_none(), @@ -254,6 +264,232 @@ fn shrink_only_reorg_rewinds_tip_meta() { assert_eq!(stored_balance(&dbio), 200); } +#[test] +fn checkpoint_lands_with_an_orphan_only_update() { + let temp_dir = tempdir().unwrap(); + let (dbio, genesis) = dbio_with_genesis(temp_dir.path()); + + let block2 = produce_dummy_block(2, Some(genesis.header.hash), vec![]); + dbio.store_followed_block(&block2, &state_with_balance(200), false) + .unwrap(); + let block3 = produce_dummy_block(3, Some(block2.header.hash), vec![]); + dbio.store_followed_block(&block3, &state_with_balance(300), false) + .unwrap(); + + // Orphan-only update: no payload to write, but the checkpoint covering it + // must still land, or a restart resumes past the orphan. + let head_tip = BlockMeta { + id: 2, + hash: block2.header.hash, + }; + dbio.store_update(&StoreUpdate { + checkpoint: Some(b"cp-orphan"), + head_tip: Some(&head_tip), + ..StoreUpdate::new(&state_with_balance(200)) + }) + .unwrap(); + + assert_eq!( + dbio.get_zone_sdk_checkpoint_bytes().unwrap().as_deref(), + Some(b"cp-orphan".as_slice()) + ); + assert!(dbio.get_block(3).unwrap().is_none()); +} + +#[test] +fn checkpoint_only_update_does_not_rewrite_the_head_state() { + let temp_dir = tempdir().unwrap(); + let (dbio, genesis) = dbio_with_genesis(temp_dir.path()); + + let block2 = produce_dummy_block(2, Some(genesis.header.hash), vec![]); + dbio.store_followed_block(&block2, &state_with_balance(200), false) + .unwrap(); + + // An event carrying nothing but a checkpoint (the common case) must not + // drag a full state serialization along with it — the caller's state is + // ignored while the chain stands still. + let head_tip = BlockMeta::from(&block2); + dbio.store_update(&StoreUpdate { + checkpoint: Some(b"cp-idle"), + head_tip: Some(&head_tip), + ..StoreUpdate::new(&state_with_balance(999)) + }) + .unwrap(); + + assert_eq!( + dbio.get_zone_sdk_checkpoint_bytes().unwrap().as_deref(), + Some(b"cp-idle".as_slice()) + ); + assert_eq!(stored_balance(&dbio), 200); +} + +#[test] +fn several_deposits_in_one_update_are_all_recorded() { + let temp_dir = tempdir().unwrap(); + let (dbio, _genesis) = dbio_with_genesis(temp_dir.path()); + + // The records live in one whole-vector cell: staged per event against a + // fresh disk read, the second would clobber the first. + let first = deposit_record(1); + let second = deposit_record(2); + let already_known = dbio.get_pending_deposit_events().unwrap(); + assert!(already_known.is_empty()); + + let outcome = dbio + .store_update(&StoreUpdate { + new_deposit_events: &[first.clone(), second.clone()], + ..StoreUpdate::new(&state_with_balance(100)) + }) + .unwrap(); + + assert_eq!(outcome.accepted_deposits, 2); + let stored = dbio.get_pending_deposit_events().unwrap(); + assert_eq!(stored.len(), 2); + assert!(stored.contains(&first)); + assert!(stored.contains(&second)); +} + +#[test] +fn redelivered_deposit_is_not_accepted_twice() { + let temp_dir = tempdir().unwrap(); + let (dbio, _genesis) = dbio_with_genesis(temp_dir.path()); + + let record = deposit_record(1); + dbio.store_update(&StoreUpdate { + new_deposit_events: std::slice::from_ref(&record), + ..StoreUpdate::new(&state_with_balance(100)) + }) + .unwrap(); + + let outcome = dbio + .store_update(&StoreUpdate { + new_deposit_events: &[record], + ..StoreUpdate::new(&state_with_balance(100)) + }) + .unwrap(); + + assert_eq!( + outcome.accepted_deposits, 0, + "a re-delivered deposit is already owed, not newly accepted" + ); + assert_eq!(dbio.get_pending_deposit_events().unwrap().len(), 1); +} + +#[test] +fn finalized_deposit_records_are_removed_by_op_id() { + let temp_dir = tempdir().unwrap(); + let (dbio, _genesis) = dbio_with_genesis(temp_dir.path()); + + let first = deposit_record(1); + let second = deposit_record(2); + dbio.store_update(&StoreUpdate { + new_deposit_events: &[first.clone(), second.clone()], + ..StoreUpdate::new(&state_with_balance(100)) + }) + .unwrap(); + + // Only the finalized op id is dropped; the other record stays. + dbio.store_update(&StoreUpdate { + remove_deposit_records: &[first.deposit_op_id], + ..StoreUpdate::new(&state_with_balance(100)) + }) + .unwrap(); + + let stored = dbio.get_pending_deposit_events().unwrap(); + assert_eq!(stored, vec![second]); +} + +#[test] +fn repeated_withdrawal_key_in_one_update_folds_once_per_occurrence() { + let temp_dir = tempdir().unwrap(); + let (dbio, _genesis) = dbio_with_genesis(temp_dir.path()); + + let key = WithdrawalReconciliationKey { + amount: 7, + bedrock_account_pk: [3; 32], + }; + + // Two local intents for the same key in one update — two withdrawals of the + // same amount to the same L1 key. A per-occurrence disk read would miss the + // staged increment and record the pair as one. + dbio.store_update(&StoreUpdate { + new_withdraw_intents: &[key, key], + ..StoreUpdate::new(&state_with_balance(100)) + }) + .unwrap(); + let recorded = dbio + .get_opt::(key) + .unwrap() + .map(|cell| cell.0); + assert_eq!(recorded, Some(2)); + + // Both L1 events arrive in one update; a per-occurrence disk read would + // miss the staged decrement and consume only one. + let outcome = dbio + .store_update(&StoreUpdate { + consumed_withdrawals: &[key, key], + ..StoreUpdate::new(&state_with_balance(100)) + }) + .unwrap(); + + assert!(outcome.unmatched_withdrawals.is_empty()); + // Both decrements landed; a per-occurrence disk read would leave `Some(1)`. + // (The absolute value trails the intent count by one — `consume` still + // treats a stored 0 as consumable — but that predates the batching and is + // replicated as-is.) + let remaining = dbio + .get_opt::(key) + .unwrap() + .map(|cell| cell.0); + assert_eq!(remaining, Some(0)); +} + +#[test] +fn unmatched_withdrawal_is_reported_and_writes_nothing() { + let temp_dir = tempdir().unwrap(); + let (dbio, _genesis) = dbio_with_genesis(temp_dir.path()); + + let key = WithdrawalReconciliationKey { + amount: 5, + bedrock_account_pk: [4; 32], + }; + let outcome = dbio + .store_update(&StoreUpdate { + consumed_withdrawals: &[key], + ..StoreUpdate::new(&state_with_balance(100)) + }) + .unwrap(); + + assert_eq!(outcome.unmatched_withdrawals.len(), 1); + assert!( + dbio.get_opt::(key) + .unwrap() + .is_none(), + "an unmatched withdraw must not leave a counter behind" + ); +} + +#[test] +fn produced_block_persists_its_publish_checkpoint() { + let temp_dir = tempdir().unwrap(); + let (dbio, genesis) = dbio_with_genesis(temp_dir.path()); + + let block2 = produce_dummy_block(2, Some(genesis.header.hash), vec![]); + dbio.atomic_update(&block2, &[], &state_with_balance(200), Some(b"cp-produced")) + .unwrap(); + + // Storing the block without the checkpoint would let a restart restore a + // pending set that no longer holds the inscription we just published. + assert_eq!( + dbio.get_zone_sdk_checkpoint_bytes().unwrap().as_deref(), + Some(b"cp-produced".as_slice()) + ); + assert_eq!( + dbio.get_block(2).unwrap().unwrap().header.hash, + block2.header.hash + ); +} + #[test] fn produced_block_below_disk_head_pins_meta_and_prunes() { let temp_dir = tempdir().unwrap(); @@ -270,7 +506,7 @@ fn produced_block_below_disk_head_pins_meta_and_prunes() { // pins the tip meta to the produced block and drops the stale suffix in // the same write, mirroring the follow path. let block2b = produce_dummy_block(2, Some(genesis.header.hash), vec![]); - dbio.atomic_update(&block2b, &[], vec![], &state_with_balance(400)) + dbio.atomic_update(&block2b, &[], &state_with_balance(400), None) .unwrap(); let stored2 = dbio.get_block(2).unwrap().expect("block 2 is stored"); diff --git a/test_fixtures/fixtures/prebuilt_sequencer_db.dump b/test_fixtures/fixtures/prebuilt_sequencer_db.dump index 1301fa0b..903d660d 100644 Binary files a/test_fixtures/fixtures/prebuilt_sequencer_db.dump and b/test_fixtures/fixtures/prebuilt_sequencer_db.dump differ