fix(sequencer): tend to PR comments & self-review findings

lez/sequencer/core/src/lib.rs

@@ -584,7 +584,7 @@ impl<BP: BlockPublisherTrait> SequencerCore<BP> {
         block_height: u64,
         timestamp: u64,
         withdrawals: &mut Vec<WithdrawArg>,
-    ) -> Result<bool> {
+    ) -> bool {
         let tx_hash = tx.hash();
         match origin {
             TransactionOrigin::User => {
@@ -594,7 +594,7 @@ impl<BP: BlockPublisherTrait> SequencerCore<BP> {
                         error!(
                             "Transaction with hash {tx_hash} failed execution check with error: {err:#?}, skipping it",
                         );
-                        return Ok(false);
+                        return false;
                     }
                 };
 
@@ -612,14 +612,27 @@ impl<BP: BlockPublisherTrait> SequencerCore<BP> {
                 // 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.
-                state
-                    .transition_from_public_transaction(public_tx, block_height, timestamp)
-                    .context("Failed to execute sequencer-generated transaction")?;
+                //
+                // 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;
+                }
             }
         }
 
         info!("Validated transaction with hash {tx_hash}, including it in block");
-        Ok(true)
+        true
     }
 
     fn build_block_from_mempool(&mut self) -> Result<BlockWithMeta> {
@@ -725,7 +738,7 @@ impl<BP: BlockPublisherTrait> SequencerCore<BP> {
                 new_block_height,
                 new_block_timestamp,
                 &mut withdrawals,
-            )? {
+            ) {
                 valid_transactions.push(tx);
             }
 
@@ -787,10 +800,9 @@ impl<BP: BlockPublisherTrait> SequencerCore<BP> {
     }
 
     /// 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<()> {

lez/sequencer/core/src/tests.rs

@@ -346,6 +346,137 @@ async fn a_drained_deposit_is_not_minted_twice_across_turns() {
     );
 }
 
+#[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(),
+    })
+    .unwrap();
+    let LeeTransaction::Public(public_tx) = &deposit_tx else {
+        panic!("bridge deposit tx is public");
+    };
+
+    let vault_id = vault_core::compute_vault_account_id(programs::vault().id(), recipient_id);
+    let mut state = sequencer.chain().lock().unwrap().head_state().clone();
+
+    // 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"
+    );
+
+    // 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]
 fn transaction_pre_check_pass() {
     let tx = common::test_utils::produce_dummy_empty_transaction();

lez/storage/src/sequencer/mod.rs

@@ -496,36 +496,43 @@ impl RocksDBIO {
             return Ok(0);
         }
 
-        let mut records = self.get_pending_deposit_events()?;
-        let before = records.len();
+        // 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 records
-                .iter()
-                .any(|record| record.deposit_op_id == event.deposit_op_id)
+            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);
+        let accepted = records.len().saturating_sub(before_append);
 
         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();
+            let before_retain = records.len();
             records.retain(|record| !to_remove.contains(&record.deposit_op_id));
-            after_append.saturating_sub(records.len())
+            before_retain.saturating_sub(records.len())
         };
 
-        // Tracked separately rather than inferred from the length: one event can
-        // both observe a deposit and finalize another, which nets to the same
-        // count while the contents changed. The common finalizing event owns no
-        // pending record at all, and must not rewrite the cell.
+        // 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)?;
         }
@@ -569,7 +576,8 @@ impl RocksDBIO {
             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)).unwrap_or(usize::MAX),
+                usize::try_from(times.saturating_sub(matched))
+                    .expect("unmatched withdrawal count fits usize"),
             ));
 
             match stored.and_then(|count| count.checked_sub(times)) {