Merge pull request #642 from logos-blockchain/erhant/lez-chain-state-follow-up-fixes

fix(sequencer): follow-up fixes for #633 and #639

Cargo.lock

@@ -1193,6 +1193,7 @@ name = "bridge_core"
 version = "0.1.0"
 dependencies = [
  "lee_core",
+ "risc0-zkvm",
  "serde",
 ]
 

integration_tests/tests/bridge.rs

@@ -47,10 +47,11 @@ async fn public_bridge_deposit_invocation_is_dropped() -> anyhow::Result<()> {
     let bridge_account_id = system_accounts::bridge_account_id();
     let vault_program_id = programs::vault().id();
     let recipient_vault_id = vault_core::compute_vault_account_id(vault_program_id, recipient_id);
+    let receipt_id = bridge_core::deposit_receipt_account_id(programs::bridge().id(), [0_u8; 32]);
 
     let message = public_transaction::Message::try_new(
         programs::bridge().id(),
-        vec![bridge_account_id, recipient_vault_id],
+        vec![bridge_account_id, recipient_vault_id, receipt_id],
         vec![],
         bridge_core::Instruction::Deposit {
             l1_deposit_op_id: [0_u8; 32],
@@ -95,10 +96,11 @@ async fn public_bridge_deposit_with_zero_amount_is_rejected() -> anyhow::Result<
     let bridge_account_id = system_accounts::bridge_account_id();
     let vault_program_id = programs::vault().id();
     let recipient_vault_id = vault_core::compute_vault_account_id(vault_program_id, recipient_id);
+    let receipt_id = bridge_core::deposit_receipt_account_id(programs::bridge().id(), [0_u8; 32]);
 
     let message = public_transaction::Message::try_new(
         programs::bridge().id(),
-        vec![bridge_account_id, recipient_vault_id],
+        vec![bridge_account_id, recipient_vault_id, receipt_id],
         vec![],
         bridge_core::Instruction::Deposit {
             l1_deposit_op_id: [0_u8; 32],
@@ -155,8 +157,10 @@ async fn private_bridge_deposit_invocation_is_dropped() -> anyhow::Result<()> {
     let bridge_account_id = system_accounts::bridge_account_id();
     let vault_program_id = programs::vault().id();
     let recipient_vault_id = vault_core::compute_vault_account_id(vault_program_id, recipient_id);
+    let receipt_id = bridge_core::deposit_receipt_account_id(programs::bridge().id(), [0_u8; 32]);
 
-    // Get pre-state of bridge and vault accounts
+    // Get pre-state of bridge and vault accounts; the receipt is unminted (a
+    // default account), so the program would create it on a first mint.
     let bridge_pre = AccountWithMetadata::new(
         get_account(&ctx, bridge_account_id).await?,
         false,
@@ -167,6 +171,8 @@ async fn private_bridge_deposit_invocation_is_dropped() -> anyhow::Result<()> {
         false,
         recipient_vault_id,
     );
+    let receipt_pre =
+        AccountWithMetadata::new(lee_core::account::Account::default(), false, receipt_id);
 
     // Create program with dependencies
     let program_with_deps =
@@ -193,17 +199,25 @@ async fn private_bridge_deposit_invocation_is_dropped() -> anyhow::Result<()> {
 
     // Execute and prove the bridge deposit
     let (output, proof) = execute_and_prove(
-        vec![bridge_pre.clone(), vault_pre.clone()],
+        vec![bridge_pre.clone(), vault_pre.clone(), receipt_pre.clone()],
         instruction,
-        vec![InputAccountIdentity::Public, InputAccountIdentity::Public],
+        vec![
+            InputAccountIdentity::Public,
+            InputAccountIdentity::Public,
+            InputAccountIdentity::Public,
+        ],
         &program_with_deps,
     )
     .context("Failed to execute/prove bridge deposit")?;
 
     // Create privacy-preserving transaction from circuit output
     let message = privacy_preserving_transaction::Message::try_from_circuit_output(
-        vec![bridge_account_id, recipient_vault_id],
-        vec![bridge_pre.account.nonce, vault_pre.account.nonce],
+        vec![bridge_account_id, recipient_vault_id, receipt_id],
+        vec![
+            bridge_pre.account.nonce,
+            vault_pre.account.nonce,
+            receipt_pre.account.nonce,
+        ],
         output,
     )
     .context("Failed to build privacy-preserving bridge deposit message")?;

lee/state_machine/src/state/mod.rs

@@ -270,6 +270,12 @@ impl V03State {
             .unwrap_or_else(Account::default)
     }
 
+    /// Borrowing counterpart of [`Self::get_account_by_id`].
+    #[must_use]
+    pub fn get_account_by_id_ref(&self, account_id: AccountId) -> Option<&Account> {
+        self.public_state.get(&account_id)
+    }
+
     #[must_use]
     pub fn get_proof_for_commitment(&self, commitment: &Commitment) -> Option<MembershipProof> {
         self.private_state.0.get_proof_for(commitment)

lez/programs/bridge/core/Cargo.toml

@@ -9,4 +9,5 @@ workspace = true
 
 [dependencies]
 lee_core.workspace = true
+risc0-zkvm.workspace = true
 serde = { workspace = true, default-features = false }

lez/programs/bridge/core/src/lib.rs

@@ -3,14 +3,19 @@ use lee_core::{account::AccountId, program::ProgramId};
 use serde::{Deserialize, Serialize};
 
 const BRIDGE_SEED_DOMAIN_SEPARATOR: [u8; 32] = *b"/LEZ/v0.3/BridgeSeed/0000000000/";
+const DEPOSIT_RECEIPT_SEED_DOMAIN: [u8; 32] = *b"/LEZ/v0.3/BridgeDepositReceipt/0";
 
 #[derive(Serialize, Deserialize)]
 pub enum Instruction {
-    /// Transfers native tokens from the bridge PDA account to a recipient vault.
+    /// Transfers native tokens from the bridge PDA account to a recipient vault,
+    /// exactly once per `l1_deposit_op_id`.
     ///
-    /// Required accounts (2):
+    /// Required accounts (3):
     /// - Bridge PDA account
     /// - Recipient vault PDA account
+    /// - Deposit-receipt PDA account, derived from `l1_deposit_op_id`. Its existence records that
+    ///   this op id was already minted; a second application of the same op id finds it present and
+    ///   transfers nothing.
     Deposit {
         /// Deposit OP ID from L1, stored here to pin each [`Deposit`](Instruction::Deposit) to a
         /// Deposit Event on L1.
@@ -43,3 +48,58 @@ pub const fn compute_bridge_seed() -> PdaSeed {
 pub fn compute_bridge_account_id(bridge_program_id: ProgramId) -> AccountId {
     AccountId::for_public_pda(&bridge_program_id, &compute_bridge_seed())
 }
+
+/// Seed of the deposit-receipt PDA for `l1_deposit_op_id`, exposed so the guest
+/// can claim the account. Domain-separated from [`compute_bridge_seed`].
+#[must_use]
+pub fn deposit_receipt_seed(l1_deposit_op_id: [u8; 32]) -> PdaSeed {
+    use risc0_zkvm::sha::{Impl, Sha256 as _};
+
+    let mut bytes = [0_u8; 64];
+    bytes[..32].copy_from_slice(&DEPOSIT_RECEIPT_SEED_DOMAIN);
+    bytes[32..].copy_from_slice(&l1_deposit_op_id);
+
+    let seed: [u8; 32] = Impl::hash_bytes(&bytes)
+        .as_bytes()
+        .try_into()
+        .unwrap_or_else(|_| unreachable!());
+    PdaSeed::new(seed)
+}
+
+/// The deposit-receipt PDA whose existence marks `l1_deposit_op_id` as minted.
+#[must_use]
+pub fn deposit_receipt_account_id(
+    bridge_program_id: ProgramId,
+    l1_deposit_op_id: [u8; 32],
+) -> AccountId {
+    AccountId::for_public_pda(&bridge_program_id, &deposit_receipt_seed(l1_deposit_op_id))
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    const BRIDGE_ID: ProgramId = [7; 8];
+
+    #[test]
+    fn receipt_id_is_deterministic_per_op_id() {
+        let op = [3_u8; 32];
+        assert_eq!(
+            deposit_receipt_account_id(BRIDGE_ID, op),
+            deposit_receipt_account_id(BRIDGE_ID, op)
+        );
+    }
+
+    #[test]
+    fn distinct_op_ids_and_domains_do_not_collide() {
+        let a = deposit_receipt_account_id(BRIDGE_ID, [1; 32]);
+        let b = deposit_receipt_account_id(BRIDGE_ID, [2; 32]);
+        assert_ne!(a, b, "different op ids must derive different receipts");
+        // The op-id-derived seed must not alias the plain bridge PDA, even if an
+        // op id ever equals the bridge seed's raw bytes.
+        assert_ne!(
+            deposit_receipt_account_id(BRIDGE_ID, *compute_bridge_seed().as_bytes()),
+            compute_bridge_account_id(BRIDGE_ID)
+        );
+    }
+}

lez/programs/bridge/src/main.rs

@@ -1,6 +1,7 @@
 use bridge_core::Instruction;
-use lee_core::program::{
-    AccountPostState, ChainedCall, ProgramInput, ProgramOutput, read_lee_inputs,
+use lee_core::{
+    account::Account,
+    program::{AccountPostState, ChainedCall, Claim, ProgramInput, ProgramOutput, read_lee_inputs},
 };
 
 fn unchanged_post_states(
@@ -29,18 +30,17 @@ fn main() {
     );
 
     let pre_states_clone = pre_states.clone();
-    let post_states = unchanged_post_states(&pre_states_clone);
 
-    let chained_calls = match instruction {
+    let (post_states, chained_calls) = match instruction {
         Instruction::Deposit {
-            l1_deposit_op_id: _,
+            l1_deposit_op_id,
             vault_program_id,
             recipient_id,
             amount,
         } => {
-            let [bridge, recipient_vault] = pre_states
+            let [bridge, recipient_vault, receipt] = pre_states
                 .try_into()
-                .expect("Deposit requires exactly 2 accounts");
+                .expect("Deposit requires exactly 3 accounts");
 
             assert_eq!(
                 bridge.account_id,
@@ -54,20 +54,54 @@ fn main() {
                 "Second account must be recipient vault PDA"
             );
 
-            let mut bridge_for_vault = bridge;
-            bridge_for_vault.is_authorized = true;
+            assert_eq!(
+                receipt.account_id,
+                bridge_core::deposit_receipt_account_id(self_program_id, l1_deposit_op_id),
+                "Third account must be the deposit-receipt PDA"
+            );
 
-            vec![
-                ChainedCall::new(
-                    vault_program_id,
-                    vec![bridge_for_vault, recipient_vault],
-                    &vault_core::Instruction::Transfer {
-                        recipient_id,
-                        amount: u128::from(amount),
-                    },
-                )
-                .with_pda_seeds(vec![bridge_core::compute_bridge_seed()]),
-            ]
+            // Replay protection: the receipt PDA exists iff this op id was
+            // already minted. On replay it is non-default and the whole
+            // instruction is a no-op.
+            //
+            // Observability note: a no-op replay and a real first mint are both
+            // successful txs, so an indexer cannot tell "credited here" from
+            // "already credited by a peer" without deriving the receipt id and
+            // checking whether it existed before this block — the receipt claim
+            // is the only on-chain signal. Relevant once the explorer surfaces
+            // deposits.
+            if receipt.account != Account::default() {
+                (unchanged_post_states(&pre_states_clone), vec![])
+            } else {
+                // First mint: claim the receipt — its existence is the record,
+                // the account's contents are never read — and chain the vault
+                // transfer.
+                let receipt_post = AccountPostState::new_claimed_if_default(
+                    receipt.account,
+                    Claim::Pda(bridge_core::deposit_receipt_seed(l1_deposit_op_id)),
+                );
+
+                let post_states = vec![
+                    AccountPostState::new(bridge.account.clone()),
+                    AccountPostState::new(recipient_vault.account.clone()),
+                    receipt_post,
+                ];
+
+                let mut bridge_for_vault = bridge;
+                bridge_for_vault.is_authorized = true;
+                let chained_calls = vec![
+                    ChainedCall::new(
+                        vault_program_id,
+                        vec![bridge_for_vault, recipient_vault],
+                        &vault_core::Instruction::Transfer {
+                            recipient_id,
+                            amount: u128::from(amount),
+                        },
+                    )
+                    .with_pda_seeds(vec![bridge_core::compute_bridge_seed()]),
+                ];
+                (post_states, chained_calls)
+            }
         }
         Instruction::Withdraw {
             amount,
@@ -89,13 +123,14 @@ fn main() {
                 "Sender account must be owned by the authenticated transfer program"
             );
 
-            vec![ChainedCall::new(
+            let chained_calls = vec![ChainedCall::new(
                 auth_transfer_program_id,
                 vec![sender, bridge],
                 &authenticated_transfer_core::Instruction::Transfer {
                     amount: u128::from(amount),
                 },
-            )]
+            )];
+            (unchanged_post_states(&pre_states_clone), chained_calls)
         }
     };
 

lez/sequencer/core/src/block_publisher.rs

@@ -1,4 +1,4 @@
-use std::{pin::Pin, sync::Arc, time::Duration};
+use std::{sync::Arc, time::Duration};
 
 use anyhow::{Context as _, Result, anyhow, ensure};
 use common::block::Block;
@@ -47,26 +47,15 @@ use crate::config::BedrockConfig;
 /// backpressure if the drive task stalls (reconnect, long backfill).
 const PUBLISH_INBOX_CAPACITY: usize = 32;
 
-/// Sink for `Event::Published` checkpoints emitted by the drive task.
-/// Caller is responsible for persistence (e.g. writing to rocksdb).
-pub type CheckpointSink = Box<dyn Fn(SequencerCheckpoint) + Send + 'static>;
-
-/// Sink for finalized L2 block ids derived from `Event::TxsFinalized` and
-/// `Event::FinalizedInscriptions`. Caller is responsible for cleanup
-/// (e.g. marking pending blocks as finalized in storage).
-pub type FinalizedBlockSink = Box<dyn Fn(u64) + Send + 'static>;
-
-/// Sink for finalized Bedrock deposit events.
-pub type OnDepositEventSink =
-    Box<dyn Fn(DepositInfo) -> Pin<Box<dyn Future<Output = ()> + Send>> + Send + 'static>;
-
-/// Sink for finalized Bedrock withdraw events.
-pub type OnWithdrawEventSink =
-    Box<dyn Fn(WithdrawInfo) -> Pin<Box<dyn Future<Output = ()> + Send>> + Send + 'static>;
-
-/// The channel delta the follow path consumes from one `Event::BlocksProcessed`,
-/// with inscription payloads decoded into `(MsgId, Block)` pairs.
+/// Everything one `Event::BlocksProcessed` carries, with inscription payloads
+/// decoded into `(MsgId, Block)` pairs.
+///
+/// One struct rather than a sink per effect, because the `checkpoint` and
+/// everything it covers must reach the store in a single write.
 pub struct FollowUpdate {
+    /// Resume cursor for this event. Persist only together with the effects
+    /// below, never ahead of them.
+    pub checkpoint: SequencerCheckpoint,
     /// Inscriptions newly on the followed L1 branch, in channel order: they
     /// extend (or, after a reorg, replace part of) the `head` tier.
     pub adopted: Vec<(MsgId, Block)>,
@@ -76,19 +65,24 @@ pub struct FollowUpdate {
     /// Inscriptions whose containing L1 block reached finality: their blocks
     /// move into the irreversible `final` tier.
     pub finalized: Vec<(MsgId, Block)>,
+    /// Finalized Bedrock deposit events, to record and mint on L2.
+    pub deposits: Vec<DepositInfo>,
+    /// Finalized Bedrock withdraw events, to reconcile against local intents.
+    pub withdrawals: Vec<WithdrawInfo>,
 }
 
-/// Sink for the follow path: apply adopted/finalized blocks to chain state and
-/// revert orphaned ones.
+/// Sink for the follow path: apply the channel delta to chain state and
+/// persist the whole event in one write.
 pub type OnFollowSink = Box<dyn Fn(FollowUpdate) + Send + 'static>;
 
 /// Commands the drive task executes with `&mut sequencer`.
 enum Command {
-    /// Publish an inscription (+ atomic withdrawals); responds with the assigned `MsgId`.
+    /// Publish an inscription (+ atomic withdrawals); responds with the assigned
+    /// `MsgId` and the checkpoint that now includes it as pending.
     Publish {
         inscription: Inscription,
         withdrawals: Vec<WithdrawArg>,
-        resp: oneshot::Sender<Result<MsgId>>,
+        resp: oneshot::Sender<Result<(MsgId, SequencerCheckpoint)>>,
     },
 }
 
@@ -96,25 +90,25 @@ type CommandSender = mpsc::Sender<Command>;
 
 #[expect(async_fn_in_trait, reason = "We don't care about Send/Sync here")]
 pub trait BlockPublisherTrait: Sized {
-    #[expect(
-        clippy::too_many_arguments,
-        reason = "Looks better than bundling all those callbacks into a struct"
-    )]
     async fn new(
         config: &BedrockConfig,
         bedrock_signing_key: Ed25519Key,
         resubmit_interval: Duration,
         initial_checkpoint: Option<SequencerCheckpoint>,
-        on_checkpoint: CheckpointSink,
-        on_finalized_block: FinalizedBlockSink,
-        on_deposit_event: OnDepositEventSink,
-        on_withdraw_event: OnWithdrawEventSink,
         on_follow: OnFollowSink,
     ) -> Result<Self>;
 
-    /// Publish a block and return the `MsgId` zone-sdk assigned its inscription.
-    /// Zone-sdk drives the actual submission and retries internally.
-    async fn publish_block(&self, block: &Block, withdrawals: Vec<WithdrawArg>) -> Result<MsgId>;
+    /// Publish a block and return the `MsgId` zone-sdk assigned its inscription
+    /// together with the checkpoint that now holds it as pending. Zone-sdk
+    /// drives the actual submission and retries internally.
+    ///
+    /// The checkpoint must be persisted with the block — restoring an older one
+    /// drops the inscription from the pending set, and it is never resubmitted.
+    async fn publish_block(
+        &self,
+        block: &Block,
+        withdrawals: Vec<WithdrawArg>,
+    ) -> Result<(MsgId, SequencerCheckpoint)>;
 
     fn channel_id(&self) -> ChannelId;
 
@@ -168,10 +162,6 @@ impl BlockPublisherTrait for ZoneSdkPublisher {
         bedrock_signing_key: Ed25519Key,
         resubmit_interval: Duration,
         initial_checkpoint: Option<SequencerCheckpoint>,
-        on_checkpoint: CheckpointSink,
-        on_finalized_block: FinalizedBlockSink,
-        on_deposit_event: OnDepositEventSink,
-        on_withdraw_event: OnWithdrawEventSink,
         on_follow: OnFollowSink,
     ) -> Result<Self> {
         let basic_auth = config.auth.clone().map(Into::into);
@@ -229,7 +219,7 @@ impl BlockPublisherTrait for ZoneSdkPublisher {
                                 };
 
                                 let msg_result = published
-                                    .map(|(result, _checkpoint)| result.tx.inscription().this_msg);
+                                    .map(|(result, checkpoint)| (result.tx.inscription().this_msg, checkpoint));
                                 match &msg_result {
                                     Ok(_) if withdraw_count == 0 => {
                                         info!("Published block with the size of {data_byte_size} bytes");
@@ -254,8 +244,6 @@ impl BlockPublisherTrait for ZoneSdkPublisher {
                                     channel_update,
                                     finalized,
                                 } => {
-                                    on_checkpoint(checkpoint);
-
                                     let adopted = channel_update
                                         .adopted
                                         .iter()
@@ -269,29 +257,35 @@ impl BlockPublisherTrait for ZoneSdkPublisher {
                                         .collect();
 
                                     let mut finalized_blocks = Vec::new();
+                                    let mut deposits = Vec::new();
+                                    let mut withdrawals = Vec::new();
                                     for op in finalized.into_iter().flat_map(|item| item.ops) {
                                         match op {
                                             FinalizedOp::Inscription(inscription) => {
-                                                if let Some((msg, block)) =
+                                                if let Some(entry) =
                                                     block_from_inscription(&inscription)
                                                 {
-                                                    on_finalized_block(block.header.block_id);
-                                                    finalized_blocks.push((msg, block));
+                                                    finalized_blocks.push(entry);
                                                 }
                                             }
-                                            FinalizedOp::Deposit(deposit) => {
-                                                on_deposit_event(deposit).await;
-                                            }
+                                            FinalizedOp::Deposit(deposit) => deposits.push(deposit),
                                             FinalizedOp::Withdraw(withdraw) => {
-                                                on_withdraw_event(withdraw).await;
+                                                withdrawals.push(withdraw);
                                             }
                                         }
                                     }
 
+                                    // Nothing is awaited here: an await in this
+                                    // arm blocks the same task `publish_block`
+                                    // needs, and a non-turn sequencer never
+                                    // drains what it would be waiting on.
                                     on_follow(FollowUpdate {
+                                        checkpoint,
                                         adopted,
                                         orphaned,
                                         finalized: finalized_blocks,
+                                        deposits,
+                                        withdrawals,
                                     });
                                 }
                                 Event::Ready => {}
@@ -328,7 +322,11 @@ impl BlockPublisherTrait for ZoneSdkPublisher {
         })
     }
 
-    async fn publish_block(&self, block: &Block, withdrawals: Vec<WithdrawArg>) -> Result<MsgId> {
+    async fn publish_block(
+        &self,
+        block: &Block,
+        withdrawals: Vec<WithdrawArg>,
+    ) -> Result<(MsgId, SequencerCheckpoint)> {
         let data = borsh::to_vec(block).context("Failed to serialize block")?;
         let data_bounded: Inscription = data
             .try_into()

lez/sequencer/core/src/block_store.rs

@@ -155,13 +155,13 @@ impl SequencerStore {
     pub(crate) fn update(
         &mut self,
         block: &Block,
-        deposit_event_ids: &[HashType],
-        withdrawals: Vec<WithdrawalReconciliationKey>,
+        withdrawals: &[WithdrawalReconciliationKey],
         state: &V03State,
+        checkpoint: Option<&[u8]>,
     ) -> DbResult<()> {
         let new_transactions_map = block_to_transactions_map(block);
         self.dbio
-            .atomic_update(block, deposit_event_ids, withdrawals, state)?;
+            .atomic_update(block, withdrawals, state, checkpoint)?;
         self.tx_hash_to_block_map.extend(new_transactions_map);
         Ok(())
     }
@@ -194,10 +194,12 @@ impl SequencerStore {
         Ok(Some(checkpoint))
     }
 
+    /// Persists `checkpoint` on its own. Only valid when the effects it covers
+    /// are already durable — otherwise it must ride in the same write as them,
+    /// via [`storage::sequencer::StoreUpdate`].
     pub fn set_zone_checkpoint(&self, checkpoint: &SequencerCheckpoint) -> Result<()> {
-        let bytes =
-            serde_json::to_vec(checkpoint).context("Failed to serialize zone-sdk checkpoint")?;
-        self.dbio.put_zone_sdk_checkpoint_bytes(&bytes)?;
+        self.dbio
+            .put_zone_sdk_checkpoint_bytes(&checkpoint_bytes(checkpoint)?)?;
         Ok(())
     }
 
@@ -211,23 +213,15 @@ impl SequencerStore {
         self.dbio.put_zone_anchor(anchor)
     }
 
-    pub fn get_unfulfilled_deposit_events(&self) -> DbResult<Vec<PendingDepositEventRecord>> {
+    pub fn get_pending_deposit_events(&self) -> DbResult<Vec<PendingDepositEventRecord>> {
         self.dbio.get_pending_deposit_events()
     }
+}
 
-    pub fn is_deposit_event_submitted(&self, deposit_op_id: HashType) -> DbResult<bool> {
-        self.dbio.is_deposit_event_submitted(deposit_op_id)
-    }
-
-    /// Marks the given deposit events submitted in `block_id`, in one write.
-    pub fn mark_deposit_events_submitted(
-        &self,
-        deposit_op_ids: &[HashType],
-        submitted_block_id: u64,
-    ) -> DbResult<()> {
-        self.dbio
-            .mark_deposit_events_submitted(deposit_op_ids, submitted_block_id)
-    }
+/// The checkpoint's on-disk encoding. `serde_json` because `SequencerCheckpoint`
+/// derives serde but not borsh; paired with `get_zone_checkpoint`'s decode.
+pub(crate) fn checkpoint_bytes(checkpoint: &SequencerCheckpoint) -> Result<Vec<u8>> {
+    serde_json::to_vec(checkpoint).context("Failed to serialize zone-sdk checkpoint")
 }
 
 pub(crate) fn block_to_transactions_map(block: &Block) -> HashMap<HashType, u64> {
@@ -278,9 +272,7 @@ mod tests {
         assert_eq!(None, retrieved_tx);
         // Add the block with the transaction
         let dummy_state = V03State::new();
-        node_store
-            .update(&block, &[], vec![], &dummy_state)
-            .unwrap();
+        node_store.update(&block, &[], &dummy_state, None).unwrap();
         // Try again
         let output = node_store.get_transaction_by_hash(tx.hash());
         assert_eq!(Some((tx, 1)), output);
@@ -345,9 +337,7 @@ mod tests {
         let block_hash = block.header.hash;
 
         let dummy_state = V03State::new();
-        node_store
-            .update(&block, &[], vec![], &dummy_state)
-            .unwrap();
+        node_store.update(&block, &[], &dummy_state, None).unwrap();
 
         // Verify that the latest block meta now equals the new block's hash
         let latest_meta = node_store.latest_block_meta().unwrap().unwrap();
@@ -383,9 +373,7 @@ mod tests {
         let block_id = block.header.block_id;
 
         let dummy_state = V03State::new();
-        node_store
-            .update(&block, &[], vec![], &dummy_state)
-            .unwrap();
+        node_store.update(&block, &[], &dummy_state, None).unwrap();
 
         // Verify initial status is Pending
         let retrieved_block = node_store.get_block_at_id(block_id).unwrap().unwrap();
@@ -434,7 +422,7 @@ mod tests {
             // Add a new block
             let block = common::test_utils::produce_dummy_block(1, None, vec![tx.clone()]);
             node_store
-                .update(&block, &[], vec![], &V03State::new())
+                .update(&block, &[], &V03State::new(), None)
                 .unwrap();
         }
 

lez/sequencer/core/src/lib.rs

@@ -1,4 +1,5 @@
 use std::{
+    collections::VecDeque,
     path::Path,
     sync::{Arc, Mutex},
     time::Instant,
@@ -31,7 +32,7 @@ pub use mock::SequencerCoreWithMockClients;
 use num_bigint::BigUint;
 pub use storage::error::DbError;
 use storage::sequencer::{
-    RocksDBIO,
+    RocksDBIO, StoreUpdate,
     sequencer_cells::{PendingDepositEventRecord, WithdrawalReconciliationKey, ZoneAnchorRecord},
 };
 
@@ -188,17 +189,12 @@ impl<BP: BlockPublisherTrait> SequencerCore<BP> {
         let is_fresh_start = initial_checkpoint.is_none();
 
         let (mempool, mempool_handle) = MemPool::new(config.mempool_max_size);
-        replay_unfulfilled_deposit_events(&store, mempool_handle.clone());
 
         let block_publisher = BP::new(
             &config.bedrock_config,
             bedrock_signing_key,
             config.retry_pending_blocks_timeout,
             initial_checkpoint,
-            Self::on_checkpoint(store.dbio()),
-            Self::on_finalized_block(store.dbio()),
-            Self::on_deposit_event(store.dbio(), mempool_handle.clone()),
-            Self::on_withdraw_event(store.dbio()),
             Self::on_follow(store.dbio(), Arc::clone(&chain), mempool_handle.clone()),
         )
         .await
@@ -241,8 +237,9 @@ impl<BP: BlockPublisherTrait> SequencerCore<BP> {
                 "First pending block on fresh start should be the genesis block"
             );
 
+            let mut last_checkpoint = None;
             for block in &pending_blocks {
-                block_publisher
+                let (_msg, checkpoint) = block_publisher
                     .publish_block(block, vec![])
                     .await
                     .unwrap_or_else(|err| {
@@ -251,6 +248,16 @@ impl<BP: BlockPublisherTrait> SequencerCore<BP> {
                             block.header.block_id
                         )
                     });
+                last_checkpoint = Some(checkpoint);
+            }
+
+            // These blocks are already stored, so only the sdk's pending set
+            // moved. Checkpoints are cumulative — persisting just the last one
+            // is both sufficient and the only way to keep this loop linear.
+            if let Some(checkpoint) = last_checkpoint {
+                store
+                    .set_zone_checkpoint(&checkpoint)
+                    .expect("Failed to persist checkpoint after republishing on fresh start");
             }
         }
 
@@ -448,171 +455,36 @@ impl<BP: BlockPublisherTrait> SequencerCore<BP> {
             }
         }
 
-        // Persist like the follow path: the tip meta stays pinned to the head
-        // tip even when the reconstructed block lands below it.
-        let head_tip = chain.head_tip().map(|head| BlockMeta::from(&head));
-        let final_meta = chain.final_tip().map(|meta| BlockMeta::from(&meta));
-        store
-            .dbio()
-            .store_followed_blocks(
-                &[(block, true)],
-                head_tip.as_ref(),
-                chain.head_state(),
-                final_meta.as_ref().map(|meta| (chain.final_state(), meta)),
-            )
-            .context("Failed to persist reconstructed block")?;
-
-        // Mark the deposits' pending records submitted so the production-time
-        // guard drops the mints cold-start backfill re-queued for them. Withdraw
-        // intents are deliberately not counted: backfill already re-delivered and
-        // dropped their finalized L1 events, so an increment here would never be
-        // consumed and would leave a phantom count.
-        let deposit_event_ids: Vec<_> = block
+        // A reconstructed block is finalized, so any deposit it mints is
+        // permanently reflected in state (its receipt PDA); drop the pending
+        // record backfill may have re-delivered, so the drain stops re-minting.
+        let finalized_deposit_ids: Vec<_> = block
             .body
             .transactions
             .iter()
             .filter_map(extract_bridge_deposit_id)
             .collect();
-        store
-            .mark_deposit_events_submitted(&deposit_event_ids, block_id)
-            .context("Failed to mark reconstructed deposits submitted")?;
 
+        // The tip meta stays pinned to the head tip even when the reconstructed
+        // block lands below it, and the anchor only advances if the block
+        // itself landed.
+        let head_tip = chain.head_tip().map(|head| BlockMeta::from(&head));
+        let final_meta = chain.final_tip().map(|meta| BlockMeta::from(&meta));
         store
-            .set_zone_anchor(&record)
-            .context("Failed to persist zone anchor")?;
+            .dbio()
+            .store_update(&StoreUpdate {
+                blocks: &[(block, true)],
+                head_tip: head_tip.as_ref(),
+                final_snapshot: final_meta.as_ref().map(|meta| (chain.final_state(), meta)),
+                remove_deposit_records: &finalized_deposit_ids,
+                zone_anchor: Some(&record),
+                ..StoreUpdate::new(chain.head_state())
+            })
+            .context("Failed to persist reconstructed block")?;
+
         Ok(())
     }
 
-    fn on_checkpoint(dbio: Arc<RocksDBIO>) -> block_publisher::CheckpointSink {
-        Box::new(move |cp| {
-            let bytes = match serde_json::to_vec(&cp) {
-                Ok(b) => b,
-                Err(err) => {
-                    error!("Failed to serialize zone-sdk checkpoint: {err:#}");
-                    return;
-                }
-            };
-            if let Err(err) = dbio.put_zone_sdk_checkpoint_bytes(&bytes) {
-                error!("Failed to persist zone-sdk checkpoint: {err:#}");
-            }
-        })
-    }
-
-    fn on_finalized_block(dbio: Arc<RocksDBIO>) -> block_publisher::FinalizedBlockSink {
-        Box::new(move |block_id| {
-            // NOTE: Theoretically Zone SDK may report finalization happening multiple times for the
-            // same block. In practice this is very unlikely to happen. For that to
-            // happen Sequencer should crash between receiving Finalized and Checkpoint events while
-            // these events happen very fast (because Checkpoints are generated by Zone SDK
-            // locally).
-
-            if let Err(err) = dbio.clean_pending_blocks_up_to(block_id) {
-                error!("Failed to mark pending blocks finalized up to {block_id}: {err:#}");
-            }
-
-            match dbio.remove_fulfilled_pending_deposit_events_up_to_block(block_id) {
-                Ok(0) => {}
-                Ok(removed) => {
-                    info!(
-                        "Removed {removed} fulfilled pending deposit events up to finalized block {block_id}"
-                    );
-                }
-                Err(err) => {
-                    error!(
-                        "Failed to remove fulfilled pending deposit events up to block {block_id}: {err:#}"
-                    );
-                }
-            }
-        })
-    }
-
-    fn on_deposit_event(
-        dbio: Arc<RocksDBIO>,
-        mempool_handle: MemPoolHandle<(TransactionOrigin, LeeTransaction)>,
-    ) -> block_publisher::OnDepositEventSink {
-        Box::new(move |deposit| {
-            // NOTE: Theoretically Zone SDK may report multiple identical deposits. In practice this
-            // is very unlikely to happen. For that to happen Sequencer should crash
-            // between receiving Deposit and Checkpoint events while these events happen
-            // very fast (because Checkpoints are generated by Zone SDK locally).
-
-            let dbio = Arc::clone(&dbio);
-            let mempool_handle = mempool_handle.clone();
-
-            Box::pin(async move {
-                let id_hex = hex::encode(deposit.op_id);
-                info!("Observed Bedrock Deposit event with id: {id_hex}");
-
-                let event_record = pending_deposit_event_record(&deposit);
-
-                match dbio.add_pending_deposit_event(event_record.clone()) {
-                    Ok(true) => {}
-                    Ok(false) => {
-                        info!(
-                            "Deposit event {id_hex} already persisted as unfulfilled, skipping duplicate enqueue",
-                        );
-                        return;
-                    }
-                    Err(err) => {
-                        error!(
-                            "Failed to persist unfulfilled deposit event {id_hex} before enqueue: {err:#}. Deposit will be lost.",
-                        );
-                        return;
-                    }
-                }
-
-                let tx = match build_bridge_deposit_tx_from_event(&event_record) {
-                    Ok(tx) => tx,
-                    Err(err) => {
-                        error!(
-                            "Failed to build transaction from Bedrock deposit event {id_hex}: {err:#}. Deposit will be lost.",
-                        );
-                        return;
-                    }
-                };
-
-                if let Err(err) = mempool_handle
-                    .push((TransactionOrigin::Sequencer, tx))
-                    .await
-                {
-                    error!(
-                        "Failed to queue sequencer transaction built from finalized Bedrock event: {err:#}. Deposit will be lost."
-                    );
-                }
-            })
-        })
-    }
-
-    fn on_withdraw_event(dbio: Arc<RocksDBIO>) -> block_publisher::OnWithdrawEventSink {
-        Box::new(move |withdraw| {
-            let dbio = Arc::clone(&dbio);
-            Box::pin(async move {
-                let hash_encoded = hex::encode(withdraw.tx_hash.as_ref());
-                let withdraw_key = match withdraw_event_reconciliation_key(&withdraw.op.outputs) {
-                    Ok(key) => key,
-                    Err(err) => {
-                        error!(
-                            "Failed to build reconciliation key for Bedrock Withdraw event with tx_hash {hash_encoded}: {err:#}"
-                        );
-                        return;
-                    }
-                };
-
-                match dbio.consume_unseen_withdraw_count(withdraw_key) {
-                    Ok(true) => {
-                        info!("Validated Bedrock Withdraw event with tx_hash: {hash_encoded}");
-                    }
-                    Ok(false) => warn!(
-                        "Unexpected Bedrock Withdraw event with tx_hash {hash_encoded}: no matching unseen withdraw found"
-                    ),
-                    Err(err) => error!(
-                        "Failed to reconcile Bedrock Withdraw event with tx_hash {hash_encoded}: {err:#}"
-                    ),
-                }
-            })
-        })
-    }
-
     /// Publisher sink adapter over [`apply_follow_update`].
     fn on_follow(
         dbio: Arc<RocksDBIO>,
@@ -626,21 +498,17 @@ impl<BP: BlockPublisherTrait> SequencerCore<BP> {
 
     /// Produces a new block from mempool transactions and publishes it via zone-sdk.
     pub async fn produce_new_block(&mut self) -> Result<u64> {
-        let BlockWithMeta {
-            block,
-            deposit_event_ids,
-            withdrawals,
-        } = self
+        let BlockWithMeta { block, withdrawals } = self
             .build_block_from_mempool()
             .context("Failed to build block from mempool transactions")?;
 
-        let withdrawal_reconciliation_keys = withdrawals
+        let withdrawal_reconciliation_keys: Vec<_> = withdrawals
             .iter()
             .map(|withdraw| withdraw_event_reconciliation_key(&withdraw.outputs))
-            .collect::<Result<_>>()
+            .collect::<Result<Vec<_>>>()
             .context("Failed to build reconciliation keys for block withdrawals")?;
 
-        let this_msg = self
+        let (this_msg, checkpoint) = self
             .block_publisher
             .publish_block(&block, withdrawals)
             .await
@@ -649,8 +517,8 @@ impl<BP: BlockPublisherTrait> SequencerCore<BP> {
         self.record_produced_block(
             this_msg,
             &block,
-            &deposit_event_ids,
-            withdrawal_reconciliation_keys,
+            &withdrawal_reconciliation_keys,
+            &checkpoint,
         )?;
 
         Ok(block.header.block_id)
@@ -669,9 +537,11 @@ impl<BP: BlockPublisherTrait> SequencerCore<BP> {
         &mut self,
         this_msg: MsgId,
         block: &Block,
-        deposit_event_ids: &[HashType],
-        withdrawal_reconciliation_keys: Vec<WithdrawalReconciliationKey>,
+        withdrawal_reconciliation_keys: &[WithdrawalReconciliationKey],
+        checkpoint: &block_publisher::SequencerCheckpoint,
     ) -> Result<()> {
+        let checkpoint_bytes = block_store::checkpoint_bytes(checkpoint)?;
+
         let mut chain = self.chain.lock().expect("chain state mutex poisoned");
         match chain.apply_produced(this_msg, block) {
             AcceptOutcome::Applied => {
@@ -679,11 +549,14 @@ impl<BP: BlockPublisherTrait> SequencerCore<BP> {
                 // with the follow path.
                 self.store.update(
                     block,
-                    deposit_event_ids,
                     withdrawal_reconciliation_keys,
                     chain.head_state(),
+                    Some(&checkpoint_bytes),
                 )?;
             }
+            // Neither branch persists anything, checkpoint included: the
+            // inscription it holds as pending belongs to a block that is not
+            // ours to keep.
             AcceptOutcome::AlreadyApplied => {
                 warn!(
                     "Produced block {} lost a competing-write race, skipping persistence",
@@ -704,20 +577,14 @@ impl<BP: BlockPublisherTrait> SequencerCore<BP> {
     /// Validates and applies a single mempool transaction to the current state.
     /// Returns `Ok(true)` if the transaction was valid and applied, `Ok(false)` if
     /// it was skipped due to validation failure.
-    #[expect(
-        clippy::too_many_arguments,
-        reason = "splitting the produce-path accumulators into a struct buys nothing"
-    )]
     fn apply_mempool_transaction(
-        store: &SequencerStore,
         state: &mut lee::V03State,
         origin: TransactionOrigin,
         tx: &LeeTransaction,
         block_height: u64,
         timestamp: u64,
-        deposit_event_ids: &mut Vec<HashType>,
         withdrawals: &mut Vec<WithdrawArg>,
-    ) -> Result<bool> {
+    ) -> bool {
         let tx_hash = tx.hash();
         match origin {
             TransactionOrigin::User => {
@@ -727,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;
                     }
                 };
 
@@ -742,25 +609,30 @@ impl<BP: BlockPublisherTrait> SequencerCore<BP> {
                     panic!("Sequencer may only generate Public transactions, found {tx:#?}");
                 };
 
-                if let Some(deposit_op_id) = extract_bridge_deposit_id(tx) {
-                    if store
-                        .is_deposit_event_submitted(deposit_op_id)
-                        .context("Failed to check whether deposit was already submitted")?
-                    {
-                        info!("Skipping already-submitted bridge deposit {deposit_op_id}");
-                        return Ok(false);
-                    }
-                    deposit_event_ids.push(deposit_op_id);
+                // 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.
+                //
+                // 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;
                 }
-
-                state
-                    .transition_from_public_transaction(public_tx, block_height, timestamp)
-                    .context("Failed to execute sequencer-generated transaction")?;
             }
         }
 
         info!("Validated transaction with hash {tx_hash}, including it in block");
-        Ok(true)
+        true
     }
 
     fn build_block_from_mempool(&mut self) -> Result<BlockWithMeta> {
@@ -780,9 +652,36 @@ impl<BP: BlockPublisherTrait> SequencerCore<BP> {
         };
 
         let mut valid_transactions = Vec::new();
-        let mut deposit_event_ids = Vec::new();
         let mut withdrawals = Vec::new();
 
+        // Bridge deposit mints are drained from the store, not the mempool: the
+        // follow path records the event durably but cannot enqueue the mint
+        // itself (it runs on the publisher's drive task, where an await stalls
+        // the very task production needs). Draining here also subsumes the old
+        // startup replay.
+        //
+        // Skip any deposit whose receipt PDA already exists in the state we
+        // build on — it was minted by us or by a peer whose block we adopted.
+        // An orphan reverts the receipt with the block, so the next turn
+        // re-mints without any bookkeeping of our own.
+        let mut pending_deposits: VecDeque<LeeTransaction> = self
+            .store
+            .get_pending_deposit_events()
+            .context("Failed to load pending deposit events")?
+            .into_iter()
+            .filter(|record| !deposit_already_minted(&working_state, record.deposit_op_id))
+            .filter_map(|record| {
+                build_bridge_deposit_tx_from_event(&record)
+                    .inspect_err(|err| {
+                        warn!(
+                            "Skipping pending deposit event {} due to tx build failure: {err:#}",
+                            hex::encode(record.deposit_op_id)
+                        );
+                    })
+                    .ok()
+            })
+            .collect();
+
         let max_block_size = usize::try_from(self.sequencer_config.max_block_size.as_u64())
             .expect("`max_block_size` should fit into usize");
 
@@ -792,7 +691,14 @@ impl<BP: BlockPublisherTrait> SequencerCore<BP> {
         let clock_tx = clock_invocation(new_block_timestamp);
         let clock_lee_tx = LeeTransaction::Public(clock_tx.clone());
 
-        while let Some((origin, tx)) = self.mempool.pop() {
+        // Pending deposit mints first, then user work. `from_store` is not the
+        // same as a `Sequencer` origin — the cross-zone watcher pushes those
+        // into the mempool too.
+        while let Some((origin, tx, from_store)) = pending_deposits
+            .pop_front()
+            .map(|tx| (TransactionOrigin::Sequencer, tx, true))
+            .or_else(|| self.mempool.pop().map(|(origin, tx)| (origin, tx, false)))
+        {
             let tx_hash = tx.hash();
 
             let temp_valid_transactions = [
@@ -817,20 +723,22 @@ impl<BP: BlockPublisherTrait> SequencerCore<BP> {
                     "Transaction with hash {tx_hash} deferred to next block: \
                      block size {block_size} bytes would exceed limit of {max_block_size} bytes",
                 );
-                self.mempool.push_front((origin, tx));
+                // A deposit mint needs no requeue: its record stays unfulfilled
+                // in the store and is drained again on the next turn.
+                if !from_store {
+                    self.mempool.push_front((origin, tx));
+                }
                 break;
             }
 
             if Self::apply_mempool_transaction(
-                &self.store,
                 &mut working_state,
                 origin,
                 &tx,
                 new_block_height,
                 new_block_timestamp,
-                &mut deposit_event_ids,
                 &mut withdrawals,
-            )? {
+            ) {
                 valid_transactions.push(tx);
             }
 
@@ -861,11 +769,7 @@ impl<BP: BlockPublisherTrait> SequencerCore<BP> {
             now.elapsed().as_secs()
         );
 
-        Ok(BlockWithMeta {
-            block,
-            deposit_event_ids,
-            withdrawals,
-        })
+        Ok(BlockWithMeta { block, withdrawals })
     }
 
     /// Reads the current head state under the lock without cloning it, so callers
@@ -896,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<()> {
@@ -941,15 +844,26 @@ impl<BP: BlockPublisherTrait> SequencerCore<BP> {
 
 struct BlockWithMeta {
     block: Block,
-    deposit_event_ids: Vec<HashType>,
     withdrawals: Vec<WithdrawArg>,
 }
 
+/// Whether `deposit_op_id`'s mint is already reflected in `state` — its receipt
+/// PDA exists. The receipt is the exactly-once ledger the bridge program keeps.
+fn deposit_already_minted(state: &lee::V03State, deposit_op_id: HashType) -> bool {
+    let receipt_id =
+        bridge_core::deposit_receipt_account_id(programs::bridge().id(), deposit_op_id.0);
+    state
+        .get_account_by_id_ref(receipt_id)
+        .is_some_and(|receipt| *receipt != lee::Account::default())
+}
+
 /// Feed one channel delta into the follow state and mirror it to the store:
 /// revert orphaned, then apply and persist adopted and finalized blocks.
 /// Production builds on this same head. Wired to the publisher via
 /// [`SequencerCore::on_follow`]; a free function so tests can drive it directly.
 ///
+/// Everything the event produced lands in one write — see [`StoreUpdate`].
+///
 /// TODO: unlike the indexer's ingest loop, this path does not retry
 /// `is_retryable` (transient) apply failures — a failed block just parks and
 /// relies on a valid successor or a restart. `ChainState` never emits
@@ -962,14 +876,42 @@ fn apply_follow_update(
     update: block_publisher::FollowUpdate,
 ) {
     let block_publisher::FollowUpdate {
+        checkpoint,
         adopted,
         orphaned,
         finalized,
+        deposits,
+        withdrawals,
     } = update;
 
+    let checkpoint_bytes = block_store::checkpoint_bytes(&checkpoint)
+        .unwrap_or_else(|err| panic!("Failed to serialize zone-sdk checkpoint: {err:#}"));
+
+    // NOTE: Theoretically Zone SDK may re-deliver an already seen deposit or
+    // finalization. Both are idempotent here: a deposit already on record is
+    // not re-appended, and a finalization only ever moves the tier forward.
+    let deposit_records: Vec<PendingDepositEventRecord> =
+        deposits.iter().map(pending_deposit_event_record).collect();
+
+    // A withdraw whose outputs we cannot read has no counter to reconcile
+    // against; log and drop it rather than fail the whole update.
+    let consumed_withdrawals: Vec<WithdrawalReconciliationKey> = withdrawals
+        .iter()
+        .filter_map(|withdraw| {
+            withdraw_event_reconciliation_key(&withdraw.op.outputs)
+                .inspect_err(|err| {
+                    error!(
+                        "Failed to build reconciliation key for Bedrock Withdraw event with tx_hash {}: {err:#}",
+                        hex::encode(withdraw.tx_hash.as_ref())
+                    );
+                })
+                .ok()
+        })
+        .collect();
+
     // The lock is held across the persist below so disk writes land in apply
     // order — the produce path persists under this same lock.
-    let resubmit_txs = {
+    let (resubmit_txs, outcome) = {
         let mut chain = chain.lock().expect("chain state mutex poisoned");
 
         // User txs of orphaned blocks, returned to the mempool below.
@@ -987,16 +929,28 @@ 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;
+            // FIXME: thread the finalized inscription's L1 slot instead of
+            // `Slot::from(0)`; only used for the invalid-finalized stall.
+            // logos-blockchain PR #3147 surfaces it as `FinalizedTx.l1_slot` —
+            // wire it through `FollowUpdate::finalized` once the zone-sdk pin is
+            // bumped past that (a separate PR).
+            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.
@@ -1006,36 +960,65 @@ fn apply_follow_update(
         });
         let head_tip = chain.head_tip().map(|tip| BlockMeta::from(&tip));
 
-        // One atomic write for the whole update: blocks, tip meta and the
-        // state after the last block land together, so a crash can never
-        // leave the stored state ahead of the stored blocks. A persist
-        // failure is fatal: the in-memory chain has already advanced, and
-        // continuing would leave a permanent gap in the store.
-        //
-        // The `panic!` ends the drive task, whose cancellation halts the node.
-        //
-        // TODO: the zone-sdk checkpoint is persisted by `on_checkpoint`
-        // before this write; a crash in between resumes past these blocks
-        // without them ever landing in the store. Full `BlocksProcessed`
-        // atomicity (checkpoint + blocks + state in one batch, per the sdk's
-        // event contract) is a follow-up.
-        dbio.store_followed_blocks(
-            &to_persist,
-            head_tip.as_ref(),
-            chain.head_state(),
-            final_meta.as_ref().map(|meta| (chain.final_state(), meta)),
-        )
-        .unwrap_or_else(|err| panic!("Failed to persist followed blocks: {err:#}"));
+        // Every block at or below the highest finalized one is irreversible, so
+        // stored blocks there can be marked finalized.
+        let last_finalized = irreversible.iter().map(|block| block.header.block_id).max();
 
-        resubmit_txs
+        // 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> = irreversible
+            .iter()
+            .flat_map(|block| block.body.transactions.iter())
+            .filter_map(extract_bridge_deposit_id)
+            .collect();
+
+        // A persist failure is fatal: the in-memory chain has already advanced,
+        // and continuing would leave a permanent gap in the store. The `panic!`
+        // ends the drive task, whose cancellation halts the node.
+        let outcome = dbio
+            .store_update(&StoreUpdate {
+                checkpoint: Some(&checkpoint_bytes),
+                blocks: &to_persist,
+                head_tip: head_tip.as_ref(),
+                final_snapshot: final_meta.as_ref().map(|meta| (chain.final_state(), meta)),
+                finalized_up_to: last_finalized,
+                new_deposit_events: &deposit_records,
+                remove_deposit_records: &finalized_deposit_ids,
+                consumed_withdrawals: &consumed_withdrawals,
+                ..StoreUpdate::new(chain.head_state())
+            })
+            .unwrap_or_else(|err| panic!("Failed to persist follow update: {err:#}"));
+
+        (resubmit_txs, outcome)
     };
 
+    if outcome.accepted_deposits > 0 {
+        info!(
+            "Recorded {} Bedrock Deposit event(s); their mints are drained from the store on our next turn",
+            outcome.accepted_deposits
+        );
+    }
+    for withdrawal in &outcome.unmatched_withdrawals {
+        warn!(
+            "Unexpected Bedrock Withdraw event of {} to {}: no matching unseen withdraw found",
+            withdrawal.amount,
+            hex::encode(withdrawal.bedrock_account_pk)
+        );
+    }
+
     // 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.
+    // We use [`try_push`] here because this is called from the publisher's
+    // drive task, and only block production drains the mempool. A blocking
+    // push would stall the drive task, and a sequencer that is not on turn
+    // never produces — so nothing would ever drain it again.
+    //
+    // TODO: a full mempool still drops the transaction; a durable resubmit
+    // queue is a follow-up.
     for tx in resubmit_txs {
         let tx_hash = tx.hash();
         if let Err(err) = mempool_handle.try_push((TransactionOrigin::User, tx)) {
@@ -1044,55 +1027,6 @@ fn apply_follow_update(
     }
 }
 
-/// Checks the database for any pending deposit events that have not yet been marked as submitted in
-/// a block, and re-queues them in the mempool in a separate async task for inclusion in the next
-/// block.
-fn replay_unfulfilled_deposit_events(
-    store: &SequencerStore,
-    mempool_handle: MemPoolHandle<(TransactionOrigin, LeeTransaction)>,
-) {
-    let replay_records: Vec<PendingDepositEventRecord> = store
-        .get_unfulfilled_deposit_events()
-        .expect("Failed to load unfulfilled deposit events")
-        .into_iter()
-        .filter(|record| record.submitted_in_block_id.is_none())
-        .collect();
-
-    if replay_records.is_empty() {
-        return;
-    }
-
-    info!(
-        "Found {} unfulfilled deposit events in DB, re-queueing",
-        replay_records.len()
-    );
-    tokio::spawn(async move {
-        for record in replay_records {
-            let tx = match build_bridge_deposit_tx_from_event(&record) {
-                Ok(tx) => tx,
-                Err(err) => {
-                    warn!(
-                        "Skipping replay of pending deposit event {} due to tx build failure: {err:#}",
-                        hex::encode(record.deposit_op_id)
-                    );
-                    continue;
-                }
-            };
-
-            if let Err(err) = mempool_handle
-                .push((TransactionOrigin::Sequencer, tx))
-                .await
-            {
-                error!(
-                    "Failed to re-queue unfulfilled deposit event {} from DB: {err:#}",
-                    hex::encode(record.deposit_op_id)
-                );
-                break;
-            }
-        }
-    });
-}
-
 /// The pre-genesis state: `testnet_initial_state` plus the bridge-lock holdings,
 /// the only accounts seeded outside any transaction. Cross-zone config is seeded
 /// by genesis `InitConfig` transactions and reconstructed by replaying them.
@@ -1232,7 +1166,6 @@ fn pending_deposit_event_record(deposit: &DepositInfo) -> PendingDepositEventRec
         source_tx_hash: HashType(deposit.tx_hash.0),
         amount: deposit.amount,
         metadata: deposit.metadata.clone().into(),
-        submitted_in_block_id: None,
     }
 }
 
@@ -1244,10 +1177,18 @@ fn build_bridge_deposit_tx_from_event(event: &PendingDepositEventRecord) -> Resu
     let vault_program_id = programs::vault().id();
     let recipient_vault_id =
         vault_core::compute_vault_account_id(vault_program_id, metadata.recipient_id);
+    // The receipt PDA carries the exactly-once check: the program reads it to
+    // detect a replay, so it must be in the tx's account list.
+    let receipt_id =
+        bridge_core::deposit_receipt_account_id(bridge_program_id, event.deposit_op_id.0);
 
     let message = Message::try_new(
         bridge_program_id,
-        vec![system_accounts::bridge_account_id(), recipient_vault_id],
+        vec![
+            system_accounts::bridge_account_id(),
+            recipient_vault_id,
+            receipt_id,
+        ],
         vec![],
         bridge_core::Instruction::Deposit {
             l1_deposit_op_id: event.deposit_op_id.0,

lez/sequencer/core/src/mock.rs

@@ -3,16 +3,16 @@ use std::time::Duration;
 use anyhow::Result;
 use common::block::Block;
 use futures::Stream;
-use logos_blockchain_core::mantle::ops::channel::{ChannelId, MsgId};
+use logos_blockchain_core::{
+    header::HeaderId,
+    mantle::ops::channel::{ChannelId, MsgId},
+};
 use logos_blockchain_key_management_system_service::keys::Ed25519Key;
 use logos_blockchain_zone_sdk::{Slot, ZoneMessage, sequencer::WithdrawArg};
 use tokio_util::sync::CancellationToken;
 
 use crate::{
-    block_publisher::{
-        BlockPublisherTrait, CheckpointSink, FinalizedBlockSink, OnDepositEventSink, OnFollowSink,
-        OnWithdrawEventSink, SequencerCheckpoint,
-    },
+    block_publisher::{BlockPublisherTrait, OnFollowSink, SequencerCheckpoint},
     config::BedrockConfig,
 };
 
@@ -54,16 +54,15 @@ impl BlockPublisherTrait for MockBlockPublisher {
         _bedrock_signing_key: Ed25519Key,
         _resubmit_interval: Duration,
         _initial_checkpoint: Option<SequencerCheckpoint>,
-        _on_checkpoint: CheckpointSink,
-        _on_finalized_block: FinalizedBlockSink,
-        _on_deposit_event: OnDepositEventSink,
-        _on_withdraw_event: OnWithdrawEventSink,
         _on_follow: OnFollowSink,
     ) -> Result<Self> {
         Ok(Self {
             channel_id: config.channel_id,
             driver_cancellation: CancellationToken::new(),
-            tip_slot: None,
+            // An existing but empty channel: `None` means *missing*, which the
+            // startup guard reads as a wiped Bedrock. Tests that want that say
+            // so via [`Self::with_canned_channel`].
+            tip_slot: Some(Slot::from(0)),
             messages: Vec::new(),
         })
     }
@@ -72,11 +71,11 @@ impl BlockPublisherTrait for MockBlockPublisher {
         &self,
         block: &Block,
         _bridge_withdrawals: Vec<WithdrawArg>,
-    ) -> Result<MsgId> {
+    ) -> Result<(MsgId, SequencerCheckpoint)> {
         // Deterministic per-block id so head dedup behaves in tests.
         //
         // TODO: should we allow more "mockability" here?
-        Ok(MsgId::from(block.header.hash.0))
+        Ok((MsgId::from(block.header.hash.0), mock_checkpoint()))
     }
 
     fn channel_id(&self) -> ChannelId {
@@ -108,3 +107,16 @@ impl BlockPublisherTrait for MockBlockPublisher {
         Ok(futures::stream::iter(messages))
     }
 }
+
+/// A zeroed checkpoint, for [`MockBlockPublisher::publish_block`] and for tests
+/// building a [`crate::block_publisher::FollowUpdate`]. Tests only assert *that*
+/// a checkpoint was persisted alongside its effects, never what is in it.
+#[must_use]
+pub(crate) fn mock_checkpoint() -> SequencerCheckpoint {
+    SequencerCheckpoint {
+        last_msg_id: MsgId::from([0; 32]),
+        pending_txs: Vec::new(),
+        lib: HeaderId::from([0; 32]),
+        lib_slot: Slot::from(0),
+    }
+}

lez/sequencer/core/src/tests.rs

@@ -22,7 +22,12 @@ use lee_core::{
     account::{AccountWithMetadata, Nonce},
     program::PdaSeed,
 };
-use logos_blockchain_core::mantle::ops::channel::{ChannelId, MsgId};
+use logos_blockchain_core::mantle::{
+    ledger::Inputs,
+    ops::channel::{ChannelId, MsgId, deposit::Metadata},
+    tx::TxHash,
+};
+use logos_blockchain_zone_sdk::sequencer::DepositInfo;
 use mempool::MemPoolHandle;
 use storage::sequencer::sequencer_cells::PendingDepositEventRecord;
 use tempfile::tempdir;
@@ -33,9 +38,9 @@ use crate::{
     block_publisher::FollowUpdate,
     block_store::SequencerStore,
     build_bridge_deposit_tx_from_event, build_genesis_state,
-    config::{BedrockConfig, SequencerConfig},
-    is_sequencer_only_program,
-    mock::SequencerCoreWithMockClients,
+    config::{BedrockConfig, GenesisAction, SequencerConfig},
+    deposit_already_minted, is_sequencer_only_program,
+    mock::{SequencerCoreWithMockClients, mock_checkpoint},
     resubmittable_txs,
 };
 
@@ -46,6 +51,19 @@ struct DepositMetadataForEncoding {
     recipient_id: lee::AccountId,
 }
 
+/// A follow update carrying nothing, to fill in the fields a test does not
+/// exercise via `..empty_follow_update()`.
+fn empty_follow_update() -> FollowUpdate {
+    FollowUpdate {
+        checkpoint: mock_checkpoint(),
+        adopted: Vec::new(),
+        orphaned: Vec::new(),
+        finalized: Vec::new(),
+        deposits: Vec::new(),
+        withdrawals: Vec::new(),
+    }
+}
+
 fn setup_sequencer_config() -> SequencerConfig {
     let tempdir = tempfile::tempdir().unwrap();
     let home = tempdir.path().to_path_buf();
@@ -211,8 +229,10 @@ async fn start_from_config_panics_when_db_open_returns_non_not_found_error() {
 }
 
 #[tokio::test]
-async fn start_from_config_replays_unfulfilled_deposit_events_from_db() {
-    let config = setup_sequencer_config();
+async fn unfulfilled_deposit_events_are_drained_from_the_store_on_production() {
+    let mut config = setup_sequencer_config();
+    // The mint moves funds out of the bridge account, so it has to hold some.
+    config.genesis = vec![GenesisAction::SupplyBridgeAccount { balance: 1_000_000 }];
     let deposit_op_id = [13_u8; 32];
     let expected_amount = 1_u64;
     let recipient_id = initial_public_user_accounts()[0].account_id;
@@ -227,7 +247,6 @@ async fn start_from_config_replays_unfulfilled_deposit_events_from_db() {
         source_tx_hash: HashType([7_u8; 32]),
         amount: expected_amount,
         metadata: borsh::to_vec(&DepositMetadataForEncoding { recipient_id }).unwrap(),
-        submitted_in_block_id: None,
     };
 
     {
@@ -241,36 +260,221 @@ async fn start_from_config_replays_unfulfilled_deposit_events_from_db() {
         assert!(inserted);
     }
 
+    // The mint never goes through the mempool: the record is the queue, and
+    // production drains it. That is what makes a restart — or a follow event
+    // arriving while a full mempool would have dropped the push — lossless.
     let (mut sequencer, _mempool_handle) =
         SequencerCoreWithMockClients::start_from_config(config).await;
+    assert!(
+        sequencer.mempool.pop().is_none(),
+        "deposit mints are drained from the store, never queued in the mempool"
+    );
 
-    let (origin, tx) = tokio::time::timeout(Duration::from_secs(5), async {
-        loop {
-            if let Some((origin, tx)) = sequencer.mempool.pop() {
-                return (origin, tx);
-            }
+    let block_id = sequencer.produce_new_block().await.unwrap();
+    let block = sequencer
+        .store
+        .get_block_at_id(block_id)
+        .unwrap()
+        .expect("produced block is stored");
+    assert!(
+        block
+            .body
+            .transactions
+            .iter()
+            .any(|tx| tx_is_bridge_deposit(tx, deposit_op_id, expected_amount)),
+        "the drained deposit mint should be included in the produced block"
+    );
 
-            tokio::time::sleep(Duration::from_millis(100)).await;
-        }
+    // The record stays until its deposit finalizes; exactly-once is enforced by
+    // the receipt PDA now in head state, not by any marker on the record.
+    assert!(
+        sequencer
+            .store
+            .get_pending_deposit_events()
+            .unwrap()
+            .iter()
+            .any(|event| event.deposit_op_id == HashType(deposit_op_id)),
+        "the record remains until the deposit finalizes"
+    );
+    assert!(
+        sequencer.with_state(|state| deposit_already_minted(state, HashType(deposit_op_id))),
+        "the deposit's receipt PDA marks it minted in head state"
+    );
+}
+
+#[tokio::test]
+async fn a_drained_deposit_is_not_minted_twice_across_turns() {
+    let mut config = setup_sequencer_config();
+    config.genesis = vec![GenesisAction::SupplyBridgeAccount { balance: 1_000_000 }];
+    let deposit_op_id = [17_u8; 32];
+    let recipient_id = initial_public_user_accounts()[0].account_id;
+
+    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: 1,
+            metadata: borsh::to_vec(&DepositMetadataForEncoding { recipient_id }).unwrap(),
+        })
+        .unwrap();
+
+    let first = sequencer.produce_new_block().await.unwrap();
+    let second = sequencer.produce_new_block().await.unwrap();
+
+    let minted_in = |block_id: u64| {
+        sequencer
+            .store
+            .get_block_at_id(block_id)
+            .unwrap()
+            .expect("produced block is stored")
+            .body
+            .transactions
+            .iter()
+            .filter(|tx| tx_is_bridge_deposit(tx, deposit_op_id, 1))
+            .count()
+    };
+
+    assert_eq!(minted_in(first), 1);
+    assert_eq!(
+        minted_in(second),
+        0,
+        "the receipt PDA from the first mint must keep the drain from re-minting"
+    );
+}
+
+#[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(),
     })
-    .await
-    .expect("Timed out waiting for pending deposit event to be replayed into mempool");
+    .unwrap();
+    let LeeTransaction::Public(public_tx) = &deposit_tx else {
+        panic!("bridge deposit tx is public");
+    };
 
-    match origin {
-        TransactionOrigin::Sequencer => {}
-        TransactionOrigin::User => {
-            panic!("Unexpected user transaction in empty mempool replay test")
-        }
-    }
+    let vault_id = vault_core::compute_vault_account_id(programs::vault().id(), recipient_id);
+    let mut state = sequencer.chain().lock().unwrap().head_state().clone();
 
-    assert!(tx_is_bridge_deposit(&tx, deposit_op_id, expected_amount));
+    // 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"
+    );
 
-    let pending_events = sequencer.store.get_unfulfilled_deposit_events().unwrap();
-    let replayed_event = pending_events
-        .into_iter()
-        .find(|event| event.deposit_op_id == HashType(deposit_op_id))
-        .expect("Pending deposit event should remain in DB until included in a block");
-    assert!(replayed_event.submitted_in_block_id.is_none());
+    // 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]
@@ -1288,7 +1492,6 @@ fn resubmittable_txs_drops_clock_and_bridge_deposits() {
             recipient_id: initial_public_user_accounts()[0].account_id,
         })
         .unwrap(),
-        submitted_in_block_id: None,
     })
     .unwrap();
     let withdraw_tx = {
@@ -1335,6 +1538,67 @@ fn resubmittable_txs_of_blocks_without_user_txs_is_empty() {
     assert!(resubmittable_txs(&clock_only).is_empty());
 }
 
+#[tokio::test]
+async fn follow_update_persists_the_checkpoint_with_its_effects() {
+    let config = setup_sequencer_config();
+    let (sequencer, mempool_handle) = SequencerCoreWithMockClients::start_from_config(config).await;
+    let genesis_meta = sequencer
+        .store
+        .latest_block_meta()
+        .unwrap()
+        .expect("genesis meta is set");
+
+    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,
+        FollowUpdate {
+            adopted: vec![(MsgId::from([1; 32]), peer_block)],
+            ..empty_follow_update()
+        },
+    );
+
+    // The checkpoint is the sdk resume cursor; landing it without the block
+    // would let a restart stream past a block the store never got.
+    assert!(
+        sequencer.store.get_zone_checkpoint().unwrap().is_some(),
+        "the event's checkpoint must be persisted alongside the block it covers"
+    );
+    assert!(sequencer.store.get_block_at_id(2).unwrap().is_some());
+}
+
+#[tokio::test]
+async fn follow_update_records_deposits_for_the_production_drain() {
+    let config = setup_sequencer_config();
+    let (sequencer, mempool_handle) = SequencerCoreWithMockClients::start_from_config(config).await;
+
+    let recipient_id = initial_public_user_accounts()[0].account_id;
+    let metadata = borsh::to_vec(&DepositMetadataForEncoding { recipient_id }).unwrap();
+    let deposit = DepositInfo {
+        op_id: [21; 32],
+        tx_hash: TxHash::from([9; 32]),
+        channel_id: ChannelId::from([0; 32]),
+        inputs: Inputs::empty(),
+        amount: 5,
+        metadata: Metadata::try_from(metadata).expect("deposit metadata fits"),
+    };
+
+    apply_follow_update(
+        &sequencer.store.dbio(),
+        &sequencer.chain(),
+        &mempool_handle,
+        FollowUpdate {
+            deposits: vec![deposit],
+            ..empty_follow_update()
+        },
+    );
+
+    let pending = sequencer.store.get_pending_deposit_events().unwrap();
+    assert_eq!(pending.len(), 1);
+    assert_eq!(pending[0].deposit_op_id, HashType([21; 32]));
+}
+
 #[tokio::test]
 async fn follow_adopted_peer_block_applies_and_persists() {
     let config = setup_sequencer_config();
@@ -1362,8 +1626,7 @@ async fn follow_adopted_peer_block_applies_and_persists() {
         &mempool_handle,
         FollowUpdate {
             adopted: vec![(MsgId::from([1; 32]), peer_block.clone())],
-            orphaned: vec![],
-            finalized: vec![],
+            ..empty_follow_update()
         },
     );
 
@@ -1410,8 +1673,7 @@ async fn follow_redelivery_of_own_block_is_deduped() {
         &mempool_handle,
         FollowUpdate {
             adopted: vec![(MsgId::from(block2.header.hash.0), block2)],
-            orphaned: vec![],
-            finalized: vec![],
+            ..empty_follow_update()
         },
     );
 
@@ -1452,7 +1714,7 @@ async fn follow_orphan_reverts_head_and_requeues_user_txs() {
         FollowUpdate {
             adopted: vec![],
             orphaned: vec![(MsgId::from(block2.header.hash.0), block2)],
-            finalized: vec![],
+            ..empty_follow_update()
         },
     );
 
@@ -1496,6 +1758,7 @@ async fn follow_finalized_own_block_moves_final_tier_and_marks_store() {
             adopted: vec![],
             orphaned: vec![],
             finalized: vec![(MsgId::from(block2.header.hash.0), block2)],
+            ..empty_follow_update()
         },
     );
 
@@ -1533,6 +1796,7 @@ async fn follow_finalized_backfill_block_is_applied_and_marked_finalized() {
             adopted: vec![],
             orphaned: vec![],
             finalized: vec![(MsgId::from([2; 32]), peer_block.clone())],
+            ..empty_follow_update()
         },
     );
 
@@ -1550,6 +1814,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();
@@ -1593,7 +1927,7 @@ async fn restart_restores_head_tier_and_recovers_from_orphan() {
         FollowUpdate {
             adopted: vec![(MsgId::from([21; 32]), block2_prime.clone())],
             orphaned: vec![(MsgId::from([20; 32]), block2)],
-            finalized: vec![],
+            ..empty_follow_update()
         },
     );
 
@@ -1646,6 +1980,7 @@ async fn restart_reanchors_on_the_persisted_final_snapshot() {
                 adopted: vec![],
                 orphaned: vec![],
                 finalized: vec![(MsgId::from(block2.header.hash.0), block2)],
+                ..empty_follow_update()
             },
         );
     }
@@ -1695,7 +2030,7 @@ async fn record_produced_block_skips_persistence_on_lost_race() {
             MsgId::from(our_block.header.hash.0),
             &our_block,
             &[],
-            vec![],
+            &mock_checkpoint(),
         )
         .unwrap();
 
@@ -1719,7 +2054,12 @@ async fn record_produced_block_skips_persistence_when_block_no_longer_chains() {
     // The head reorged under us: our block's parent is no longer the tip.
     let stale = common::test_utils::produce_dummy_block(2, Some(HashType([9; 32])), vec![]);
     sequencer
-        .record_produced_block(MsgId::from(stale.header.hash.0), &stale, &[], vec![])
+        .record_produced_block(
+            MsgId::from(stale.header.hash.0),
+            &stale,
+            &[],
+            &mock_checkpoint(),
+        )
         .unwrap();
 
     assert!(sequencer.store.get_block_at_id(2).unwrap().is_none());
@@ -1760,6 +2100,7 @@ async fn follow_update_persists_blocks_meta_and_state_atomically() {
             ],
             orphaned: vec![],
             finalized: vec![(MsgId::from([2; 32]), block2)],
+            ..empty_follow_update()
         },
     );
 

lez/sequencer/core/src/tests/reconstruction.rs

@@ -295,7 +295,6 @@ fn deposit_event_record(
             recipient_id: recipient,
         })
         .unwrap(),
-        submitted_in_block_id: None,
     }
 }
 
@@ -321,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;
@@ -343,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();
@@ -367,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()
@@ -378,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(
@@ -397,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);
@@ -525,7 +535,6 @@ async fn reconstruction_reconciles_already_finished_deposit() {
         .await
         .unwrap();
     seq_a.produce_new_block().await.unwrap();
-    let deposit_block_id = seq_a.block_store().latest_block_meta().unwrap().unwrap().id;
 
     let messages = channel_from_store(seq_a.block_store(), 10);
     let tip_slot = messages.last().unwrap().1;
@@ -561,18 +570,19 @@ async fn reconstruction_reconciles_already_finished_deposit() {
         "already-finished deposit must be applied exactly once"
     );
 
-    // The pending event is now marked submitted in the reconstructed block, so the
-    // startup replay would not re-queue it — no double mint on restart.
-    let record = store_b
-        .get_unfulfilled_deposit_events()
-        .unwrap()
-        .into_iter()
-        .find(|event| event.deposit_op_id == HashType(deposit_op_id))
-        .expect("pending deposit event should still be recorded");
-    assert_eq!(
-        record.submitted_in_block_id,
-        Some(deposit_block_id),
-        "reconstruction must reconcile the already-finished deposit against its channel block"
+    // The mint's receipt PDA is in the reconstructed state, and reconstruction
+    // dropped the pending record backfill had re-delivered — so the production
+    // drain sees the deposit as minted and never re-emits it.
+    assert!(
+        crate::deposit_already_minted(
+            chain_b.lock().unwrap().head_state(),
+            HashType(deposit_op_id)
+        ),
+        "the reconstructed deposit's receipt marks it minted"
+    );
+    assert!(
+        store_b.get_pending_deposit_events().unwrap().is_empty(),
+        "reconstruction drops the finalized deposit's pending record"
     );
 }
 

lez/storage/src/lib.rs

@@ -67,6 +67,18 @@ pub trait DBIO {
         cell.put_batch(self.db(), params, write_batch)
     }
 
+    /// Stage a cell deletion into `write_batch`, the counterpart of
+    /// [`Self::put_batch`]. Deleting an absent key is a no-op (rocksdb
+    /// semantics).
+    fn del_batch<T: SimpleStorableCell>(
+        &self,
+        params: T::KeyParams,
+        write_batch: &mut WriteBatch,
+    ) -> DbResult<()> {
+        write_batch.delete_cf(&T::column_ref(self.db()), T::key_constructor(params)?);
+        Ok(())
+    }
+
     /// Delete a cell. Deleting an absent key is a no-op (rocksdb semantics).
     fn del<T: SimpleStorableCell>(&self, params: T::KeyParams) -> DbResult<()> {
         let cf_ref = T::column_ref(self.db());

lez/storage/src/sequencer/mod.rs

@@ -1,4 +1,4 @@
-use std::{path::Path, sync::Arc};
+use std::{collections::BTreeMap, path::Path, sync::Arc};
 
 use borsh::{BorshDeserialize, BorshSerialize};
 use common::{
@@ -13,10 +13,7 @@ use rocksdb::{
 
 use crate::{
     CF_BLOCK_NAME, CF_META_NAME, DB_META_FIRST_BLOCK_IN_DB_KEY, DBIO, DbResult,
-    cells::{
-        SimpleStorableCell,
-        shared_cells::{BlockCell, FirstBlockCell, FirstBlockSetCell, LastBlockCell},
-    },
+    cells::shared_cells::{BlockCell, FirstBlockCell, FirstBlockSetCell, LastBlockCell},
     error::DbError,
     sequencer::sequencer_cells::{
         FinalBlockMetaCellOwned, FinalBlockMetaCellRef, FinalLeeStateCellOwned,
@@ -97,6 +94,77 @@ impl DbDump {
     }
 }
 
+/// Everything one sequencer event writes, staged into a single [`WriteBatch`]
+/// by [`RocksDBIO::store_update`].
+///
+/// The point of the struct is the `checkpoint`: it is the zone-sdk's resume
+/// cursor, so it must land in the *same* write as the effects it covers.
+/// Persisted ahead of them, a crash in between resumes the stream past blocks
+/// that never reached the store — a gap the node cannot backfill.
+pub struct StoreUpdate<'update> {
+    /// Serialized zone-sdk checkpoint for this event.
+    pub checkpoint: Option<&'update [u8]>,
+
+    /// `(block, finalized)` payloads to write.
+    pub blocks: &'update [(&'update Block, bool)],
+
+    /// Head tip to pin the stored chain to; `None` only for an empty chain.
+    pub head_tip: Option<&'update BlockMeta>,
+    /// State after the last applied block.
+    pub head_state: &'update V03State,
+
+    /// `(state, meta)` of the final tier, when it advanced.
+    pub final_snapshot: Option<(&'update V03State, &'update BlockMeta)>,
+    /// Highest block id this event made irreversible: stored blocks at or below
+    /// it become [`BedrockStatus::Finalized`].
+    pub finalized_up_to: Option<u64>,
+
+    /// Deposit events observed on L1, recorded unless already pending.
+    pub new_deposit_events: &'update [PendingDepositEventRecord],
+    /// Deposit op ids whose mint finalized: their pending records are dropped.
+    pub remove_deposit_records: &'update [HashType],
+    /// L1 withdraw events to reconcile against the local unseen counters.
+    pub consumed_withdrawals: &'update [WithdrawalReconciliationKey],
+    /// L2 withdraw intents this update raises, awaiting their L1 event.
+    pub new_withdraw_intents: &'update [WithdrawalReconciliationKey],
+
+    /// Advance the channel-read anchor.
+    pub zone_anchor: Option<&'update ZoneAnchorRecord>,
+}
+
+impl<'update> StoreUpdate<'update> {
+    /// An update that writes nothing but the caller's head `state`, to be
+    /// filled in with `..StoreUpdate::new(state)`.
+    #[must_use]
+    pub const fn new(head_state: &'update V03State) -> Self {
+        Self {
+            checkpoint: None,
+            blocks: &[],
+            head_tip: None,
+            head_state,
+            final_snapshot: None,
+            finalized_up_to: None,
+            new_deposit_events: &[],
+            remove_deposit_records: &[],
+            consumed_withdrawals: &[],
+            new_withdraw_intents: &[],
+            zone_anchor: None,
+        }
+    }
+}
+
+/// What [`RocksDBIO::store_update`] observed while staging, for the caller to
+/// act on *after* the write committed.
+#[derive(Debug, Default)]
+pub struct StoreUpdateOutcome {
+    /// How many deposit events were newly recorded; the rest were already
+    /// pending, and so already owed.
+    pub accepted_deposits: usize,
+    /// Withdraw events with no matching local unseen counter, one entry per
+    /// unmatched occurrence.
+    pub unmatched_withdrawals: Vec<WithdrawalReconciliationKey>,
+}
+
 pub struct RocksDBIO {
     pub db: DBWithThreadMode<MultiThreaded>,
 }
@@ -384,10 +452,6 @@ impl RocksDBIO {
             .map_or_else(Vec::new, |cell| cell.0))
     }
 
-    fn put_pending_deposit_events(&self, records: &[PendingDepositEventRecord]) -> DbResult<()> {
-        self.put(&PendingDepositEventsCellRef(records), ())
-    }
-
     fn put_pending_deposit_events_batch(
         &self,
         records: &[PendingDepositEventRecord],
@@ -396,135 +460,218 @@ impl RocksDBIO {
         self.put_batch(&PendingDepositEventsCellRef(records), (), batch)
     }
 
+    /// Records a single deposit event, returning whether it was new.
+    /// One-shot form of [`RocksDBIO::store_update`]'s `new_deposit_events`.
     pub fn add_pending_deposit_event(&self, event: PendingDepositEventRecord) -> DbResult<bool> {
-        let mut records = self.get_pending_deposit_events()?;
-        if records
-            .iter()
-            .any(|record| record.deposit_op_id == event.deposit_op_id)
-        {
+        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);
         }
-        records.push(event);
-        self.put_pending_deposit_events(&records)?;
-        Ok(true)
-    }
-
-    /// Marks the given deposit events submitted in `block_id`, in one write.
-    pub fn mark_deposit_events_submitted(
-        &self,
-        deposit_op_ids: &[HashType],
-        submitted_block_id: u64,
-    ) -> DbResult<()> {
-        if deposit_op_ids.is_empty() {
-            return Ok(());
-        }
-        let mut batch = WriteBatch::default();
-        self.mark_pending_deposit_events_submitted(deposit_op_ids, submitted_block_id, &mut batch)?;
         self.db.write(batch).map_err(|rerr| {
             DbError::rocksdb_cast_message(
                 rerr,
-                Some("Failed to mark deposit events submitted".to_owned()),
+                Some("Failed to add pending deposit event".to_owned()),
             )
-        })
+        })?;
+        Ok(accepted > 0)
     }
 
-    fn mark_pending_deposit_events_submitted(
+    /// Stages every mutation of the pending-deposit records into `batch`,
+    /// returning how many were newly appended.
+    ///
+    /// The records live in a *single* whole-vector cell, so each mutation kind
+    /// cannot re-read it from disk and stage its own `put`: a later read would
+    /// not see the earlier staged write and would silently drop it. Everything
+    /// is folded in memory here instead, and written exactly once.
+    fn stage_pending_deposit_events(
         &self,
-        deposit_op_ids: &[HashType],
-        submitted_block_id: u64,
+        new_events: &[PendingDepositEventRecord],
+        remove_op_ids: &[HashType],
         batch: &mut WriteBatch,
     ) -> DbResult<usize> {
-        let mut records = self.get_pending_deposit_events()?;
-        let mut updated: usize = 0;
-
-        for record in records
-            .iter_mut()
-            .filter(|record| deposit_op_ids.contains(&record.deposit_op_id))
-        {
-            record.submitted_in_block_id = Some(submitted_block_id);
-            updated = updated.saturating_add(1);
+        if new_events.is_empty() && remove_op_ids.is_empty() {
+            return Ok(0);
         }
 
-        if updated > 0 {
+        // 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 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_append);
+
+        let removed = if remove_op_ids.is_empty() {
+            0
+        } else {
+            let before_retain = records.len();
+            records.retain(|record| !to_remove.contains(&record.deposit_op_id));
+            before_retain.saturating_sub(records.len())
+        };
+
+        // 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)?;
         }
-
-        Ok(updated)
+        Ok(accepted)
     }
 
-    pub fn remove_fulfilled_pending_deposit_events_up_to_block(
+    /// Stages the unseen-withdraw decrements for one update into `batch`,
+    /// returning one entry per occurrence that matched no local counter.
+    ///
+    /// Occurrences are folded per key for the same reason as the deposit
+    /// records: two withdrawals in one update can share a reconciliation key,
+    /// and a per-occurrence disk read would miss the staged decrement.
+    fn stage_consumed_withdrawals(
         &self,
-        finalized_block_id: u64,
-    ) -> DbResult<usize> {
-        let mut records = self.get_pending_deposit_events()?;
-        let before = records.len();
-        records.retain(|record| {
-            record
-                .submitted_in_block_id
-                .is_none_or(|submitted_id| submitted_id > finalized_block_id)
-        });
-
-        let removed = before.saturating_sub(records.len());
-        if removed > 0 {
-            self.put_pending_deposit_events(&records)?;
+        withdrawals: &[WithdrawalReconciliationKey],
+        batch: &mut WriteBatch,
+    ) -> DbResult<Vec<WithdrawalReconciliationKey>> {
+        let mut unmatched = Vec::new();
+        if withdrawals.is_empty() {
+            return Ok(unmatched);
         }
 
-        Ok(removed)
+        // A `Vec` rather than a map: the per-update count is tiny, and it keeps
+        // the staging order deterministic.
+        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)),
+            }
+        }
+
+        for (withdrawal, times) in occurrences {
+            let stored = self
+                .get_opt::<UnseenWithdrawCountCell>(withdrawal)?
+                .map(|cell| cell.0);
+
+            // A stored `count` satisfies `count + 1` occurrences: the last one
+            // consumes the key by deleting it. Matches the one-shot
+            // [`Self::consume_unseen_withdraw_count`].
+            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))
+                    .expect("unmatched withdrawal count fits usize"),
+            ));
+
+            match stored.and_then(|count| count.checked_sub(times)) {
+                Some(count) => {
+                    self.put_batch(&UnseenWithdrawCountCell(count), withdrawal, batch)?;
+                }
+                // Only stage a delete for a key that was actually there, so a
+                // fully unmatched update leaves the batch empty.
+                None if stored.is_some() => {
+                    self.del_batch::<UnseenWithdrawCountCell>(withdrawal, batch)?;
+                }
+                None => {}
+            }
+        }
+
+        Ok(unmatched)
     }
 
-    /// Whether a bridge deposit for `deposit_op_id` is already recorded as
-    /// included in a block (its pending record is marked submitted).
-    pub fn is_deposit_event_submitted(&self, deposit_op_id: HashType) -> DbResult<bool> {
-        Ok(self.get_pending_deposit_events()?.iter().any(|record| {
-            record.deposit_op_id == deposit_op_id && record.submitted_in_block_id.is_some()
-        }))
+    /// Collects the [`BedrockStatus::Finalized`] flip for every stored pending
+    /// block at or below `last_finalized` into `to_write`.
+    ///
+    /// Reads from disk, so blocks the caller is writing itself are already in
+    /// `to_write` and keep their own version — one `put` per block id, no
+    /// reliance on the order writes are staged in.
+    fn collect_finalized_up_to(&self, last_finalized: u64, to_write: &mut BTreeMap<u64, Block>) {
+        let newly_finalized: Vec<Block> = self
+            .get_all_blocks()
+            .filter_map(Result::ok)
+            .filter(|block| {
+                matches!(block.bedrock_status, BedrockStatus::Pending)
+                    && block.header.block_id <= last_finalized
+            })
+            .collect();
+
+        for mut block in newly_finalized {
+            block.bedrock_status = BedrockStatus::Finalized;
+            to_write.entry(block.header.block_id).or_insert(block);
+        }
     }
 
-    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
+    /// local intent. One-shot form of [`RocksDBIO::store_update`]'s
+    /// `consumed_withdrawals`.
     pub fn consume_unseen_withdraw_count(
         &self,
         withdrawal: WithdrawalReconciliationKey,
     ) -> DbResult<bool> {
-        let Some(current) = self
-            .get_opt::<UnseenWithdrawCountCell>(withdrawal)?
-            .map(|cell| cell.0)
-        else {
-            return Ok(false);
-        };
-
-        if let Some(next) = current.checked_sub(1) {
-            self.put(&UnseenWithdrawCountCell(next), withdrawal)?;
-        } else {
-            let cf_meta = self.meta_column();
-            let db_key =
-                <UnseenWithdrawCountCell as SimpleStorableCell>::key_constructor(withdrawal)?;
-
-            self.db.delete_cf(&cf_meta, db_key).map_err(|rerr| {
-                DbError::rocksdb_cast_message(
-                    rerr,
-                    Some("Failed to delete unseen withdraw count".to_owned()),
-                )
-            })?;
-        }
-
-        Ok(true)
+        let mut batch = WriteBatch::default();
+        let unmatched = self.stage_consumed_withdrawals(&[withdrawal], &mut batch)?;
+        self.db.write(batch).map_err(|rerr| {
+            DbError::rocksdb_cast_message(
+                rerr,
+                Some("Failed to consume unseen withdraw count".to_owned()),
+            )
+        })?;
+        Ok(unmatched.is_empty())
     }
 
     pub fn put_block(&self, block: &Block, first: bool, batch: &mut WriteBatch) -> DbResult<()> {
@@ -623,20 +770,23 @@ impl RocksDBIO {
         Ok(())
     }
 
-    /// Mark every pending block with `block_id <= last_finalized` as finalized.
-    /// Idempotent — already-finalized blocks are skipped.
+    /// Mark every pending block with `block_id <= last_finalized` as finalized,
+    /// in one atomic write. Idempotent — already-finalized blocks are skipped.
+    /// One-shot form of [`RocksDBIO::store_update`]'s `finalized_up_to`.
     pub fn clean_pending_blocks_up_to(&self, last_finalized: u64) -> DbResult<()> {
-        let pending_ids: Vec<u64> = self
-            .get_all_blocks()
-            .filter_map(Result::ok)
-            .filter(|b| matches!(b.bedrock_status, BedrockStatus::Pending))
-            .map(|b| b.header.block_id)
-            .filter(|id| *id <= last_finalized)
-            .collect();
-        for id in pending_ids {
-            self.mark_block_as_finalized(id)?;
+        let mut to_write = BTreeMap::new();
+        self.collect_finalized_up_to(last_finalized, &mut to_write);
+
+        let mut batch = WriteBatch::default();
+        for block in to_write.values() {
+            self.put_block_payload(block, &mut batch)?;
         }
-        Ok(())
+        self.db.write(batch).map_err(|rerr| {
+            DbError::rocksdb_cast_message(
+                rerr,
+                Some("Failed to mark pending blocks finalized".to_owned()),
+            )
+        })
     }
 
     pub fn mark_block_as_finalized(&self, block_id: u64) -> DbResult<()> {
@@ -691,8 +841,8 @@ impl RocksDBIO {
         Ok(())
     }
 
-    /// One-block form of [`Self::store_followed_blocks`], with the block as the
-    /// head tip and no final snapshot. Production always uses the batch form.
+    /// One-block form of [`Self::store_update`], with the block as the head tip
+    /// and no final snapshot. Production always uses the batch form.
     #[cfg(test)]
     fn store_followed_block(
         &self,
@@ -700,16 +850,17 @@ impl RocksDBIO {
         state: &V03State,
         finalized: bool,
     ) -> DbResult<()> {
-        self.store_followed_blocks(
-            &[(block, finalized)],
-            Some(&BlockMeta::from(block)),
-            state,
-            None,
-        )
+        self.store_update(&StoreUpdate {
+            blocks: &[(block, finalized)],
+            head_tip: Some(&BlockMeta::from(block)),
+            ..StoreUpdate::new(state)
+        })
+        .map(|_outcome| ())
     }
 
-    /// Persists a batch of followed blocks, the caller's head-tip `state`, and
-    /// the optional final-tier snapshot in one atomic write.
+    /// Persists everything one sequencer event produced — checkpoint, blocks,
+    /// tip meta, head state, final snapshot, deposit and withdraw bookkeeping
+    /// and the channel anchor — in one atomic write.
     ///
     /// The tip meta is pinned to `head_tip`, and blocks stored above it (left
     /// behind by a net-shortening reorg) are deleted in the same write, so
@@ -717,69 +868,115 @@ impl RocksDBIO {
     ///
     /// Per block: skips the payload write when the store already holds it (by
     /// id and hash), unless `finalized` is set, which rewrites it with the
-    /// finalized status. A no-op update (nothing to write, tip unchanged)
-    /// writes nothing.
+    /// finalized status.
     ///
-    /// TODO: the zone-sdk checkpoint is persisted by `on_checkpoint` *before*
-    /// this write. Full `BlocksProcessed` atomicity (checkpoint, blocks, state
-    /// and orphan reverts in one batch) is a follow-up.
-    pub fn store_followed_blocks(
-        &self,
-        blocks: &[(&Block, bool)],
-        head_tip: Option<&BlockMeta>,
-        state: &V03State,
-        final_snapshot: Option<(&V03State, &BlockMeta)>,
-    ) -> DbResult<()> {
+    /// The head state and tip meta are only rewritten when the chain actually
+    /// moved. A checkpoint alone (the common case — every follow event carries
+    /// one, most carry nothing else) must not drag a full state serialization
+    /// with it.
+    pub fn store_update(&self, update: &StoreUpdate<'_>) -> DbResult<StoreUpdateOutcome> {
+        let StoreUpdate {
+            checkpoint,
+            blocks,
+            head_tip,
+            head_state,
+            final_snapshot,
+            finalized_up_to,
+            new_deposit_events,
+            remove_deposit_records,
+            consumed_withdrawals,
+            new_withdraw_intents,
+            zone_anchor,
+        } = *update;
+
         let last_block_in_db = self.get_meta_last_block_in_db()?;
         let mut batch = WriteBatch::default();
 
+        if let Some(bytes) = checkpoint {
+            self.put_batch(&ZoneSdkCheckpointCellRef(bytes), (), &mut batch)?;
+        }
+        if let Some(anchor) = zone_anchor {
+            self.put_batch(&ZoneAnchorCell(*anchor), (), &mut batch)?;
+        }
+
+        // Every block payload this update writes, keyed by id so a block that
+        // is both explicitly written and swept by `finalized_up_to` is written
+        // once, with the caller's version.
+        let mut to_write: BTreeMap<u64, Block> = BTreeMap::new();
+
+        // Whether the stored chain moved, and with it the head state. A
+        // shrink-only update (orphans without adopted replacements) writes no
+        // payloads but still rewinds the tip, or the stored state tears
+        // against the stale disk head on the next produce.
+        let mut chain_changed =
+            final_snapshot.is_some() || head_tip.is_some_and(|tip| tip.id != last_block_in_db);
+
         for (block, finalized) in blocks {
             let already_stored = self
                 .get_block(block.header.block_id)?
                 .filter(|stored| stored.header.hash == block.header.hash);
 
-            let mut to_write = match already_stored {
+            let mut block_to_write = match already_stored {
                 Some(_) if !finalized => continue,
                 Some(stored) => stored,
                 None => (*block).clone(),
             };
             if *finalized {
-                to_write.bedrock_status = BedrockStatus::Finalized;
+                block_to_write.bedrock_status = BedrockStatus::Finalized;
             }
-            self.put_block_payload(&to_write, &mut batch)?;
+            to_write.insert(block_to_write.header.block_id, block_to_write);
+            chain_changed = true;
         }
 
+        if let Some(last_finalized) = finalized_up_to {
+            self.collect_finalized_up_to(last_finalized, &mut to_write);
+        }
+        for block in to_write.values() {
+            self.put_block_payload(block, &mut batch)?;
+        }
+
+        let accepted_deposits = self.stage_pending_deposit_events(
+            new_deposit_events,
+            remove_deposit_records,
+            &mut batch,
+        )?;
+        let unmatched_withdrawals =
+            self.stage_consumed_withdrawals(consumed_withdrawals, &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
-        // implies nothing was applied — and the store, created with genesis,
-        // cannot represent it. No tip to pin, nothing to persist.
-        let Some(tip) = head_tip else {
-            debug_assert!(batch.is_empty() && final_snapshot.is_none());
-            return Ok(());
+        // the store — created with genesis — cannot represent. Nothing to pin.
+        if chain_changed && let Some(tip) = head_tip {
+            // `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)?;
+            self.put_meta_latest_block_meta_batch(tip, &mut batch)?;
+            self.put_lee_state_in_db_batch(head_state, &mut batch)?;
+            if let Some((final_state, final_meta)) = final_snapshot {
+                self.put_final_snapshot_batch(final_state, final_meta, &mut batch)?;
+            }
+        }
+
+        let outcome = StoreUpdateOutcome {
+            accepted_deposits,
+            unmatched_withdrawals,
         };
 
-        // A shrink-only update (orphans without adopted replacements) has no
-        // payloads to write but must still rewind the tip meta, or the stored
-        // state tears against the stale disk head on the next produce.
-        if batch.is_empty() && final_snapshot.is_none() && tip.id == last_block_in_db {
-            return Ok(());
-        }
-
-        for stale_id in tip.id.saturating_add(1)..=last_block_in_db {
-            self.delete_block_payload(stale_id, &mut batch)?;
-        }
-        self.put_meta_last_block_in_db_batch(tip.id, &mut batch)?;
-        self.put_meta_latest_block_meta_batch(tip, &mut batch)?;
-        self.put_lee_state_in_db_batch(state, &mut batch)?;
-        if let Some((final_state, final_meta)) = final_snapshot {
-            self.put_final_snapshot_batch(final_state, final_meta, &mut batch)?;
+        if batch.is_empty() {
+            return Ok(outcome);
         }
 
         self.db.write(batch).map_err(|rerr| {
-            DbError::rocksdb_cast_message(
-                rerr,
-                Some("Failed to write followed blocks batch".to_owned()),
-            )
-        })
+            DbError::rocksdb_cast_message(rerr, Some("Failed to write store update".to_owned()))
+        })?;
+        Ok(outcome)
     }
 
     pub fn get_all_blocks(&self) -> impl Iterator<Item = DbResult<Block>> {
@@ -803,32 +1000,30 @@ impl RocksDBIO {
             })
     }
 
+    /// Persists a block we produced, its withdraw intents, the resulting state
+    /// and the publish `checkpoint` in one atomic write.
+    ///
+    /// The produce path is [`Self::store_update`] with a single block that is
+    /// the new tip; the checkpoint belongs in the same write for the same
+    /// reason it does there — it carries the sdk's `pending_txs`, so a
+    /// checkpoint persisted without this block would restore a pending set
+    /// that no longer contains the inscription we just published, and the sdk
+    /// would never resubmit it.
     pub fn atomic_update(
         &self,
         block: &Block,
-        deposit_op_ids: &[HashType],
-        withdrawals: Vec<WithdrawalReconciliationKey>,
+        withdrawals: &[WithdrawalReconciliationKey],
         state: &V03State,
+        checkpoint: Option<&[u8]>,
     ) -> DbResult<()> {
-        let block_id = block.header.block_id;
-        let mut batch = WriteBatch::default();
-
-        self.put_block(block, false, &mut batch)?;
-
-        self.mark_pending_deposit_events_submitted(deposit_op_ids, block_id, &mut batch)?;
-
-        for withdrawal in withdrawals {
-            self.increment_unseen_withdraw_count(withdrawal, &mut batch)?;
-        }
-
-        self.put_lee_state_in_db_batch(state, &mut batch)?;
-
-        self.db.write(batch).map_err(|rerr| {
-            DbError::rocksdb_cast_message(
-                rerr,
-                Some(format!("Failed to udpate db with block {block_id}")),
-            )
+        self.store_update(&StoreUpdate {
+            checkpoint,
+            blocks: &[(block, false)],
+            head_tip: Some(&BlockMeta::from(block)),
+            new_withdraw_intents: withdrawals,
+            ..StoreUpdate::new(state)
         })
+        .map(|_outcome| ())
     }
 }
 

lez/storage/src/sequencer/sequencer_cells.rs

@@ -232,14 +232,17 @@ impl SimpleWritableCell for ZoneAnchorCell {
     }
 }
 
+/// An L1 deposit event observed but not yet seen finalized.
+///
+/// Purely a liveness queue: whether to actually emit a mint is decided against
+/// chain state (the deposit-receipt PDA), and the record is dropped once its
+/// mint finalizes.
 #[derive(Debug, Clone, PartialEq, Eq, BorshSerialize, BorshDeserialize)]
 pub struct PendingDepositEventRecord {
     pub deposit_op_id: HashType,
     pub source_tx_hash: HashType,
     pub amount: u64,
     pub metadata: Vec<u8>,
-    /// Set when block containing the deposit event is submitted, but not necessarily finalized.
-    pub submitted_in_block_id: Option<u64>,
 }
 
 #[derive(BorshDeserialize)]
@@ -275,7 +278,7 @@ impl SimpleWritableCell for PendingDepositEventsCellRef<'_> {
     }
 }
 
-#[derive(Debug, Clone, Copy)]
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
 pub struct WithdrawalReconciliationKey {
     pub amount: u64,
     pub bedrock_account_pk: [u8; 32],

lez/storage/src/sequencer/tests.rs

@@ -28,6 +28,15 @@ fn dbio_with_genesis(path: &Path) -> (RocksDBIO, Block) {
     (dbio, genesis)
 }
 
+fn deposit_record(seed: u8) -> PendingDepositEventRecord {
+    PendingDepositEventRecord {
+        deposit_op_id: HashType([seed; 32]),
+        source_tx_hash: HashType([seed; 32]),
+        amount: u64::from(seed),
+        metadata: vec![seed],
+    }
+}
+
 fn stored_balance(dbio: &RocksDBIO) -> u128 {
     dbio.get_lee_state()
         .unwrap()
@@ -106,12 +115,11 @@ fn store_followed_blocks_batch_lands_meta_and_state_on_last_block() {
         id: 3,
         hash: block3.header.hash,
     };
-    dbio.store_followed_blocks(
-        &[(&block2, true), (&block3, false)],
-        Some(&head_tip),
-        &state_with_balance(300),
-        None,
-    )
+    dbio.store_update(&StoreUpdate {
+        blocks: &[(&block2, true), (&block3, false)],
+        head_tip: Some(&head_tip),
+        ..StoreUpdate::new(&state_with_balance(300))
+    })
     .unwrap();
 
     let stored2 = dbio.get_block(2).unwrap().expect("block 2 is stored");
@@ -140,12 +148,12 @@ fn final_snapshot_round_trips_and_is_absent_on_fresh_store() {
         id: 2,
         hash: block2.header.hash,
     };
-    dbio.store_followed_blocks(
-        &[(&block2, true)],
-        Some(&final_meta),
-        &state_with_balance(300),
-        Some((&state_with_balance(200), &final_meta)),
-    )
+    dbio.store_update(&StoreUpdate {
+        blocks: &[(&block2, true)],
+        head_tip: Some(&final_meta),
+        final_snapshot: Some((&state_with_balance(200), &final_meta)),
+        ..StoreUpdate::new(&state_with_balance(300))
+    })
     .unwrap();
 
     let (final_state, meta) = dbio
@@ -204,12 +212,11 @@ fn net_shortening_reorg_drops_stale_blocks() {
         id: 2,
         hash: block2b.header.hash,
     };
-    dbio.store_followed_blocks(
-        &[(&block2b, false)],
-        Some(&head_tip),
-        &state_with_balance(400),
-        None,
-    )
+    dbio.store_update(&StoreUpdate {
+        blocks: &[(&block2b, false)],
+        head_tip: Some(&head_tip),
+        ..StoreUpdate::new(&state_with_balance(400))
+    })
     .unwrap();
 
     let stored2 = dbio.get_block(2).unwrap().expect("block 2 is stored");
@@ -241,8 +248,11 @@ fn shrink_only_reorg_rewinds_tip_meta() {
         id: 2,
         hash: block2.header.hash,
     };
-    dbio.store_followed_blocks(&[], Some(&head_tip), &state_with_balance(200), None)
-        .unwrap();
+    dbio.store_update(&StoreUpdate {
+        head_tip: Some(&head_tip),
+        ..StoreUpdate::new(&state_with_balance(200))
+    })
+    .unwrap();
 
     assert!(
         dbio.get_block(3).unwrap().is_none(),
@@ -254,6 +264,232 @@ fn shrink_only_reorg_rewinds_tip_meta() {
     assert_eq!(stored_balance(&dbio), 200);
 }
 
+#[test]
+fn checkpoint_lands_with_an_orphan_only_update() {
+    let temp_dir = tempdir().unwrap();
+    let (dbio, genesis) = dbio_with_genesis(temp_dir.path());
+
+    let block2 = produce_dummy_block(2, Some(genesis.header.hash), vec![]);
+    dbio.store_followed_block(&block2, &state_with_balance(200), false)
+        .unwrap();
+    let block3 = produce_dummy_block(3, Some(block2.header.hash), vec![]);
+    dbio.store_followed_block(&block3, &state_with_balance(300), false)
+        .unwrap();
+
+    // Orphan-only update: no payload to write, but the checkpoint covering it
+    // must still land, or a restart resumes past the orphan.
+    let head_tip = BlockMeta {
+        id: 2,
+        hash: block2.header.hash,
+    };
+    dbio.store_update(&StoreUpdate {
+        checkpoint: Some(b"cp-orphan"),
+        head_tip: Some(&head_tip),
+        ..StoreUpdate::new(&state_with_balance(200))
+    })
+    .unwrap();
+
+    assert_eq!(
+        dbio.get_zone_sdk_checkpoint_bytes().unwrap().as_deref(),
+        Some(b"cp-orphan".as_slice())
+    );
+    assert!(dbio.get_block(3).unwrap().is_none());
+}
+
+#[test]
+fn checkpoint_only_update_does_not_rewrite_the_head_state() {
+    let temp_dir = tempdir().unwrap();
+    let (dbio, genesis) = dbio_with_genesis(temp_dir.path());
+
+    let block2 = produce_dummy_block(2, Some(genesis.header.hash), vec![]);
+    dbio.store_followed_block(&block2, &state_with_balance(200), false)
+        .unwrap();
+
+    // An event carrying nothing but a checkpoint (the common case) must not
+    // drag a full state serialization along with it — the caller's state is
+    // ignored while the chain stands still.
+    let head_tip = BlockMeta::from(&block2);
+    dbio.store_update(&StoreUpdate {
+        checkpoint: Some(b"cp-idle"),
+        head_tip: Some(&head_tip),
+        ..StoreUpdate::new(&state_with_balance(999))
+    })
+    .unwrap();
+
+    assert_eq!(
+        dbio.get_zone_sdk_checkpoint_bytes().unwrap().as_deref(),
+        Some(b"cp-idle".as_slice())
+    );
+    assert_eq!(stored_balance(&dbio), 200);
+}
+
+#[test]
+fn several_deposits_in_one_update_are_all_recorded() {
+    let temp_dir = tempdir().unwrap();
+    let (dbio, _genesis) = dbio_with_genesis(temp_dir.path());
+
+    // The records live in one whole-vector cell: staged per event against a
+    // fresh disk read, the second would clobber the first.
+    let first = deposit_record(1);
+    let second = deposit_record(2);
+    let already_known = dbio.get_pending_deposit_events().unwrap();
+    assert!(already_known.is_empty());
+
+    let outcome = dbio
+        .store_update(&StoreUpdate {
+            new_deposit_events: &[first.clone(), second.clone()],
+            ..StoreUpdate::new(&state_with_balance(100))
+        })
+        .unwrap();
+
+    assert_eq!(outcome.accepted_deposits, 2);
+    let stored = dbio.get_pending_deposit_events().unwrap();
+    assert_eq!(stored.len(), 2);
+    assert!(stored.contains(&first));
+    assert!(stored.contains(&second));
+}
+
+#[test]
+fn redelivered_deposit_is_not_accepted_twice() {
+    let temp_dir = tempdir().unwrap();
+    let (dbio, _genesis) = dbio_with_genesis(temp_dir.path());
+
+    let record = deposit_record(1);
+    dbio.store_update(&StoreUpdate {
+        new_deposit_events: std::slice::from_ref(&record),
+        ..StoreUpdate::new(&state_with_balance(100))
+    })
+    .unwrap();
+
+    let outcome = dbio
+        .store_update(&StoreUpdate {
+            new_deposit_events: &[record],
+            ..StoreUpdate::new(&state_with_balance(100))
+        })
+        .unwrap();
+
+    assert_eq!(
+        outcome.accepted_deposits, 0,
+        "a re-delivered deposit is already owed, not newly accepted"
+    );
+    assert_eq!(dbio.get_pending_deposit_events().unwrap().len(), 1);
+}
+
+#[test]
+fn finalized_deposit_records_are_removed_by_op_id() {
+    let temp_dir = tempdir().unwrap();
+    let (dbio, _genesis) = dbio_with_genesis(temp_dir.path());
+
+    let first = deposit_record(1);
+    let second = deposit_record(2);
+    dbio.store_update(&StoreUpdate {
+        new_deposit_events: &[first.clone(), second.clone()],
+        ..StoreUpdate::new(&state_with_balance(100))
+    })
+    .unwrap();
+
+    // Only the finalized op id is dropped; the other record stays.
+    dbio.store_update(&StoreUpdate {
+        remove_deposit_records: &[first.deposit_op_id],
+        ..StoreUpdate::new(&state_with_balance(100))
+    })
+    .unwrap();
+
+    let stored = dbio.get_pending_deposit_events().unwrap();
+    assert_eq!(stored, vec![second]);
+}
+
+#[test]
+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());
+
+    let key = WithdrawalReconciliationKey {
+        amount: 7,
+        bedrock_account_pk: [3; 32],
+    };
+
+    // 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.
+    let outcome = dbio
+        .store_update(&StoreUpdate {
+            consumed_withdrawals: &[key, key],
+            ..StoreUpdate::new(&state_with_balance(100))
+        })
+        .unwrap();
+
+    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 — `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()
+        .map(|cell| cell.0);
+    assert_eq!(remaining, Some(0));
+}
+
+#[test]
+fn unmatched_withdrawal_is_reported_and_writes_nothing() {
+    let temp_dir = tempdir().unwrap();
+    let (dbio, _genesis) = dbio_with_genesis(temp_dir.path());
+
+    let key = WithdrawalReconciliationKey {
+        amount: 5,
+        bedrock_account_pk: [4; 32],
+    };
+    let outcome = dbio
+        .store_update(&StoreUpdate {
+            consumed_withdrawals: &[key],
+            ..StoreUpdate::new(&state_with_balance(100))
+        })
+        .unwrap();
+
+    assert_eq!(outcome.unmatched_withdrawals.len(), 1);
+    assert!(
+        dbio.get_opt::<UnseenWithdrawCountCell>(key)
+            .unwrap()
+            .is_none(),
+        "an unmatched withdraw must not leave a counter behind"
+    );
+}
+
+#[test]
+fn produced_block_persists_its_publish_checkpoint() {
+    let temp_dir = tempdir().unwrap();
+    let (dbio, genesis) = dbio_with_genesis(temp_dir.path());
+
+    let block2 = produce_dummy_block(2, Some(genesis.header.hash), vec![]);
+    dbio.atomic_update(&block2, &[], &state_with_balance(200), Some(b"cp-produced"))
+        .unwrap();
+
+    // Storing the block without the checkpoint would let a restart restore a
+    // pending set that no longer holds the inscription we just published.
+    assert_eq!(
+        dbio.get_zone_sdk_checkpoint_bytes().unwrap().as_deref(),
+        Some(b"cp-produced".as_slice())
+    );
+    assert_eq!(
+        dbio.get_block(2).unwrap().unwrap().header.hash,
+        block2.header.hash
+    );
+}
+
 #[test]
 fn produced_block_below_disk_head_pins_meta_and_prunes() {
     let temp_dir = tempdir().unwrap();
@@ -270,7 +506,7 @@ fn produced_block_below_disk_head_pins_meta_and_prunes() {
     // pins the tip meta to the produced block and drops the stale suffix in
     // the same write, mirroring the follow path.
     let block2b = produce_dummy_block(2, Some(genesis.header.hash), vec![]);
-    dbio.atomic_update(&block2b, &[], vec![], &state_with_balance(400))
+    dbio.atomic_update(&block2b, &[], &state_with_balance(400), None)
         .unwrap();
 
     let stored2 = dbio.get_block(2).unwrap().expect("block 2 is stored");