wait for n block confirmations for evm chains before processing events

crates/relayer-config/src/evm/mod.rs

@@ -25,6 +25,9 @@ pub struct EvmChainConfig {
     /// Websocket Endpoint for long living connections
     #[serde(skip_serializing)]
     pub ws_endpoint: RpcUrl,
+    /// Block confirmations
+    #[serde(skip_serializing)]
+    pub block_confirmations: u32,
     /// Block Explorer for this chain.
     ///
     /// Optional, and only used for printing a clickable links

src/events_watcher/mod.rs

@@ -30,6 +30,7 @@ use std::ops::Deref;
 use std::sync::Arc;
 use std::time::Duration;
 
+use crate::context::RelayerContext;
 use crate::metric;
 use webb::substrate::subxt::client::OfflineClientT;
 use webb::{
@@ -44,7 +45,6 @@ use webb::{
         subxt::{self, client::OnlineClientT, ext::sp_runtime::traits::Header},
     },
 };
-
 use webb_relayer_store::sled::SledQueueKey;
 use webb_relayer_store::{
     BridgeCommand, BridgeKey, EventHashStore, HistoryStore, ProposalStore,
@@ -123,18 +123,28 @@ pub trait EventWatcher {
         store: Arc<Self::Store>,
         contract: Self::Contract,
         handlers: Vec<EventHandlerFor<Self>>,
-        metrics: Arc<metric::Metrics>,
+        ctx: &RelayerContext,
     ) -> crate::Result<()> {
         let backoff = backoff::backoff::Constant::new(Duration::from_secs(1));
         let task = || async {
             let step = contract.max_blocks_per_step();
             // saves the last time we printed sync progress.
             let mut instant = std::time::Instant::now();
+            let metrics = &ctx.metrics;
             let chain_id = client
                 .get_chainid()
                 .map_err(Into::into)
                 .map_err(backoff::Error::transient)
                 .await?;
+            let chain_config =
+                ctx.config.evm.get(&chain_id.to_string()).ok_or_else(|| {
+                    crate::Error::ChainNotFound {
+                        chain_id: chain_id.clone().to_string(),
+                    }
+                })?;
+            // no of blocks confirmation required before processing it
+            let block_confirmations: U64 =
+                chain_config.block_confirmations.into();
             // now we start polling for new events.
             loop {
                 let block = store.get_last_block_number(
@@ -150,9 +160,12 @@ pub trait EventWatcher {
                     "Latest block number: #{}",
                     current_block_number
                 );
-                let dest_block = cmp::min(block + step, current_block_number);
+                // latest finalized block after n block_confirmations
+                let latest_finalized_block =
+                    current_block_number.saturating_sub(block_confirmations);
+                let dest_block = cmp::min(block + step, latest_finalized_block);
                 // check if we are now on the latest block.
-                let should_cooldown = dest_block == current_block_number;
+                let should_cooldown = dest_block == latest_finalized_block;
                 tracing::trace!("Reading from #{} to #{}", block, dest_block);
                 // Only handle events from found blocks if they are new
                 if dest_block != block {

src/service.rs

@@ -723,7 +723,7 @@ async fn start_evm_vanchor_events_watcher(
                         Box::new(leaves_handler),
                         Box::new(encrypted_output_handler),
                     ],
-                    my_ctx.metrics.clone(),
+                    &my_ctx,
                 );
                 tokio::select! {
                     _ = vanchor_watcher_task => {
@@ -754,7 +754,7 @@ async fn start_evm_vanchor_events_watcher(
                         Box::new(leaves_handler),
                         Box::new(encrypted_output_handler),
                     ],
-                    my_ctx.metrics.clone(),
+                    &my_ctx,
                 );
                 tokio::select! {
                     _ = vanchor_watcher_task => {
@@ -783,7 +783,7 @@ async fn start_evm_vanchor_events_watcher(
                         Box::new(leaves_handler),
                         Box::new(encrypted_output_handler),
                     ],
-                    my_ctx.metrics.clone(),
+                    &my_ctx,
                 );
                 tokio::select! {
                     _ = vanchor_watcher_task => {
@@ -828,6 +828,7 @@ async fn start_signature_bridge_events_watcher(
     let wrapper =
         SignatureBridgeContractWrapper::new(config.clone(), client.clone());
     let metrics = ctx.metrics.clone();
+    let my_ctx = ctx.clone();
     let task = async move {
         tracing::debug!(
             "Signature Bridge watcher for ({}) Started.",
@@ -842,7 +843,7 @@ async fn start_signature_bridge_events_watcher(
             store.clone(),
             wrapper.clone(),
             vec![Box::new(governance_transfer_handler)],
-            metrics.clone(),
+            &my_ctx,
         );
         let cmd_handler_task = BridgeWatcher::run(
             &bridge_contract_watcher,

tests/lib/localTestnet.ts

@@ -55,6 +55,7 @@ export type ExportedConfigOptions = {
   withdrawConfig?: WithdrawConfig;
   relayerWallet?: Wallet;
   linkedAnchors?: LinkedAnchor[];
+  blockConfirmations?: number;
 };
 
 // Default Events watcher for the contracts.
@@ -364,6 +365,7 @@ export class LocalChain {
       enabled: true,
       httpEndpoint: this.endpoint,
       wsEndpoint: this.endpoint.replace('http', 'ws'),
+      blockConfirmations: opts.blockConfirmations ?? 1,
       chainId: this.underlyingChainId,
       beneficiary: (wallet as ethers.Wallet).address,
       privateKey: (wallet as ethers.Wallet).privateKey,
@@ -380,6 +382,7 @@ export class LocalChain {
       enabled: true,
       httpEndpoint: this.endpoint,
       wsEndpoint: this.endpoint.replace('http', 'ws'),
+      blockConfirmations: opts.blockConfirmations ?? 1,
       chainId: this.underlyingChainId,
       beneficiary: '',
       privateKey: '',
@@ -432,6 +435,7 @@ export class LocalChain {
       'http-endpoint': config.httpEndpoint,
       'ws-endpoint': config.wsEndpoint,
       'chain-id': config.chainId,
+      'block-confirmations': config.blockConfirmations,
       beneficiary: config.beneficiary,
       'private-key': config.privateKey,
       contracts: config.contracts.map((contract) => ({

tests/lib/webbRelayer.ts

@@ -503,6 +503,7 @@ export interface ChainInfo {
   chainId: number;
   beneficiary?: string;
   contracts: Contract[];
+  blockConfirmations: number;
 }
 
 export interface Contract {

tests/test/evm/evmToSubstrateTransaction.test.ts

@@ -196,6 +196,7 @@ describe('Cross chain transaction <<>> Mocked Backend', function () {
       linkedAnchors: [
         { type: 'Raw', resourceId: substrateResourceId.toString() },
       ],
+      blockConfirmations: 15
     });
 
     // This are pre-requisites for creating substrate chain.