feat: collect withdrawal volumes metrics (#200)

* wip

* add method to query withdrawals

* adds metering of withdrawals values

* separates logic to new crate

* cleanup

* fix the query

* spellcheck

* fix review

* spelling

* use a component name in metering

Cargo.toml

@@ -13,4 +13,5 @@ members = [
     "tx-sender",
     "vlog",
     "watcher",
+    "withdrawals-meterer"
 ]

finalizer/Cargo.toml

@@ -18,3 +18,4 @@ client = { path = "../client" }
 storage = { path = "../storage" }
 vlog = { path = "../vlog" }
 tx-sender = { path = "../tx-sender" }
+withdrawals-meterer = { path = "../withdrawals-meterer" }

finalizer/src/lib.rs

@@ -24,6 +24,7 @@ use client::{
     l1bridge::codegen::IL1Bridge, withdrawal_finalizer::codegen::WithdrawalFinalizer,
     zksync_contract::codegen::IZkSync, WithdrawalParams, ZksyncMiddleware,
 };
+use withdrawals_meterer::WithdrawalsMeter;
 
 use crate::error::{Error, Result};
 
@@ -54,6 +55,7 @@ pub struct Finalizer<M1, M2> {
     tx_fee_limit: U256,
     tx_retry_timeout: Duration,
     account_address: Address,
+    withdrawals_meterer: WithdrawalsMeter,
 }
 
 const NO_NEW_WITHDRAWALS_BACKOFF: Duration = Duration::from_secs(5);
@@ -82,6 +84,10 @@ where
         tx_retry_timeout: usize,
         account_address: Address,
     ) -> Self {
+        let withdrawals_meterer = withdrawals_meterer::WithdrawalsMeter::new(
+            pgpool.clone(),
+            "era_withdrawal_finalizer_meter",
+        );
         let tx_fee_limit = ethers::utils::parse_ether(TX_FEE_LIMIT)
             .expect("{TX_FEE_LIMIT} ether is a parsable amount; qed");
 
@@ -98,6 +104,7 @@ where
             tx_fee_limit,
             tx_retry_timeout: Duration::from_secs(tx_retry_timeout as u64),
             account_address,
+            withdrawals_meterer,
         }
     }
 
@@ -183,6 +190,8 @@ where
         )
         .await;
 
+        let ids: Vec<_> = withdrawals.iter().map(|w| w.id as i64).collect();
+
         // Turn actual withdrawals into info to update db with.
         let withdrawals = withdrawals.into_iter().map(|w| w.key()).collect::<Vec<_>>();
 
@@ -204,6 +213,14 @@ where
                     "finalizer.highest_finalized_batch_number",
                     highest_batch_number.as_u64() as f64,
                 );
+
+                if let Err(e) = self
+                    .withdrawals_meterer
+                    .meter_withdrawals_storage(&ids)
+                    .await
+                {
+                    vlog::error!("Failed to meter the withdrawals: {e}");
+                }
             }
             // TODO: why would a pending tx resolve to `None`?
             Ok(None) => {

storage/src/lib.rs

@@ -244,6 +244,38 @@ pub async fn executed_new_batch(
     Ok(())
 }
 
+/// Gets withdrawal events from the db by a set of IDs.
+///
+/// # Arguments
+///
+/// * `conn`: Connection to the Postgres DB
+/// * `ids`: ID fields of the withdrawals to be returned.
+pub async fn get_withdrawals(pool: &PgPool, ids: &[i64]) -> Result<Vec<StoredWithdrawal>> {
+    let events = sqlx::query!(
+        "
+        SELECT * FROM
+            withdrawals
+        WHERE id in (SELECT id FROM unnest( $1 :: bigint[] ))
+        ",
+        ids
+    )
+    .fetch_all(pool)
+    .await?
+    .into_iter()
+    .map(|r| StoredWithdrawal {
+        event: WithdrawalEvent {
+            tx_hash: H256::from_slice(&r.tx_hash),
+            block_number: r.l2_block_number as u64,
+            token: Address::from_slice(&r.token),
+            amount: utils::bigdecimal_to_u256(r.amount),
+        },
+        index_in_tx: r.event_index_in_tx as usize,
+    })
+    .collect();
+
+    Ok(events)
+}
+
 /// Adds a withdrawal event to the DB.
 ///
 /// # Arguments
@@ -838,6 +870,26 @@ pub async fn inc_unsuccessful_finalization_attempts(
     Ok(())
 }
 
+/// Fetch decimals for a token.
+///
+/// # Arguments
+///
+/// * `pool` - `PgPool`
+/// * `token` - L2 token address.
+pub async fn token_decimals(pool: &PgPool, token: Address) -> Result<Option<u32>> {
+    let result = sqlx::query!(
+        "
+        SELECT decimals FROM tokens WHERE l2_token_address = $1
+        ",
+        token.as_bytes(),
+    )
+    .fetch_optional(pool)
+    .await?
+    .map(|r| r.decimals as u32);
+
+    Ok(result)
+}
+
 async fn wipe_finalization_data(pool: &PgPool, delete_batch_size: usize) -> Result<()> {
     loop {
         let deleted_ids = sqlx::query!(
@@ -942,8 +994,7 @@ async fn wipe_withdrawals(pool: &PgPool, delete_batch_size: usize) -> Result<()>
                   withdrawals
                 LIMIT
                   $1
-              )
-            RETURNING id
+              ) RETURNING id
             ",
             delete_batch_size as i64,
         )

storage/src/utils.rs

@@ -14,6 +14,18 @@ pub(crate) fn u256_to_big_decimal(value: U256) -> BigDecimal {
     ratio_to_big_decimal(&ratio, 80)
 }
 
+/// Converts `BigUint` value into the corresponding `U256` value.
+fn biguint_to_u256(value: BigUint) -> U256 {
+    let bytes = value.to_bytes_le();
+    U256::from_little_endian(&bytes)
+}
+
+/// Converts `BigDecimal` value into the corresponding `U256` value.
+pub(crate) fn bigdecimal_to_u256(value: BigDecimal) -> U256 {
+    let bigint = value.with_scale(0).into_bigint_and_exponent().0;
+    biguint_to_u256(bigint.to_biguint().unwrap())
+}
+
 fn ratio_to_big_decimal(num: &Ratio<BigUint>, precision: usize) -> BigDecimal {
     let bigint = round_precision_raw_no_div(num, precision)
         .to_bigint()

watcher/Cargo.toml

@@ -19,3 +19,4 @@ client = { path = "../client" }
 chain-events = { path = "../chain-events" }
 storage = { path = "../storage" }
 vlog = { path = "../vlog" }
+withdrawals-meterer = { path = "../withdrawals-meterer" }