fix(sequencer): harden follow-path finalization and deposit/withdraw batching

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<HashType> = finalized
+        let finalized_deposit_ids: Vec<HashType> = irreversible
             .iter()
-            .flat_map(|(_, block)| block.body.transactions.iter())
+            .flat_map(|block| block.body.transactions.iter())
             .filter_map(extract_bridge_deposit_id)
             .collect();
 

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();

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::<MockBlockPublisher>::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);

lez/storage/src/sequencer/mod.rs

@@ -465,6 +465,11 @@ impl RocksDBIO {
     pub fn add_pending_deposit_event(&self, event: PendingDepositEventRecord) -> DbResult<bool> {
         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<u64> {
-        let current = self
-            .get_opt::<UnseenWithdrawCountCell>(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::<UnseenWithdrawCountCell>(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)?;

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::<UnseenWithdrawCountCell>(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::<UnseenWithdrawCountCell>(key)
         .unwrap()