From 849ea66c0f45d61619316fe6b45a15b4a4faa369 Mon Sep 17 00:00:00 2001 From: erhant Date: Mon, 27 Jul 2026 11:43:51 +0300 Subject: [PATCH] fix(sequencer): harden follow-path finalization and deposit/withdraw batching --- lez/sequencer/core/src/lib.rs | 30 ++++---- lez/sequencer/core/src/tests.rs | 70 +++++++++++++++++++ .../core/src/tests/reconstruction.rs | 37 ++++++---- lez/storage/src/sequencer/mod.rs | 65 ++++++++++++----- lez/storage/src/sequencer/tests.rs | 29 +++++--- 5 files changed, 179 insertions(+), 52 deletions(-) diff --git a/lez/sequencer/core/src/lib.rs b/lez/sequencer/core/src/lib.rs index 29c5cf7f..c1e3a760 100644 --- a/lez/sequencer/core/src/lib.rs +++ b/lez/sequencer/core/src/lib.rs @@ -917,16 +917,25 @@ 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; + 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. @@ -938,19 +947,16 @@ fn apply_follow_update( // Every block at or below the highest finalized one is irreversible, so // stored blocks there can be marked finalized. - let last_finalized = finalized - .iter() - .map(|(_, block)| block.header.block_id) - .max(); + let last_finalized = irreversible.iter().map(|block| block.header.block_id).max(); // 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 = finalized + let finalized_deposit_ids: Vec = irreversible .iter() - .flat_map(|(_, block)| block.body.transactions.iter()) + .flat_map(|block| block.body.transactions.iter()) .filter_map(extract_bridge_deposit_id) .collect(); diff --git a/lez/sequencer/core/src/tests.rs b/lez/sequencer/core/src/tests.rs index 46d3a4af..6b8a08bc 100644 --- a/lez/sequencer/core/src/tests.rs +++ b/lez/sequencer/core/src/tests.rs @@ -1683,6 +1683,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(); diff --git a/lez/sequencer/core/src/tests/reconstruction.rs b/lez/sequencer/core/src/tests/reconstruction.rs index 7a57e496..47d84c06 100644 --- a/lez/sequencer/core/src/tests/reconstruction.rs +++ b/lez/sequencer/core/src/tests/reconstruction.rs @@ -320,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; @@ -342,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(); @@ -366,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() @@ -377,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( @@ -396,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); diff --git a/lez/storage/src/sequencer/mod.rs b/lez/storage/src/sequencer/mod.rs index e6bb6679..1a398c75 100644 --- a/lez/storage/src/sequencer/mod.rs +++ b/lez/storage/src/sequencer/mod.rs @@ -465,6 +465,11 @@ impl RocksDBIO { pub fn add_pending_deposit_event(&self, event: PendingDepositEventRecord) -> DbResult { 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); + } self.db.write(batch).map_err(|rerr| { DbError::rocksdb_cast_message( rerr, @@ -508,8 +513,12 @@ impl RocksDBIO { let removed = if remove_op_ids.is_empty() { 0 } else { + // 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 after_append = records.len(); - records.retain(|record| !remove_op_ids.contains(&record.deposit_op_id)); + records.retain(|record| !to_remove.contains(&record.deposit_op_id)); after_append.saturating_sub(records.len()) }; @@ -601,22 +610,42 @@ impl RocksDBIO { } } - 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 @@ -905,14 +934,18 @@ impl RocksDBIO { )?; let unmatched_withdrawals = self.stage_consumed_withdrawals(consumed_withdrawals, &mut batch)?; - for withdrawal in new_withdraw_intents { - self.increment_unseen_withdraw_count(*withdrawal, &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 // the store — created with genesis — cannot represent. Nothing to pin. if chain_changed && let Some(tip) = head_tip { - for stale_id in tip.id.saturating_add(1)..=last_block_in_db { + // `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)?; diff --git a/lez/storage/src/sequencer/tests.rs b/lez/storage/src/sequencer/tests.rs index b3e8c070..b50a5591 100644 --- a/lez/storage/src/sequencer/tests.rs +++ b/lez/storage/src/sequencer/tests.rs @@ -400,7 +400,7 @@ fn finalized_deposit_records_are_removed_by_op_id() { } #[test] -fn repeated_withdrawal_key_in_one_update_decrements_once_per_occurrence() { +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()); @@ -408,13 +408,20 @@ fn repeated_withdrawal_key_in_one_update_decrements_once_per_occurrence() { amount: 7, bedrock_account_pk: [3; 32], }; - // Two local intents for the same key. - for _ in 0..2 { - let mut batch = WriteBatch::default(); - dbio.increment_unseen_withdraw_count(key, &mut batch) - .unwrap(); - dbio.db.write(batch).unwrap(); - } + + // 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. @@ -427,9 +434,9 @@ fn repeated_withdrawal_key_in_one_update_decrements_once_per_occurrence() { 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 — `increment` stores 1 - // for the first intent while `consume` still treats a stored 0 as - // consumable — but that predates the batching and is replicated as-is.) + // (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()