feat: add bedrock withdraw events validation

integration_tests/tests/bridge.rs

@@ -490,6 +490,9 @@ async fn bedrock_deposit_claim_and_withdraw_round_trip_succeeds() -> anyhow::Res
     observe_result
         .context("Failed while waiting for finalized withdraw event from zone indexer")?;
 
+    // Sleep to observe sequencer log about validated withdraw event
+    tokio::time::sleep(Duration::from_secs(1)).await;
+
     Ok(())
 }
 

lez/sequencer/core/src/block_publisher.rs

@@ -3,19 +3,21 @@ use std::{pin::Pin, sync::Arc, time::Duration};
 use anyhow::{Context as _, Result};
 use common::block::Block;
 use log::{info, warn};
-use logos_blockchain_core::mantle::{Note, ledger::Outputs, ops::channel::inscribe::Inscription};
+use logos_blockchain_core::mantle::ops::channel::inscribe::Inscription;
 pub use logos_blockchain_key_management_system_service::keys::{Ed25519Key, ZkKey};
 pub use logos_blockchain_zone_sdk::sequencer::SequencerCheckpoint;
 use logos_blockchain_zone_sdk::{
     CommonHttpClient,
     adapter::NodeHttpClient,
-    sequencer::{Event, SequencerConfig as ZoneSdkSequencerConfig, SequencerHandle, ZoneSequencer},
+    sequencer::{
-    state::{DepositInfo, FinalizedOp, InscriptionInfo},
+        Event, SequencerConfig as ZoneSdkSequencerConfig, SequencerHandle, WithdrawArg,
+        ZoneSequencer,
+    },
+    state::{DepositInfo, FinalizedOp, InscriptionInfo, WithdrawInfo},
 };
-use num_bigint::BigUint;
 use tokio::task::JoinHandle;
 
-use crate::{BridgeWithdrawData, config::BedrockConfig};
+use crate::config::BedrockConfig;
 
 /// Sink for `Event::Published` checkpoints emitted by the drive task.
 /// Caller is responsible for persistence (e.g. writing to rocksdb).
@@ -30,8 +32,16 @@ pub type FinalizedBlockSink = Box<dyn Fn(u64) + Send + 'static>;
 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>;
+
 #[expect(async_fn_in_trait, reason = "We don't care about Send/Sync here")]
 pub trait BlockPublisherTrait: Clone {
+    #[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,
@@ -40,15 +50,12 @@ pub trait BlockPublisherTrait: Clone {
         on_checkpoint: CheckpointSink,
         on_finalized_block: FinalizedBlockSink,
         on_deposit_event: OnDepositEventSink,
+        on_withdraw_event: OnWithdrawEventSink,
     ) -> Result<Self>;
 
     /// Fire-and-forget publish. Zone-sdk drives the actual submission and
     /// retries internally; this just hands the payload off.
-    async fn publish_block(
+    async fn publish_block(&self, block: &Block, withdraws: Vec<WithdrawArg>) -> Result<()>;
-        &self,
-        block: &Block,
-        bridge_withdrawals: Vec<BridgeWithdrawData>,
-    ) -> Result<()>;
 }
 
 /// Real block publisher backed by zone-sdk's `ZoneSequencer`.
@@ -76,6 +83,7 @@ impl BlockPublisherTrait for ZoneSdkPublisher {
         on_checkpoint: CheckpointSink,
         on_finalized_block: FinalizedBlockSink,
         on_deposit_event: OnDepositEventSink,
+        on_withdraw_event: OnWithdrawEventSink,
     ) -> Result<Self> {
         let basic_auth = config.auth.clone().map(Into::into);
         let node = NodeHttpClient::new(CommonHttpClient::new(basic_auth), config.node_url.clone());
@@ -112,7 +120,9 @@ impl BlockPublisherTrait for ZoneSdkPublisher {
                                 FinalizedOp::Deposit(deposit) => {
                                     on_deposit_event(deposit).await;
                                 }
-                                FinalizedOp::Withdraw(_) => {}
+                                FinalizedOp::Withdraw(withdraw) => {
+                                    on_withdraw_event(withdraw).await;
+                                }
                             }
                         }
                     }
@@ -132,18 +142,14 @@ impl BlockPublisherTrait for ZoneSdkPublisher {
         })
     }
 
