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

fix(chain_state, sequencer, indexer): finalized re-delivery dedup, non-blocking orphan resubmit, fail-fast validation

Browse Source
This commit is contained in:
erhant 2026-07-16 14:10:54 +03:00
parent c9e3bf2a05
commit f8bc1581ce
7 changed files with 144 additions and 58 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
lez/chain_state/src/apply.rs
View File

@ -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

63
lez/chain_state/src/chain.rs
View File

@ -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<HeadEntry> = 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();

2
lez/chain_state/src/lib.rs
View File

@ -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;

14
lez/indexer/core/src/block_store.rs
View File

@ -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));
}

18
lez/mempool/src/lib.rs
View File

@ -65,6 +65,11 @@ impl<T> MemPoolHandle<T> {
pub async fn push(&self, item: T) -> Result<(), tokio::sync::mpsc::error::SendError<T>> {
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<T>> {
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);

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

@ -363,12 +363,8 @@ impl<BP: BlockPublisherTrait> SequencerCore<BP> {
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<RocksDBIO>,
chain: Arc<Mutex<ChainState>>,
mempool_handle: MemPoolHandle<(TransactionOrigin, LeeTransaction)>,
fn apply_follow_update(
dbio: &RocksDBIO,
chain: &Mutex<ChainState>,
mempool_handle: &MemPoolHandle<(TransactionOrigin, LeeTransaction)>,
update: block_publisher::FollowUpdate,
) {
let block_publisher::FollowUpdate {
adopted,
orphaned,
finalized,
} = update;
// 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}");
}
}
}

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

@ -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.