feat: collect withdrawal volumes metrics

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

@@ -183,6 +183,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 +206,16 @@ where
                     "finalizer.highest_finalized_batch_number",
                     highest_batch_number.as_u64() as f64,
                 );
+
+                if let Err(e) = withdrawals_meterer::meter_finalized_withdrawals_storage(
+                    &self.pgpool,
+                    ids,
+                    "era_withdrawal_finalizer_withdrawn_tokens",
+                )
+                .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 * 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" }

watcher/src/lib.rs

@@ -295,6 +295,16 @@ async fn process_withdrawals_in_block(pool: &PgPool, events: Vec<WithdrawalEvent
         });
     }
 
+    if let Err(e) = withdrawals_meterer::meter_finalized_withdrawals(
+        pool,
+        &stored_withdrawals,
+        "era_withdrawn_tokens_amounts_tracker",
+    )
+    .await
+    {
+        vlog::error!("Failed to meter requested withdrawals: {e}");
+    }
+
     storage::add_withdrawals(pool, &stored_withdrawals).await?;
     Ok(())
 }

withdrawals-meterer/Cargo.toml

@@ -0,0 +1,18 @@
+[package]
+name = "withdrawals-meterer"
+version = "0.1.11"
+authors = ["The Matter Labs Team <[email protected]>"]
+homepage = "https://zksync.io/"
+license = "MIT OR Apache-2.0"
+edition = "2021"
+
+[dependencies]
+ethers = "2.0.10"
+lazy_static = "1.4.0"
+metrics = "0.21.1"
+tokio = "1.32.0"
+
+client = { path = "../client" }
+sqlx = { version = "0.7", features = ["postgres", "runtime-tokio-rustls"] }
+storage = { path = "../storage" }
+vlog = { path = "../vlog" }

withdrawals-meterer/src/lib.rs

@@ -0,0 +1,92 @@
+#![deny(unused_crate_dependencies)]
+#![warn(missing_docs)]
+#![warn(unused_extern_crates)]
+#![warn(unused_imports)]
+
+//! A utility crate that meters withdrawals amounts.
+
+use std::{collections::HashMap, str::FromStr, sync::Arc};
+
+use client::ETH_TOKEN_ADDRESS;
+use ethers::types::Address;
+use lazy_static::lazy_static;
+use sqlx::PgPool;
+use storage::StoredWithdrawal;
+use tokio::sync::RwLock;
+
+lazy_static! {
+    static ref TOKEN_DECIMALS: Arc<RwLock<HashMap<Address, u32>>> = {
+        let mut map = HashMap::new();
+        map.insert(ETH_TOKEN_ADDRESS, 18_u32);
+
+        Arc::new(RwLock::new(map))
+    };
+}
+
+/// Given a set of withdrawal ids meter all of them to a metric
+/// with a given name.
+pub async fn meter_finalized_withdrawals_storage(
+    pool: &PgPool,
+    ids: Vec<i64>,
+    metric_name: &'static str,
+) -> Result<(), storage::Error> {
+    let withdrawals = storage::get_withdrawals(pool, &ids).await?;
+
+    meter_finalized_withdrawals(pool, &withdrawals, metric_name).await?;
+
+    Ok(())
+}
+
+/// Given a set of [`StoredWithdrawal`], meter all of them to a
+/// metric with a given name.
+///
+/// This function returns only storage error, all formatting, etc
+/// errors will be just logged.
+pub async fn meter_finalized_withdrawals(
+    pool: &PgPool,
+    withdrawals: &[StoredWithdrawal],
+    metric_name: &'static str,
+) -> Result<(), storage::Error> {
+    for w in withdrawals {
+        let guard = TOKEN_DECIMALS.read().await;
+        let decimals = guard.get(&w.event.token).copied();
+        drop(guard);
+
+        let decimals = match decimals {
+            None => {
+                let Some(decimals) = storage::token_decimals(pool, w.event.token).await? else {
+                    vlog::error!("Received withdrawal from unknown token {:?}", w.event.token);
+                    continue;
+                };
+
+                TOKEN_DECIMALS.write().await.insert(w.event.token, decimals);
+                decimals
+            }
+            Some(decimals) => decimals,
+        };
+
+        let formatted = match ethers::utils::format_units(w.event.amount, decimals) {
+            Ok(f) => f,
+            Err(e) => {
+                vlog::error!("failed to format units: {e}");
+                continue;
+            }
+        };
+
+        let formatted_f64 = match f64::from_str(&formatted) {
+            Ok(f) => f,
+            Err(e) => {
+                vlog::error!("failed to format units: {e}");
+                continue;
+            }
+        };
+
+        metrics::increment_gauge!(
+            metric_name,
+            formatted_f64,
+            "token" => format!("{:?}", w.event.token)
+        )
+    }
+
+    Ok(())
+}