-    async fn publish_block(
+    async fn publish_block(&self, block: &Block, withdraws: Vec<WithdrawArg>) -> Result<()> {
-        &self,
-        block: &Block,
-        bridge_withdrawals: Vec<BridgeWithdrawData>,
-    ) -> Result<()> {
         let data = borsh::to_vec(block).context("Failed to serialize block")?;
         let data_bounded: Inscription = data
             .try_into()
             .context("Block data exceeds maximum allowed size")?;
         let data_byte_size = data_bounded.len();
 
-        if bridge_withdrawals.is_empty() {
+        if withdraws.is_empty() {
             self.handle
                 .publish_message(data_bounded)
                 .await
@@ -154,20 +160,6 @@ impl BlockPublisherTrait for ZoneSdkPublisher {
             return Ok(());
         }
 
-        let withdraws: Vec<_> = bridge_withdrawals
-            .into_iter()
-            .map(|withdrawal| {
-                let recipient_pk =
-                    logos_blockchain_key_management_system_service::keys::ZkPublicKey::from(
-                        BigUint::from_bytes_le(&withdrawal.bedrock_account_pk),
-                    );
-
-                logos_blockchain_zone_sdk::sequencer::WithdrawArg {
-                    outputs: Outputs::new(Note::new(withdrawal.amount, recipient_pk)),
-                }
-            })
-            .collect();
-
         let withdraw_count = withdraws.len();
         self.handle
             .publish_atomic_withdraw(data_bounded, withdraws)

lez/sequencer/core/src/block_store.rs

@@ -186,6 +186,24 @@ impl SequencerStore {
         self.dbio
             .remove_fulfilled_pending_deposit_events_up_to_block(finalized_block_id)
     }
+
+    pub fn record_unseen_withdraw(
+        &self,
+        amount: u64,
+        bedrock_account_pk: [u8; 32],
+    ) -> DbResult<u64> {
+        self.dbio
+            .increment_unseen_withdraw_count(amount, bedrock_account_pk)
+    }
+
+    pub fn consume_unseen_withdraw(
+        &self,
+        amount: u64,
+        bedrock_account_pk: [u8; 32],
+    ) -> DbResult<bool> {
+        self.dbio
+            .consume_unseen_withdraw_count(amount, bedrock_account_pk)
+    }
 }
 
 pub(crate) fn block_to_transactions_map(block: &Block) -> HashMap<HashType, u64> {

lez/sequencer/core/src/lib.rs

@@ -12,11 +12,13 @@ use lee::{AccountId, PublicTransaction, program::Program, public_transaction::Me
 use lee_core::GENESIS_BLOCK_ID;
 use log::{error, info, warn};
 use logos_blockchain_key_management_system_service::keys::{ED25519_SECRET_KEY_SIZE, Ed25519Key};
+use logos_blockchain_zone_sdk::sequencer::WithdrawArg;
 use mempool::{MemPool, MemPoolHandle};
 #[cfg(feature = "mock")]
 pub use mock::SequencerCoreWithMockClients;
+use num_bigint::BigUint;
 pub use storage::error::DbError;
-use storage::sequencer::sequencer_cells::PendingDepositEventRecord;
+use storage::sequencer::{RocksDBIO, sequencer_cells::PendingDepositEventRecord};
 
 use crate::{
     block_publisher::{BlockPublisherTrait, ZoneSdkPublisher},
@@ -132,112 +134,18 @@ impl<BP: BlockPublisherTrait> SequencerCore<BP> {
             .expect("Failed to load zone-sdk checkpoint");
         let is_fresh_start = initial_checkpoint.is_none();
 
-        let dbio_for_checkpoint = store.dbio();
-        let on_checkpoint: 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_for_checkpoint.put_zone_sdk_checkpoint_bytes(&bytes) {
-                error!("Failed to persist zone-sdk checkpoint: {err:#}");
-            }
-        });
-
-        let dbio_for_finalized = store.dbio();
-        let on_finalized_block: 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_for_finalized.clean_pending_blocks_up_to(block_id) {
-                error!("Failed to mark pending blocks finalized up to {block_id}: {err:#}");
-            }
-
-            match dbio_for_finalized.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:#}"
-                    );
-                }
-            }
-        });
-
         let (mempool, mempool_handle) = MemPool::new(config.mempool_max_size);
-
         replay_unfulfilled_deposit_events(&store, mempool_handle.clone());
 
