chore(withdrawal-finalizer): adds metrics for unfinalized eth withdra…

…wals (#351)

# What ❔

<!-- What are the changes this PR brings about? -->
<!-- Example: This PR adds a PR template to the repo. -->
<!-- (For bigger PRs adding more context is appreciated) -->

## Why ❔

<!-- Why are these changes done? What goal do they contribute to? What
are the principles behind them? -->
<!-- Example: PR templates ensure PR reviewers, observers, and future
iterators are in context about the evolution of repos. -->

## Checklist

<!-- Check your PR fulfills the following items. -->
<!-- For draft PRs check the boxes as you complete them. -->

- [ ] PR title corresponds to the body of PR (we generate changelog
entries from PRs).
- [ ] Tests for the changes have been added / updated.
- [ ] Documentation comments have been added / updated.
- [ ] Code has been formatted via `cargo fmt`.

bin/withdrawal-finalizer/src/main.rs

@@ -304,6 +304,11 @@ async fn main() -> Result<()> {
     );
     let finalizer_handle = tokio::spawn(finalizer.run(client_l2));
 
+    let metrics_handle = tokio::spawn(metrics::meter_unfinalized_withdrawals(
+        pgpool.clone(),
+        eth_finalization_threshold,
+    ));
+
     let api_server = tokio::spawn(api::run_server(pgpool));
 
     tokio::select! {
@@ -322,6 +327,9 @@ async fn main() -> Result<()> {
         r = finalizer_handle => {
             tracing::error!("Finalizer ended with {r:?}");
         }
+        _ = metrics_handle => {
+            tracing::error!("Metrics loop has ended");
+        }
     }
 
     stop_vise_exporter.send_replace(());

bin/withdrawal-finalizer/src/metrics.rs

@@ -1,7 +1,13 @@
 //! Metrics for main binary
 
+use std::time::Duration;
+
+use ethers::types::U256;
+use sqlx::PgPool;
 use vise::{Gauge, Metrics};
 
+const METRICS_REFRESH_PERIOD: Duration = Duration::from_secs(15);
+
 /// Main finalizer binary metrics
 #[derive(Debug, Metrics)]
 #[metrics(prefix = "withdrawal_finalizer")]
@@ -11,7 +17,37 @@ pub(super) struct FinalizerMainMetrics {
 
     /// Capacity of the channel sending L2 events.
     pub watcher_l2_channel_capacity: Gauge,
+
+    /// The withdrawals that were not finalized but are executed
+    pub executed_eth_withdrawals_not_finalized: Gauge,
+
+    /// The withdrawals that
+    pub unexecuted_eth_withdrawals_below_current_threshold: Gauge,
 }
 
 #[vise::register]
 pub(super) static MAIN_FINALIZER_METRICS: vise::Global<FinalizerMainMetrics> = vise::Global::new();
+
+pub async fn meter_unfinalized_withdrawals(pool: PgPool, eth_threshold: Option<U256>) {
+    loop {
+        tokio::time::sleep(METRICS_REFRESH_PERIOD).await;
+
+        let Ok(executed_not_finalized) =
+            storage::get_executed_and_not_finalized_withdrawals_count(&pool).await
+        else {
+            continue;
+        };
+        let Ok(unexecuted) = storage::get_unexecuted_withdrawals_count(&pool, eth_threshold).await
+        else {
+            continue;
+        };
+
+        MAIN_FINALIZER_METRICS
+            .executed_eth_withdrawals_not_finalized
+            .set(executed_not_finalized);
+
+        MAIN_FINALIZER_METRICS
+            .unexecuted_eth_withdrawals_below_current_threshold
+            .set(unexecuted);
+    }
+}

storage/src/lib.rs

@@ -974,6 +974,77 @@ pub async fn withdrawals_to_finalize(
     Ok(data)
 }
 
+/// Get the number of ETH withdrawals not yet executed and finalized and above some threshold
+pub async fn get_unexecuted_withdrawals_count(
+    pool: &PgPool,
+    eth_threshold: Option<U256>,
+) -> Result<i64> {
+    // if no threshold, query _all_ ethereum withdrawals since all of them are >= 0.
+    let eth_threshold = eth_threshold.unwrap_or(U256::zero());
+
+    let count = sqlx::query!(
+        "
+        SELECT
+            COUNT(*)
+        FROM
+          finalization_data
+          JOIN withdrawals w ON finalization_data.withdrawal_id = w.id
+        WHERE
+          finalization_tx IS NULL
+          AND finalization_data.l2_block_number > COALESCE(
+            (
+              SELECT
+                MAX(l2_block_number)
+              FROM
+                l2_blocks
+              WHERE
+                execute_l1_block_number IS NOT NULL
+            ),
+            1
+          )
+          AND token = decode('000000000000000000000000000000000000800A', 'hex') 
+          AND amount >= $1
+        ",
+        u256_to_big_decimal(eth_threshold),
+    )
+    .fetch_one(pool)
+    .await?;
+
+    Ok(count.count.unwrap_or(0))
+}
+
+/// Get the number of ETH withdrawals executed but not finalized
+pub async fn get_executed_and_not_finalized_withdrawals_count(pool: &PgPool) -> Result<i64> {
+    let count = sqlx::query!(
+        "
+        SELECT
+            COUNT(*)
+        FROM
+          finalization_data
+          JOIN withdrawals w ON finalization_data.withdrawal_id = w.id
+        WHERE
+          finalization_tx IS NULL
+          AND failed_finalization_attempts = 0
+          AND finalization_data.l2_block_number <= COALESCE(
+            (
+              SELECT
+                MAX(l2_block_number)
+              FROM
+                l2_blocks
+              WHERE
+                execute_l1_block_number IS NOT NULL
+            ),
+            1
+          )
+          AND token = decode('000000000000000000000000000000000000800A', 'hex') 
+        ",
+    )
+    .fetch_one(pool)
+    .await?;
+
+    Ok(count.count.unwrap_or(0))
+}
+
 /// Fetch finalization parameters for some withdrawal
 pub async fn get_finalize_withdrawal_params(
     pool: &PgPool,