refactor(eth-gear): Refactor main loop of the TxManager in eth-gear relayer (#813)

Author: gshep

Cargo.lock

@@ -44,7 +44,7 @@ members = [
 resolver = "2"
 
 [workspace.package]
-version = "1.0.4"
+version = "1.0.5"
 edition = "2021"
 
 [workspace.dependencies]

Cargo.toml

@@ -4,6 +4,9 @@ use sails_rs::{calls::*, gclient::calls::*, prelude::*};
 use vft_client::traits::*;
 use vft_manager_client::{traits::*, Order};
 
+// The constant is intentionally duplicated since vara-runtime is too heavy dependency.
+pub const UNITS: u128 = 1_000_000_000_000;
+
 /// Asynchronously migrates balances from an old VFT contract to a new one.
 ///
 /// This function performs a batched migration of token balances between two VFT contracts.

gear-common/src/lib.rs

@@ -31,6 +31,7 @@ ethereum_beacon_client.workspace = true
 ethereum-common.workspace = true
 futures.workspace = true
 gear-core.workspace = true
+gear-common.workspace = true
 gclient.workspace = true
 humantime.workspace = true
 hex.workspace = true

relayer/Cargo.toml

@@ -27,8 +27,6 @@ mod sync_update;
 const SIZE_CHANNEL: usize = 100_000;
 const COUNT_FAILURE: usize = 3;
 const DELAY_SECS_UPDATE_REQUEST: u64 = 30;
-// The constant is intentionally duplicated since vara-runtime is too heavy dependency.
-const UNITS: u128 = 1_000_000_000_000;
 
 struct SyncUpdate {
     sync_update: SyncCommitteeUpdate,
@@ -219,11 +217,12 @@ impl Relayer {
 async fn update_total_balance(client: &GearApi, update_metrics: &metrics::Updates) {
     match client.total_balance(client.account_id()).await {
         Ok(total_balance) => {
-            let total_balance = total_balance / UNITS;
+            let total_balance = total_balance / gear_common::UNITS;
             let total_balance: i64 = total_balance.try_into().unwrap_or(i64::MAX);
 
             update_metrics.account_total_balance.set(total_balance);
         }
+
         Err(e) => log::error!("Unable to get total balance: {e:?}"),
     }
 }

relayer/src/ethereum_checkpoints/mod.rs

@@ -1,7 +1,9 @@
+use crate::message_relayer::eth_to_gear::api_provider::ApiProviderConnection;
 use alloy_primitives::FixedBytes;
 use eth_events_electra_client::EthToVaraEvent;
 use futures::executor::block_on;
 use gclient::GearApi;
+use gear_common::UNITS;
 use historical_proxy_client::{traits::HistoricalProxy as _, HistoricalProxy};
 use primitive_types::H256;
 use prometheus::{
@@ -14,15 +16,13 @@ use sails_rs::{
     Encode,
 };
 use tokio::{
-    sync::mpsc::{unbounded_channel, UnboundedReceiver, UnboundedSender},
+    sync::mpsc::{error::TryRecvError, unbounded_channel, UnboundedReceiver, UnboundedSender},
     task::spawn_blocking,
 };
 use utils_prometheus::{impl_metered_service, MeteredService};
 use uuid::Uuid;
 use vft_manager_client::vft_manager::io::SubmitReceipt;
 
-use crate::message_relayer::eth_to_gear::api_provider::ApiProviderConnection;
-
 pub struct MessageSenderIo {
     requests_channel: UnboundedSender<Request>,
     responses_channel: UnboundedReceiver<Response>,
@@ -55,8 +55,8 @@ impl MessageSenderIo {
             .is_ok()
     }
 
-    pub async fn recv(&mut self) -> Option<Response> {
-        self.responses_channel.recv().await
+    pub fn try_recv(&mut self) -> Result<Response, TryRecvError> {
+        self.responses_channel.try_recv()
     }
 }
 
@@ -161,6 +161,7 @@ impl MessageSender {
         responses: &mut UnboundedSender<Response>,
     ) -> anyhow::Result<()> {
         let gear_api = self.api_provider.gclient_client(&self.suri)?;
+        self.update_balance_metric(&gear_api).await?;
 
         if let Some(request) = self.last_request.take() {
             match self.process(responses, &gear_api, &request).await {
@@ -329,6 +330,7 @@ impl MessageSender {
                 }
             }
         }
+
         Ok(true)
     }
 
@@ -338,7 +340,7 @@ impl MessageSender {
             .await
             .map_err(|e| anyhow::anyhow!("Unable to get total balance: {e:?}"))?;
 
-        let balance = balance / 1_000_000_000_000;
+        let balance = balance / UNITS;
         let balance: i64 = balance.try_into().unwrap_or(i64::MAX);
 
         self.metrics.fee_payer_balance.set(balance);
@@ -358,24 +360,20 @@ async fn task(
             break;
         }
 
-        match this.run_inner(&mut requests, &mut responses).await {
-            Ok(()) => {
-                log::warn!("Transaction manager connection terminated, exiting...");
-                break;
+        let Err(err) = this.run_inner(&mut requests, &mut responses).await else {
+            log::warn!("Transaction manager connection terminated, exiting...");
+            break;
+        };
+
+        log::error!("Gear message sender got an error: {err:?}");
+        match this.api_provider.reconnect().await {
+            Ok(_) => {
+                log::info!("Reconnected to Gear API");
             }
 
             Err(err) => {
-                log::error!("Gear message sender got an error: {err:?}");
-                match this.api_provider.reconnect().await {
-                    Ok(()) => {
-                        log::info!("Reconnected to Gear API");
-                    }
-
-                    Err(err) => {
-                        log::error!("Failed to reconnect to Gear API: {err:?}");
-                        break;
-                    }
-                }
+                log::error!("Failed to reconnect to Gear API: {err:?}");
+                break;
             }
         }
     }