-        let mempool_handle_for_deposit = mempool_handle.clone();
-        let dbio_for_deposit = store.dbio();
-        let on_deposit_event: 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_for_deposit = Arc::clone(&dbio_for_deposit);
-            let mempool_handle_for_deposit = mempool_handle_for_deposit.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_for_deposit.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_for_deposit
-                    .push((TransactionOrigin::Sequencer, tx))
-                    .await
-                {
-                    error!(
-                        "Failed to queue sequencer transaction built from finalized Bedrock event: {err:#}. Deposit will be lost."
-                    );
-                }
-            })
-        });
-
         let block_publisher = BP::new(
             &config.bedrock_config,
             bedrock_signing_key,
             config.retry_pending_blocks_timeout,
             initial_checkpoint,
-            on_checkpoint,
+            Self::on_checkpoint(store.dbio()),
-            on_finalized_block,
+            Self::on_finalized_block(store.dbio()),
-            on_deposit_event,
+            Self::on_deposit_event(store.dbio(), mempool_handle.clone()),
+            Self::on_withdraw_event(store.dbio()),
         )
         .await
         .expect("Failed to initialize Block Publisher");
@@ -265,22 +173,164 @@ impl<BP: BlockPublisherTrait> SequencerCore<BP> {
         (sequencer_core, mempool_handle)
     }
 
+    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.amount,
+                    withdraw_key.bedrock_account_pk,
+                ) {
+                    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:#}"
+                    ),
+                }
+            })
+        })
+    }
+
     /// 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,
-            bridge_withdrawals,
+            withdraws,
         } = self
             .build_block_from_mempool()
             .context("Failed to build block from mempool transactions")?;
 
+        for withdraw in &withdraws {
+            let withdraw_key = withdraw_event_reconciliation_key(&withdraw.outputs)
+                .context("Failed to derive unseen-withdraw key from withdraw data")?;
+
+            self.store
+                .record_unseen_withdraw(withdraw_key.amount, withdraw_key.bedrock_account_pk)
+                .context("Failed to persist unseen withdraw for reconciliation")?;
+        }
+
         // TODO: Remove msg_id from store.update — it is no longer needed now that
         // zone-sdk manages L1 settlement state via its own checkpoint.
         let placeholder_msg_id = [0_u8; 32];
 
         self.block_publisher
-            .publish_block(&block, bridge_withdrawals)
+            .publish_block(&block, withdraws)
             .await
             .context("Failed to publish block to Bedrock")?;
 
