diff --git a/lez/chain_state/src/apply.rs b/lez/chain_state/src/apply.rs index 5e2ff6f1..98c91a19 100644 --- a/lez/chain_state/src/apply.rs +++ b/lez/chain_state/src/apply.rs @@ -52,7 +52,7 @@ pub fn apply_block( /// Checks that `block` is the valid continuation of `tip`: hash integrity, /// then block-id continuity, then `prev_block_hash` linkage. A `None` tip /// (cold state) expects the genesis block. -fn validate_against_tip(tip: Option<&Tip>, block: &Block) -> Result<(), BlockIngestError> { +pub fn validate_against_tip(tip: Option<&Tip>, block: &Block) -> Result<(), BlockIngestError> { let computed = block.recompute_hash(); if computed != block.header.hash { return Err(BlockIngestError::HashMismatch { @@ -95,7 +95,7 @@ fn validate_against_tip(tip: Option<&Tip>, block: &Block) -> Result<(), BlockIng /// Applies a block's transactions to `state`, mapping every failure to a /// [`BlockIngestError`] so the caller can park rather than crash. Operates in /// place; the caller commits only on `Ok`. -fn apply_block_to_state(block: &Block, state: &mut V03State) -> Result<(), BlockIngestError> { +pub fn apply_block_to_state(block: &Block, state: &mut V03State) -> Result<(), BlockIngestError> { let (clock_tx, user_txs) = block .body .transactions diff --git a/lez/chain_state/src/chain.rs b/lez/chain_state/src/chain.rs index 07c9890b..c3a810c0 100644 --- a/lez/chain_state/src/chain.rs +++ b/lez/chain_state/src/chain.rs @@ -160,8 +160,8 @@ impl ChainState { block: &Block, l1_slot: Slot, ) -> AcceptOutcome { - // Match by MsgId, or by block identity to handle a re-inscribed but - // identical block arriving under a fresh MsgId. + // Match by [`MsgId`] or by block identity to handle a re-inscribed but + // identical block arriving under a fresh [`MsgId`]. let in_head = self.head_blocks.iter().position(|entry| { entry.this_msg == this_msg || entry.block.header.hash == block.header.hash }); @@ -179,7 +179,7 @@ impl ChainState { let finalized: Vec = self.head_blocks.drain(0..=idx).collect(); for entry in finalized { apply_block(self.final_tip.as_ref(), &entry.block, &mut self.final_state) - .expect("a validated head block must apply to the final tier"); + .expect("validated head block must apply to the final tier"); self.final_tip = Some(tip_of(&entry.block)); } self.final_stall = None; @@ -188,6 +188,19 @@ impl ChainState { /// Applies a finalized block straight to the final tier. On success the /// finalized chain is authoritative, so head rebases onto it. fn apply_finalized_direct(&mut self, block: &Block, l1_slot: Slot) -> AcceptOutcome { + // A finalized block at or below the final tip is a re-delivery — e.g. + // the finalization of blocks restored from the store at boot, which + // are the final tier from the start and so never sit in `head_blocks`. + // Idempotent, no stall. A *different* block at the tip height falls + // through to validation and parks: finalized is irreversible, so a + // conflicting finalized block at final height is a genuine stall. + if let Some(tip) = &self.final_tip + && (block.header.block_id < tip.block_id + || (block.header.block_id == tip.block_id && block.header.hash == tip.hash)) + { + return AcceptOutcome::AlreadyApplied; + } + let mut scratch = self.final_state.clone(); match apply_block(self.final_tip.as_ref(), block, &mut scratch) { Ok(()) => { @@ -624,6 +637,50 @@ mod tests { assert_head_matches_replay(&chain); } + #[test] + fn finalized_redelivery_at_or_below_final_tip_is_already_applied() { + // Restart shape: the store's tip (incl. not-yet-finalized blocks) is + // restored as the final tier, so their later finalization arrives for + // blocks that were never in `head_blocks`. + let mut chain = ChainState::new(initial_state()); + let genesis = produce_dummy_block(1, None, vec![]); + let block2 = produce_dummy_block(2, Some(genesis.header.hash), vec![]); + chain.apply_finalized(msg(1), &genesis, slot(10)); + chain.apply_finalized(msg(2), &block2, slot(20)); + + // Below the tip, and at the tip with a matching hash: idempotent. + assert!(matches!( + chain.apply_finalized(msg(41), &genesis, slot(30)), + AcceptOutcome::AlreadyApplied + )); + assert!(matches!( + chain.apply_finalized(msg(42), &block2, slot(30)), + AcceptOutcome::AlreadyApplied + )); + assert!(chain.final_stall().is_none()); + assert_eq!(chain.final_tip().expect("final tip").block_id, 2); + assert_head_matches_replay(&chain); + } + + #[test] + fn conflicting_finalized_at_final_tip_parks() { + let mut chain = ChainState::new(initial_state()); + let genesis = produce_dummy_block(1, None, vec![]); + let block2 = produce_dummy_block(2, Some(genesis.header.hash), vec![]); + chain.apply_finalized(msg(1), &genesis, slot(10)); + chain.apply_finalized(msg(2), &block2, slot(20)); + + // A different finalized block at the final height: finalized is + // irreversible, so this is a genuine stall, not a re-delivery. + let block2_prime = produce_dummy_block(2, Some(HashType([9; 32])), vec![]); + assert!(matches!( + chain.apply_finalized(msg(22), &block2_prime, slot(30)), + AcceptOutcome::Parked(_) + )); + assert!(chain.final_stall().is_some()); + assert_eq!(chain.final_tip().expect("final tip").block_id, 2); + } + #[test] fn finalized_unknown_block_rebases_head() { let accounts = initial_pub_accounts_private_keys(); diff --git a/lez/chain_state/src/lib.rs b/lez/chain_state/src/lib.rs index c4725ec3..6398ce77 100644 --- a/lez/chain_state/src/lib.rs +++ b/lez/chain_state/src/lib.rs @@ -2,7 +2,7 @@ //! the [`apply_block`] entry point plus [`BlockIngestError`], [`StallReason`], //! [`Tip`], and [`AcceptOutcome`]. See `DESIGN.md` for the two-tier model. -pub use apply::{AcceptOutcome, Tip, apply_block}; +pub use apply::{AcceptOutcome, Tip, apply_block, apply_block_to_state, validate_against_tip}; pub use chain::{ChainState, HeadEntry}; pub use ingest_error::BlockIngestError; pub use stall_reason::StallReason; diff --git a/lez/indexer/core/src/block_store.rs b/lez/indexer/core/src/block_store.rs index 7a9f410d..0a6efe28 100644 --- a/lez/indexer/core/src/block_store.rs +++ b/lez/indexer/core/src/block_store.rs @@ -1,7 +1,9 @@ use std::{path::Path, sync::Arc}; use anyhow::{Context as _, Result}; -use chain_state::{AcceptOutcome, BlockIngestError, StallReason, Tip, apply_block}; +use chain_state::{ + AcceptOutcome, BlockIngestError, StallReason, Tip, apply_block_to_state, validate_against_tip, +}; use common::{ block::{BedrockStatus, Block, BlockHeader}, transaction::LeeTransaction, @@ -236,9 +238,17 @@ impl IndexerStore { return Ok(AcceptOutcome::AlreadyApplied); } + // Fail fast on validation before paying for the scratch clone — while + // parked, every re-delivered non-chaining block takes this path. + // Validation failures are never retryable, so parking here is exact. + if let Err(err) = validate_against_tip(tip.as_ref(), block) { + self.record_stall(Some(&block.header), l1_slot, err.clone())?; + return Ok(AcceptOutcome::Parked(err)); + } + // TODO: we use scratch state to be atomic, but need to revisit how expensive a clone is let mut scratch = self.current_state.read().await.clone(); - if let Err(err) = apply_block(tip.as_ref(), block, &mut scratch) { + if let Err(err) = apply_block_to_state(block, &mut scratch) { if err.is_retryable() { return Ok(AcceptOutcome::RetryableFailure(err)); } diff --git a/lez/mempool/src/lib.rs b/lez/mempool/src/lib.rs index 1b36eaf7..0006f2c3 100644 --- a/lez/mempool/src/lib.rs +++ b/lez/mempool/src/lib.rs @@ -65,6 +65,11 @@ impl MemPoolHandle { pub async fn push(&self, item: T) -> Result<(), tokio::sync::mpsc::error::SendError> { self.sender.send(item).await } + + /// Send an item to the mempool, failing _immediately_ if it is full. + pub fn try_push(&self, item: T) -> Result<(), tokio::sync::mpsc::error::TrySendError> { + self.sender.try_send(item) + } } #[cfg(test)] @@ -123,6 +128,19 @@ mod tests { assert_eq!(pool.pop(), Some(2)); } + #[test] + async fn try_push_fails_when_full_without_blocking() { + let (mut pool, handle) = MemPool::new(1); + + handle.try_push(1).unwrap(); + assert!(handle.try_push(2).is_err(), "full mempool must not accept"); + + // Popping frees capacity again. + assert_eq!(pool.pop(), Some(1)); + handle.try_push(2).unwrap(); + assert_eq!(pool.pop(), Some(2)); + } + #[test] async fn push_front() { let (mut pool, handle) = MemPool::new(10); diff --git a/lez/sequencer/core/src/lib.rs b/lez/sequencer/core/src/lib.rs index 3ebc8abf..1fb52b1a 100644 --- a/lez/sequencer/core/src/lib.rs +++ b/lez/sequencer/core/src/lib.rs @@ -363,12 +363,8 @@ impl SequencerCore { mempool_handle: MemPoolHandle<(TransactionOrigin, LeeTransaction)>, ) -> block_publisher::OnFollowSink { Box::new(move |update: block_publisher::FollowUpdate| { - Box::pin(apply_follow_update( - Arc::clone(&dbio), - Arc::clone(&chain), - mempool_handle.clone(), - update, - )) + apply_follow_update(&dbio, &chain, &mempool_handle, update); + Box::pin(std::future::ready(())) }) } @@ -714,24 +710,30 @@ struct BlockWithMeta { /// relies on a valid successor or a restart. `ChainState` never emits /// `AcceptOutcome::RetryableFailure` yet; adding retry parity here is a /// follow-up. -async fn apply_follow_update( - dbio: Arc, - chain: Arc>, - mempool_handle: MemPoolHandle<(TransactionOrigin, LeeTransaction)>, +fn apply_follow_update( + dbio: &RocksDBIO, + chain: &Mutex, + mempool_handle: &MemPoolHandle<(TransactionOrigin, LeeTransaction)>, update: block_publisher::FollowUpdate, ) { + let block_publisher::FollowUpdate { + adopted, + orphaned, + finalized, + } = update; + // Apply under the lock and collect what to persist; take a single // head snapshot. Release the lock before touching disk so the // producer is never blocked on the follow path's I/O. let (to_persist, resubmit_txs, head_snapshot) = { let mut chain = chain.lock().expect("chain state mutex poisoned"); let mut resubmit_txs = Vec::new(); - for (this_msg, block) in &update.orphaned { + for (this_msg, block) in &orphaned { chain.revert_orphan(*this_msg); resubmit_txs.extend(resubmittable_txs(block)); } let mut to_persist = Vec::new(); - for (this_msg, block) in &update.adopted { + for (this_msg, block) in &adopted { if matches!( chain.apply_adopted(*this_msg, block), AcceptOutcome::Applied @@ -739,8 +741,8 @@ async fn apply_follow_update( to_persist.push((block, false)); } } - for (this_msg, block) in &update.finalized { - // TODO: thread the finalized inscription's L1 slot once the + 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)), @@ -767,9 +769,14 @@ async fn apply_follow_update( // Rebuild orphaned work: return its user txs to the mempool so the // next on-turn production re-includes them on the new head. + // + // We use [`try_push`] here because this is called from the + // publisher's drive task, and only the block production drains the mempool. + // A blocking push on a full mempool would deadlock here. for tx in resubmit_txs { - if let Err(err) = mempool_handle.push((TransactionOrigin::User, tx)).await { - error!("Failed to resubmit orphaned transaction: {err:#}"); + let tx_hash = tx.hash(); + if let Err(err) = mempool_handle.try_push((TransactionOrigin::User, tx)) { + warn!("Dropping orphaned transaction {tx_hash} on resubmit: {err}"); } } } diff --git a/lez/sequencer/core/src/tests.rs b/lez/sequencer/core/src/tests.rs index 965136ba..46af7019 100644 --- a/lez/sequencer/core/src/tests.rs +++ b/lez/sequencer/core/src/tests.rs @@ -1351,16 +1351,15 @@ async fn follow_adopted_peer_block_applies_and_persists() { let peer_block = common::test_utils::produce_dummy_block(2, Some(genesis_meta.hash), vec![tx]); apply_follow_update( - sequencer.store.dbio(), - sequencer.chain(), - mempool_handle.clone(), + &sequencer.store.dbio(), + &sequencer.chain(), + &mempool_handle, FollowUpdate { adopted: vec![(MsgId::from([1; 32]), peer_block.clone())], orphaned: vec![], finalized: vec![], }, - ) - .await; + ); assert_eq!(sequencer.chain_height(), 2); let stored = sequencer @@ -1400,16 +1399,15 @@ async fn follow_redelivery_of_own_block_is_deduped() { // The channel redelivers our own block under the MsgId the mock publisher // assigned at publish time. apply_follow_update( - sequencer.store.dbio(), - sequencer.chain(), - mempool_handle.clone(), + &sequencer.store.dbio(), + &sequencer.chain(), + &mempool_handle, FollowUpdate { - adopted: vec![(MsgId::from(block2.header.hash.0), block2.clone())], + adopted: vec![(MsgId::from(block2.header.hash.0), block2)], orphaned: vec![], finalized: vec![], }, - ) - .await; + ); assert_eq!(sequencer.chain_height(), 2); assert_eq!( @@ -1442,16 +1440,15 @@ async fn follow_orphan_reverts_head_and_requeues_user_txs() { let block2 = sequencer.store.get_block_at_id(2).unwrap().unwrap(); apply_follow_update( - sequencer.store.dbio(), - sequencer.chain(), - mempool_handle.clone(), + &sequencer.store.dbio(), + &sequencer.chain(), + &mempool_handle, FollowUpdate { adopted: vec![], - orphaned: vec![(MsgId::from(block2.header.hash.0), block2.clone())], + orphaned: vec![(MsgId::from(block2.header.hash.0), block2)], finalized: vec![], }, - ) - .await; + ); assert_eq!(sequencer.chain_height(), 1); assert_eq!( @@ -1486,16 +1483,15 @@ async fn follow_finalized_own_block_moves_final_tier_and_marks_store() { let block2 = sequencer.store.get_block_at_id(2).unwrap().unwrap(); apply_follow_update( - sequencer.store.dbio(), - sequencer.chain(), - mempool_handle.clone(), + &sequencer.store.dbio(), + &sequencer.chain(), + &mempool_handle, FollowUpdate { adopted: vec![], orphaned: vec![], - finalized: vec![(MsgId::from(block2.header.hash.0), block2.clone())], + finalized: vec![(MsgId::from(block2.header.hash.0), block2)], }, - ) - .await; + ); let final_tip = sequencer .chain() @@ -1520,16 +1516,15 @@ async fn follow_finalized_backfill_block_is_applied_and_marked_finalized() { 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.clone(), + &sequencer.store.dbio(), + &sequencer.chain(), + &mempool_handle, FollowUpdate { adopted: vec![], orphaned: vec![], finalized: vec![(MsgId::from([2; 32]), peer_block.clone())], }, - ) - .await; + ); assert_eq!( sequencer.chain_height(), @@ -1642,19 +1637,18 @@ async fn follow_update_persists_blocks_meta_and_state_atomically() { // One update carrying several blocks: both adopted, block 2 also finalized. apply_follow_update( - sequencer.store.dbio(), - sequencer.chain(), - mempool_handle.clone(), + &sequencer.store.dbio(), + &sequencer.chain(), + &mempool_handle, FollowUpdate { adopted: vec![ (MsgId::from([2; 32]), block2.clone()), (MsgId::from([3; 32]), block3.clone()), ], orphaned: vec![], - finalized: vec![(MsgId::from([2; 32]), block2.clone())], + finalized: vec![(MsgId::from([2; 32]), block2)], }, - ) - .await; + ); // Blocks, tip meta and state all reflect the end of the batch: a late // finalized entry for an earlier block must not drag the tip meta back.