feat: retry with an explicit delay for worker messages

Author: kiseln

omni-relayer/src/startup/bridge_indexer.rs

@@ -101,25 +101,21 @@ fn is_whitelisted_transaction_event(
     transfer_message: &OmniTransferMessage,
 ) -> bool {
     match transfer_message {
-        OmniTransferMessage::NearTransferMessage(transfer_message) => {
-            config
-                .bridge_indexer
-                .is_token_whitelisted(&transfer_message.token)
-        }
-        OmniTransferMessage::NearSignTransferEvent(sign_event) => {
-            config
-                .bridge_indexer
-                .is_token_whitelisted(&sign_event.message_payload.token_address)
-        }
-        OmniTransferMessage::EvmInitTransferMessage(init_transfer) => {
-            config.bridge_indexer.is_token_whitelisted(&init_transfer.token)
-        }
-        OmniTransferMessage::SolanaInitTransfer(init_transfer) => {
-            config.bridge_indexer.is_token_whitelisted(&init_transfer.token)
-        }
-        OmniTransferMessage::StarknetInitTransfer(init_transfer) => {
-            config.bridge_indexer.is_token_whitelisted(&init_transfer.token)
-        }
+        OmniTransferMessage::NearTransferMessage(transfer_message) => config
+            .bridge_indexer
+            .is_token_whitelisted(&transfer_message.token),
+        OmniTransferMessage::NearSignTransferEvent(sign_event) => config
+            .bridge_indexer
+            .is_token_whitelisted(&sign_event.message_payload.token_address),
+        OmniTransferMessage::EvmInitTransferMessage(init_transfer) => config
+            .bridge_indexer
+            .is_token_whitelisted(&init_transfer.token),
+        OmniTransferMessage::SolanaInitTransfer(init_transfer) => config
+            .bridge_indexer
+            .is_token_whitelisted(&init_transfer.token),
+        OmniTransferMessage::StarknetInitTransfer(init_transfer) => config
+            .bridge_indexer
+            .is_token_whitelisted(&init_transfer.token),
         OmniTransferMessage::NearUtxoTransferMessage { token_id, .. } => config
             .bridge_indexer
             .is_token_whitelisted(&OmniAddress::Near(token_id.clone())),
@@ -167,7 +163,11 @@ async fn handle_transaction_event(
     event: OmniTransactionEvent,
 ) -> Result<()> {
     if config.bridge_indexer.is_whitelist_active()
-        && !is_whitelisted_transaction_event(config, event.transfer_id.origin_chain, &event.transfer_message)
+        && !is_whitelisted_transaction_event(
+            config,
+            event.transfer_id.origin_chain,
+            &event.transfer_message,
+        )
     {
         debug!(
             "Whitelist mode active, skipping transaction event: {:?}",

omni-relayer/src/startup/mod.rs

@@ -177,7 +177,8 @@ fn build_utxo_bridge_client<C: utxo_bridge_client::types::UTXOChain>(
         }
     };
 
-    Ok(utxo.as_ref()
+    Ok(utxo
+        .as_ref()
         .map(|utxo| UTXOBridgeClient::new(utxo.rpc_http_url.clone(), AuthOptions::None)))
 }
 

omni-relayer/src/workers/evm.rs

@@ -1,4 +1,4 @@
-use std::sync::Arc;
+use std::{sync::Arc, time::Duration};
 
 use anyhow::{Context, Result};
 use bridge_connector_common::result::{BridgeSdkError, EthRpcError};
@@ -51,7 +51,9 @@ pub async fn process_init_transfer_event(
     let current_timestamp = chrono::Utc::now().timestamp();
 
     if current_timestamp < creation_timestamp + expected_finalization_time {
-        return Ok(EventAction::Retry);
+        let remaining =
+            (creation_timestamp + expected_finalization_time - current_timestamp) as u64;
+        return Ok(EventAction::RetryAfter(Duration::from_secs(remaining)));
     }
 
     info!(
@@ -326,7 +328,9 @@ pub async fn process_evm_transfer_event(
     let current_timestamp = chrono::Utc::now().timestamp();
 
     if current_timestamp < creation_timestamp + expected_finalization_time {
-        return Ok(EventAction::Retry);
+        let remaining =
+            (creation_timestamp + expected_finalization_time - current_timestamp) as u64;
+        return Ok(EventAction::RetryAfter(Duration::from_secs(remaining)));
     }
 
     info!("Processing FinTransfer ({chain_kind:?}): {transaction_hash:?}");
@@ -463,7 +467,9 @@ pub async fn process_deploy_token_event(
     let current_timestamp = chrono::Utc::now().timestamp();
 
     if current_timestamp < creation_timestamp + expected_finalization_time {
-        return Ok(EventAction::Retry);
+        let remaining =
+            (creation_timestamp + expected_finalization_time - current_timestamp) as u64;
+        return Ok(EventAction::RetryAfter(Duration::from_secs(remaining)));
     }
 
     info!("Processing DeployToken ({chain_kind:?}): {transaction_hash:?}");

omni-relayer/src/workers/mod.rs

@@ -56,6 +56,7 @@ impl<E> RetryableEvent<E> {
 
 pub enum EventAction {
     Retry,
+    RetryAfter(Duration),
     Remove,
 }
 
@@ -176,7 +177,7 @@ async fn handle_nats_ack(
     let max_message_age = Duration::from_secs(config.max_message_age_hours * 3600);
 
     match result {
-        Ok(EventAction::Retry) => {
+        Ok(EventAction::Retry) | Ok(EventAction::RetryAfter(_)) => {
             if let Ok(info) = msg.info() {
                 let now = chrono::Utc::now().timestamp();
                 let published_at = info.published.unix_timestamp();
@@ -190,8 +191,11 @@ async fn handle_nats_ack(
                     return;
                 }
 
-                let backoff = Duration::from_secs(4u64.saturating_pow(info.delivered as u32))
-                    .min(max_backoff);
+                let backoff = if let Ok(EventAction::RetryAfter(delay)) = result {
+                    (*delay).min(max_backoff)
+                } else {
+                    Duration::from_secs(4u64.saturating_pow(info.delivered as u32)).min(max_backoff)
+                };
                 msg.ack_with(async_nats::jetstream::AckKind::Nak(Some(backoff)))
                     .await
                     .ok();
@@ -326,7 +330,10 @@ pub async fn process_events(
             }
 
             if message_result.needs_evm_nonce_resync
-                && matches!(message_result.action, Ok(EventAction::Retry) | Err(_))
+                && matches!(
+                    message_result.action,
+                    Ok(EventAction::Retry) | Ok(EventAction::RetryAfter(_)) | Err(_)
+                )
             {
                 is_evm_nonce_resync_needed.store(true, Ordering::Relaxed);
             }