fix: monitor withdrawalinitiated events (#299)

Some bridges only emit `WithdrawalInitiated` events.

README.md

@@ -55,6 +55,7 @@ Deployment is done by deploying a dockerized image of the service.
 | `FINALIZE_ETH_TOKEN` | (Optional) Configure, whether the Ethereum withdrawal events should be monitored. Useful to turn off for custom bridges that are only interested in a particular ERC20 token and have nothing to do with main Ethereum withdrawals |
 | `CUSTOM_TOKEN_DEPLOYER_ADDRESSES` | (Optional) Normally ERC20 tokens are deployed by the bridge contract. However, in custom cases it may be necessary to override that behavior with a custom set of addresses that have deployed tokens |
 | `CUSTOM_TOKEN_ADDRESSES` | (Optional) Adds a predefined list of tokens to finalize. May be useful in case of custom bridge setups when the regular technique of finding token deployments does not work. |
+| `ENABLE_WITHDRAWAL_METERING` | (Optional, default: `"true"`) By default Finalizer collects metrics about withdrawn token volumens. Users may optionally switch off this metering. |
 
 The configuration structure describing the service config can be found in [`config.rs`](https://github.com/matter-labs/zksync-withdrawal-finalizer/blob/main/bin/withdrawal-finalizer/src/config.rs)
 

bin/withdrawal-finalizer/src/config.rs

@@ -74,4 +74,7 @@ pub struct Config {
 
     #[envconfig(from = "CUSTOM_TOKEN_ADDRESSES")]
     pub custom_token_addresses: Option<AddrList>,
+
+    #[envconfig(from = "ENABLE_WITHDRAWAL_METERING")]
+    pub enable_withdrawal_metering: Option<bool>,
 }

bin/withdrawal-finalizer/src/main.rs

@@ -203,6 +203,11 @@ async fn main() -> Result<()> {
 
     let (mut tokens, last_token_seen_at_block) = storage::get_tokens(&pgpool.clone()).await?;
 
+    if let Some(ref custom_tokens) = config.custom_token_addresses {
+        tokens.extend_from_slice(custom_tokens.0.as_slice());
+    }
+
+    tracing::info!("tokens {tokens:?}");
     if let Some(ref custom_tokens) = config.custom_token_deployer_addresses {
         tokens.extend_from_slice(custom_tokens.0.as_slice());
     }
@@ -221,7 +226,10 @@ async fn main() -> Result<()> {
 
     let zksync_contract = IZkSync::new(config.diamond_proxy_addr, client_l1.clone());
 
-    let watcher = Watcher::new(client_l2.clone(), pgpool.clone());
+    // by default meter withdrawals
+    let meter_withdrawals = config.enable_withdrawal_metering.unwrap_or(true);
+
+    let watcher = Watcher::new(client_l2.clone(), pgpool.clone(), meter_withdrawals);
 
     let withdrawal_events_handle = tokio::spawn(l2_events.run_with_reconnects(
         from_l2_block,
@@ -284,6 +292,7 @@ async fn main() -> Result<()> {
         config.tx_retry_timeout,
         finalizer_account_address,
         config.tokens_to_finalize.unwrap_or_default(),
+        meter_withdrawals,
     );
     let finalizer_handle = tokio::spawn(finalizer.run(client_l2));
 

chain-events/src/l2_events.rs

@@ -5,6 +5,7 @@ use futures::{Sink, SinkExt, StreamExt};
 use client::{
     contracts_deployer::codegen::ContractDeployedFilter,
     ethtoken::codegen::WithdrawalFilter,
+    l2bridge::codegen::WithdrawalInitiatedFilter,
     l2standard_token::codegen::{
         BridgeBurnFilter, BridgeInitializationFilter, BridgeInitializeFilter,
     },
@@ -38,6 +39,7 @@ pub struct L2EventsListener {
 enum L2Events {
     BridgeBurn(BridgeBurnFilter),
     Withdrawal(WithdrawalFilter),
+    WithdrawalInitiated(WithdrawalInitiatedFilter),
     ContractDeployed(ContractDeployedFilter),
 }
 
@@ -335,14 +337,21 @@ impl L2EventsListener {
         let last_seen_l2_token_block: BlockNumber = last_seen_l2_token_block.into();
         let from_block: BlockNumber = from_block.into();
 
-        let past_topic0 = vec![BridgeBurnFilter::signature(), WithdrawalFilter::signature()];
+        let past_topic0 = vec![
+            BridgeBurnFilter::signature(),
+            WithdrawalFilter::signature(),
+            WithdrawalInitiatedFilter::signature(),
+        ];
 
         let topic0 = vec![
             ContractDeployedFilter::signature(),
             BridgeBurnFilter::signature(),
             WithdrawalFilter::signature(),
+            WithdrawalInitiatedFilter::signature(),
         ];
 
+        tracing::info!("topic0 {topic0:?}");
+
         tracing::debug!("last_seen_l2_token_block {last_seen_l2_token_block:?}");
         tracing::debug!("from_block {from_block:?}");
 
@@ -364,7 +373,8 @@ impl L2EventsListener {
             .await?;
         }
 
-        let tokens = self.tokens.iter().cloned().collect::<Vec<_>>();
+        let mut tokens = self.tokens.iter().cloned().collect::<Vec<_>>();
+        tokens.extend_from_slice(self.token_deployer_addrs.as_slice());
 
         tracing::info!("Listeing to events from tokens {tokens:?}");
 
@@ -379,6 +389,9 @@ impl L2EventsListener {
             .address(tokens)
             .topic0(topic0);
 
+        tracing::info!("filter past {past_filter:#?}");
+        tracing::info!("filter {filter:#?}");
+
         let past_logs = middleware.get_logs_paginated(&past_filter, pagination_step);
         let current_logs = middleware
             .subscribe_logs(&filter)
@@ -473,6 +486,26 @@ impl L2EventsListener {
                         .await
                         .map_err(|_| Error::ChannelClosing)?;
                 }
+                L2Events::WithdrawalInitiated(WithdrawalInitiatedFilter {
+                    amount,
+                    l_2_token,
+                    ..
+                }) => {
+                    CHAIN_EVENTS_METRICS.withdrawal_events.inc();
+
+                    let we = WithdrawalEvent {
+                        tx_hash,
+                        block_number: block_number.as_u64(),
+                        token: *l_2_token,
+                        amount: *amount,
+                    };
+                    let event = we.into();
+                    tracing::info!("sending withdrawal event {event:?}");
+                    sender
+                        .send(event)
+                        .await
+                        .map_err(|_| Error::ChannelClosing)?;
+                }
                 L2Events::ContractDeployed(_) => {
                     let tx = middleware
                         .zks_get_transaction_receipt(log.transaction_hash.unwrap_or_else(|| {

finalizer/src/lib.rs

@@ -101,7 +101,7 @@ pub struct Finalizer<M1, M2> {
     tx_fee_limit: U256,
     tx_retry_timeout: Duration,
     account_address: Address,
-    withdrawals_meterer: WithdrawalsMeter,
+    withdrawals_meterer: Option<WithdrawalsMeter>,
     token_list: TokenList,
 }
 
@@ -131,9 +131,12 @@ where
         tx_retry_timeout: usize,
         account_address: Address,
         token_list: TokenList,
+        meter_withdrawals: bool,
     ) -> Self {
-        let withdrawals_meterer =
-            WithdrawalsMeter::new(pgpool.clone(), MeteringComponent::FinalizedWithdrawals);
+        let withdrawals_meterer = meter_withdrawals.then_some(WithdrawalsMeter::new(
+            pgpool.clone(),
+            MeteringComponent::FinalizedWithdrawals,
+        ));
         let tx_fee_limit = ethers::utils::parse_ether(TX_FEE_LIMIT)
             .expect("{TX_FEE_LIMIT} ether is a parsable amount; qed");
 
@@ -275,12 +278,10 @@ where
                     .highest_finalized_batch_number
                     .set(highest_batch_number.as_u64() as i64);
 
-                if let Err(e) = self
-                    .withdrawals_meterer
-                    .meter_withdrawals_storage(&ids)
-                    .await
-                {
-                    tracing::error!("Failed to meter the withdrawals: {e}");
+                if let Some(ref mut withdrawals_meterer) = self.withdrawals_meterer {
+                    if let Err(e) = withdrawals_meterer.meter_withdrawals_storage(&ids).await {
+                        tracing::error!("Failed to meter the withdrawals: {e}");
+                    }
                 }
             }
             // TODO: why would a pending tx resolve to `None`?

watcher/src/lib.rs

@@ -34,18 +34,19 @@ pub type Result<T> = std::result::Result<T, Error>;
 pub struct Watcher<M2> {
     l2_provider: Arc<M2>,
     pgpool: PgPool,
-    withdrawals_meterer: WithdrawalsMeter,
+    withdrawals_meterer: Option<WithdrawalsMeter>,
 }
 
 impl<M2> Watcher<M2>
 where
     M2: ZksyncMiddleware + 'static,
     <M2 as Middleware>::Provider: JsonRpcClient,
 {
-    #[allow(clippy::too_many_arguments)]
-    pub fn new(l2_provider: Arc<M2>, pgpool: PgPool) -> Self {
-        let withdrawals_meterer =
-            WithdrawalsMeter::new(pgpool.clone(), MeteringComponent::RequestedWithdrawals);
+    pub fn new(l2_provider: Arc<M2>, pgpool: PgPool, meter_withdrawals: bool) -> Self {
+        let withdrawals_meterer = meter_withdrawals.then_some(WithdrawalsMeter::new(
+            pgpool.clone(),
+            MeteringComponent::RequestedWithdrawals,
+        ));
 
         Self {
             l2_provider,
@@ -302,7 +303,7 @@ where
 async fn process_withdrawals_in_block(
     pool: &PgPool,
     events: Vec<WithdrawalEvent>,
-    withdrawals_meterer: &mut WithdrawalsMeter,
+    withdrawals_meterer: &mut Option<WithdrawalsMeter>,
 ) -> Result<()> {
     use itertools::Itertools;
     let group_by = events.into_iter().group_by(|event| event.tx_hash);
@@ -328,11 +329,13 @@ async fn process_withdrawals_in_block(
         });
     }
 
-    if let Err(e) = withdrawals_meterer
-        .meter_withdrawals(&stored_withdrawals)
-        .await
-    {
-        tracing::error!("Failed to meter requested withdrawals: {e}");
+    if let Some(ref mut withdrawals_meterer) = withdrawals_meterer {
+        if let Err(e) = withdrawals_meterer
+            .meter_withdrawals(&stored_withdrawals)
+            .await
+        {
+            tracing::error!("Failed to meter requested withdrawals: {e}");
+        }
     }
 
     storage::add_withdrawals(pool, &stored_withdrawals).await?;
@@ -376,7 +379,7 @@ async fn run_l2_events_loop<WE>(
     pool: PgPool,
     we: WE,
     from_l2_block: u64,
-    mut withdrawals_meterer: WithdrawalsMeter,
+    mut withdrawals_meterer: Option<WithdrawalsMeter>,
 ) -> Result<()>
 where
     WE: Stream<Item = L2Event>,