@@ -308,7 +358,7 @@ impl<BP: BlockPublisherTrait> SequencerCore<BP> {
 
         let mut valid_transactions = Vec::new();
         let mut deposit_event_ids = Vec::new();
-        let mut bridge_withdrawals = Vec::new();
+        let mut withdraws = Vec::new();
 
         let max_block_size = usize::try_from(self.sequencer_config.max_block_size.as_u64())
             .expect("`max_block_size` should fit into usize");
@@ -374,7 +424,7 @@ impl<BP: BlockPublisherTrait> SequencerCore<BP> {
                     };
 
                     if let Some(withdraw_data) = extract_bridge_withdraw_data(&tx) {
-                        bridge_withdrawals.push(withdraw_data);
+                        withdraws.push(withdraw_data);
                     }
 
                     self.state.apply_state_diff(validated_diff);
@@ -437,7 +487,7 @@ impl<BP: BlockPublisherTrait> SequencerCore<BP> {
         Ok(BlockWithMeta {
             block,
             deposit_event_ids,
-            bridge_withdrawals,
+            withdraws,
         })
     }
 
@@ -497,13 +547,13 @@ impl<BP: BlockPublisherTrait> SequencerCore<BP> {
 struct BlockWithMeta {
     block: Block,
     deposit_event_ids: Vec<HashType>,
-    bridge_withdrawals: Vec<BridgeWithdrawData>,
+    withdraws: Vec<WithdrawArg>,
 }
 
-#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+#[derive(Debug, Clone, Copy)]
-pub struct BridgeWithdrawData {
+struct WithdrawReconciliationKey {
-    pub amount: u64,
+    amount: u64,
-    pub bedrock_account_pk: [u8; 32],
+    bedrock_account_pk: [u8; 32],
 }
 
 /// Checks the database for any pending deposit events that have not yet been marked as submitted in
@@ -694,7 +744,7 @@ fn extract_bridge_deposit_id(tx: &LeeTransaction) -> Option<HashType> {
 }
 
 #[must_use]
-fn extract_bridge_withdraw_data(tx: &LeeTransaction) -> Option<BridgeWithdrawData> {
+fn extract_bridge_withdraw_data(tx: &LeeTransaction) -> Option<WithdrawArg> {
     let LeeTransaction::Public(tx) = tx else {
         return None;
     };
@@ -712,16 +762,55 @@ fn extract_bridge_withdraw_data(tx: &LeeTransaction) -> Option<BridgeWithdrawDat
         bridge_core::Instruction::Withdraw {
             amount,
             bedrock_account_pk,
-        } => Some(BridgeWithdrawData {
+        } => {
-            amount,
+            let recipient_pk =
-            bedrock_account_pk,
+                logos_blockchain_key_management_system_service::keys::ZkPublicKey::from(
-        }),
+                    BigUint::from_bytes_le(&bedrock_account_pk),
+                );
+
+            Some(WithdrawArg {
+                outputs: logos_blockchain_core::mantle::ledger::Outputs::new(
+                    logos_blockchain_core::mantle::Note::new(amount, recipient_pk),
+                ),
+            })
+        }
         bridge_core::Instruction::Deposit { .. } => unreachable!(
             "Deposit instructions from users should never pass validation, and thus should never be seen here"
         ),
     }
 }
 
+fn withdraw_event_reconciliation_key(
+    outputs: &logos_blockchain_core::mantle::ledger::Outputs,
+) -> Result<WithdrawReconciliationKey> {
+    let [note] = outputs.as_ref().as_slice() else {
+        return Err(anyhow!(
+            "Unsupported withdraw output count for reconciliation: {}",
+            outputs.len()
+        ));
+    };
+
+    // `extract_bridge_withdraw_data` maps [u8;32] LE -> BigUint -> ZkPublicKey.
+    // Reconcile by reversing that direction here.
+    let mut bedrock_account_pk = BigUint::from(note.pk.into_inner()).to_bytes_le();
+    if bedrock_account_pk.len() > 32 {
+        return Err(anyhow!(
+            "Withdraw recipient public key is too large: {} bytes",
+            bedrock_account_pk.len()
+        ));
+    }
+    bedrock_account_pk.resize(32, 0);
+
+    let bedrock_account_pk: [u8; 32] = bedrock_account_pk
+        .try_into()
+        .expect("Public key bytes were padded/truncated to 32 bytes");
+
+    Ok(WithdrawReconciliationKey {
+        amount: note.value,
+        bedrock_account_pk,
+    })
+}
+
 /// Load signing key from file or generate a new one if it doesn't exist.
 fn load_or_create_signing_key(path: &Path) -> Result<Ed25519Key> {
     if path.exists() {

lez/sequencer/core/src/mock.rs

@@ -3,12 +3,12 @@ use std::time::Duration;
 use anyhow::Result;
 use common::block::Block;
 use logos_blockchain_key_management_system_service::keys::Ed25519Key;
+use logos_blockchain_zone_sdk::sequencer::WithdrawArg;
 
 use crate::{
-    BridgeWithdrawData,
     block_publisher::{
         BlockPublisherTrait, CheckpointSink, FinalizedBlockSink, OnDepositEventSink,
-        SequencerCheckpoint,
+        OnWithdrawEventSink, SequencerCheckpoint,
     },
     config::BedrockConfig,
 };
@@ -27,6 +27,7 @@ impl BlockPublisherTrait for MockBlockPublisher {
         _on_checkpoint: CheckpointSink,
         _on_finalized_block: FinalizedBlockSink,
         _on_deposit_event: OnDepositEventSink,
+        _on_withdraw_event: OnWithdrawEventSink,
     ) -> Result<Self> {
         Ok(Self)
     }
@@ -34,7 +35,7 @@ impl BlockPublisherTrait for MockBlockPublisher {
     async fn publish_block(
         &self,
         _block: &Block,
-        _bridge_withdrawals: Vec<BridgeWithdrawData>,
+        _bridge_withdrawals: Vec<WithdrawArg>,
     ) -> Result<()> {
         Ok(())
     }

lez/storage/src/sequencer/mod.rs

@@ -11,12 +11,16 @@ use rocksdb::{
 
 use crate::{
     CF_BLOCK_NAME, CF_META_NAME, DB_META_FIRST_BLOCK_IN_DB_KEY, DBIO, DbResult,
-    cells::shared_cells::{BlockCell, FirstBlockCell, FirstBlockSetCell, LastBlockCell},
+    cells::{
+        SimpleStorableCell,
+        shared_cells::{BlockCell, FirstBlockCell, FirstBlockSetCell, LastBlockCell},
+    },
     error::DbError,
     sequencer::sequencer_cells::{
         LEEStateCellOwned, LEEStateCellRef, LastFinalizedBlockIdCell, LatestBlockMetaCellOwned,
         LatestBlockMetaCellRef, PendingDepositEventRecord, PendingDepositEventsCellOwned,
-        PendingDepositEventsCellRef, ZoneSdkCheckpointCellOwned, ZoneSdkCheckpointCellRef,
+        PendingDepositEventsCellRef, UnseenWithdrawCountCell, ZoneSdkCheckpointCellOwned,
+        ZoneSdkCheckpointCellRef,
     },
 };
 
@@ -31,6 +35,8 @@ pub const DB_META_ZONE_SDK_CHECKPOINT_KEY: &str = "zone_sdk_checkpoint";
 /// Key base for storing queued deposit events that were not yet
 /// fulfilled on L2.
 pub const DB_META_PENDING_DEPOSIT_EVENTS_KEY: &str = "pending_deposit_events";
+/// Key base for counting unseen L2 withdraw intents.
+pub const DB_META_UNSEEN_WITHDRAW_COUNT_KEY: &str = "unseen_withdraw_count";
 
 /// Key base for storing the LEE state.
 pub const DB_LEE_STATE_KEY: &str = "lee_state";
@@ -312,6 +318,58 @@ impl RocksDBIO {
         Ok(removed)
     }
 
+    pub fn increment_unseen_withdraw_count(
+        &self,
+        amount: u64,
+        bedrock_account_pk: [u8; 32],
+    ) -> DbResult<u64> {
+        let key_params = (amount, bedrock_account_pk);
+
+        let current = self
+            .get_opt::<UnseenWithdrawCountCell>(key_params)?
+            .map_or(0, |cell| cell.0);
+
+        let next = current.checked_add(1).ok_or_else(|| {
+            DbError::db_interaction_error("Unseen withdraw counter overflow".to_owned())
+        })?;
+
+        self.put(&UnseenWithdrawCountCell(next), key_params)?;
+
+        Ok(next)
+    }
+
+    pub fn consume_unseen_withdraw_count(
+        &self,
+        amount: u64,
+        bedrock_account_pk: [u8; 32],
+    ) -> DbResult<bool> {
+        let key_params = (amount, bedrock_account_pk);
+
+        let Some(current) = self
+            .get_opt::<UnseenWithdrawCountCell>(key_params)?
+            .map(|cell| cell.0)
+        else {
+            return Ok(false);
+        };
+
+        if let Some(next) = current.checked_sub(1) {
+            self.put(&UnseenWithdrawCountCell(next), key_params)?;
+        } else {
+            let cf_meta = self.meta_column();
+            let db_key =
+                <UnseenWithdrawCountCell as SimpleStorableCell>::key_constructor(key_params)?;
+
+            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)
+    }
+
     pub fn put_block(
         &self,
         block: &Block,

lez/storage/src/sequencer/sequencer_cells.rs

@@ -9,7 +9,7 @@ use crate::{
     sequencer::{
         CF_LEE_STATE_NAME, DB_LEE_STATE_KEY, DB_META_LAST_FINALIZED_BLOCK_ID,
         DB_META_LATEST_BLOCK_META_KEY, DB_META_PENDING_DEPOSIT_EVENTS_KEY,
-        DB_META_ZONE_SDK_CHECKPOINT_KEY,
+        DB_META_UNSEEN_WITHDRAW_COUNT_KEY, DB_META_ZONE_SDK_CHECKPOINT_KEY,
     },
 };
 
@@ -174,6 +174,40 @@ impl SimpleWritableCell for PendingDepositEventsCellRef<'_> {
         })
     }
 }
+#[derive(Debug, BorshSerialize, BorshDeserialize)]
+pub struct UnseenWithdrawCountCell(pub u64);
+
+impl SimpleStorableCell for UnseenWithdrawCountCell {
+    type KeyParams = (u64, [u8; 32]);
+
+    const CELL_NAME: &'static str = DB_META_UNSEEN_WITHDRAW_COUNT_KEY;
+    const CF_NAME: &'static str = CF_META_NAME;
+
+    fn key_constructor((amount, bedrock_account_pk): Self::KeyParams) -> DbResult<Vec<u8>> {
+        borsh::to_vec(&(Self::CELL_NAME, amount, bedrock_account_pk)).map_err(|err| {
+            DbError::borsh_cast_message(
+                err,
+                Some(format!(
+                    "Failed to serialize {:?} key params",
+                    Self::CELL_NAME
+                )),
+            )
+        })
+    }
+}
+
+impl SimpleReadableCell for UnseenWithdrawCountCell {}
+
+impl SimpleWritableCell for UnseenWithdrawCountCell {
+    fn value_constructor(&self) -> DbResult<Vec<u8>> {
+        borsh::to_vec(&self).map_err(|err| {
+            DbError::borsh_cast_message(
+                err,
+                Some("Failed to serialize unseen withdraw count".to_owned()),
+            )
+        })
+    }
+}
 
 #[cfg(test)]
 mod uniform_tests {