From 246eeed8644f957ba50a8b53a2eae70b5d0f7c3a Mon Sep 17 00:00:00 2001 From: ethanoroshiba Date: Tue, 3 Jun 2025 13:53:26 -0500 Subject: [PATCH 1/2] add retry logic for grpc calls --- Cargo.lock | 3 +- Cargo.toml | 1 + crates/relayer/Cargo.toml | 1 + crates/relayer/src/chain/astria.rs | 169 +++++++++++------- .../relayer/src/chain/cosmos/query/account.rs | 58 +++--- .../relayer/src/chain/cosmos/query/balance.rs | 98 +++++----- crates/relayer/src/util.rs | 1 + crates/relayer/src/util/grpc_retry.rs | 115 ++++++++++++ 8 files changed, 314 insertions(+), 132 deletions(-) create mode 100644 crates/relayer/src/util/grpc_retry.rs diff --git a/Cargo.lock b/Cargo.lock index 7970acf81e..da274e8feb 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1,6 +1,6 @@ # This file is automatically @generated by Cargo. # It is not intended for manual editing. -version = 3 +version = 4 [[package]] name = "Inflector" @@ -4559,6 +4559,7 @@ dependencies = [ "tonic", "tracing", "tracing-subscriber 0.3.19", + "tryhard", "uuid 1.13.1", ] diff --git a/Cargo.toml b/Cargo.toml index a356eb3d8d..f65268fefc 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -119,6 +119,7 @@ toml = "0.8.19" tonic = "0.12" tracing = { version = "0.1.36", default-features = false } tracing-subscriber = "0.3.14" +tryhard = "0.5.1" uint = "0.9" uuid = "1.10.5" diff --git a/crates/relayer/Cargo.toml b/crates/relayer/Cargo.toml index 987b751604..6eadd0136b 100644 --- a/crates/relayer/Cargo.toml +++ b/crates/relayer/Cargo.toml @@ -97,6 +97,7 @@ toml = { workspace = true } tonic = { workspace = true, features = ["tls", "tls-roots"] } tracing = { workspace = true } tracing-subscriber = { workspace = true, features = ["fmt", "env-filter", "json"] } +tryhard = { workspace = true } uuid = { workspace = true, features = ["v4"] } [dev-dependencies] diff --git a/crates/relayer/src/chain/astria.rs b/crates/relayer/src/chain/astria.rs index c39bc9cb6c..451c34711d 100644 --- a/crates/relayer/src/chain/astria.rs +++ b/crates/relayer/src/chain/astria.rs @@ -65,6 +65,7 @@ use crate::{ keyring::{Ed25519KeyPair, KeyRing}, light_client::{tendermint::LightClient, LightClient as _}, misbehaviour::MisbehaviourEvidence, + util::grpc_retry, }; use astria_core::crypto::VerificationKey; use astria_core::primitive::v1::Address; @@ -656,12 +657,21 @@ impl ChainEndpoint for AstriaChain { ) -> Result, Error> { use crate::{chain::utils::client_id_suffix, util::pretty::PrettyIdentifiedClientState}; - let mut client = self.ibc_client_grpc_client.clone(); + let retry_config = grpc_retry::GrpcRetryConfig::default(); - let request = tonic::Request::new(request.into()); let response = self - .block_on(client.client_states(request)) - .map_err(|e| Error::grpc_status(e, "query_clients".to_owned()))? + .block_on(grpc_retry::retry_grpc_call( + &retry_config, + "query_clients", + || async { + let mut client = self.ibc_client_grpc_client.clone(); + let request = tonic::Request::new(request.clone().into()); + client + .client_states(request) + .await + .map_err(|e| Error::grpc_status(e, "query_clients".to_owned())) + }, + ))? .into_inner(); // Deserialize into domain type @@ -694,24 +704,34 @@ impl ChainEndpoint for AstriaChain { request: QueryClientStateRequest, include_proof: IncludeProof, ) -> Result<(AnyClientState, Option), Error> { - let mut client = self.ibc_client_grpc_client.clone(); - - let mut req = ibc_proto::ibc::core::client::v1::QueryClientStateRequest { - client_id: request.client_id.to_string(), - } - .into_request(); - - let height = match request.height { - QueryHeight::Latest => 0.to_string(), - QueryHeight::Specific(h) => h.to_string(), - }; - - req.metadata_mut() - .insert("height", height.parse().expect("valid ascii")); + let retry_config = grpc_retry::GrpcRetryConfig::default(); let response = self - .block_on(client.client_state(req)) - .map_err(|e| Error::grpc_status(e, "query_client_state".to_owned()))? + .block_on(grpc_retry::retry_grpc_call( + &retry_config, + "query_client_state", + || async { + let mut client = self.ibc_client_grpc_client.clone(); + + let mut req = ibc_proto::ibc::core::client::v1::QueryClientStateRequest { + client_id: request.client_id.to_string(), + } + .into_request(); + + let height = match request.height { + QueryHeight::Latest => 0.to_string(), + QueryHeight::Specific(h) => h.to_string(), + }; + + req.metadata_mut() + .insert("height", height.parse().expect("valid ascii")); + + client + .client_state(req) + .await + .map_err(|e| Error::grpc_status(e, "query_client_state".to_owned())) + }, + ))? .into_inner(); let Some(client_state) = response.client_state else { @@ -733,26 +753,36 @@ impl ChainEndpoint for AstriaChain { request: QueryConsensusStateRequest, include_proof: IncludeProof, ) -> Result<(AnyConsensusState, Option), Error> { - let mut client = self.ibc_client_grpc_client.clone(); - - let mut req = ibc_proto::ibc::core::client::v1::QueryConsensusStateRequest { - client_id: request.client_id.to_string(), - revision_height: request.consensus_height.revision_height(), - revision_number: request.consensus_height.revision_number(), - latest_height: false, - } - .into_request(); - - let map = req.metadata_mut(); - let height_str: String = match request.query_height { - QueryHeight::Latest => 0.to_string(), - QueryHeight::Specific(h) => h.to_string(), - }; - map.insert("height", height_str.parse().expect("valid ascii string")); + let retry_config = grpc_retry::GrpcRetryConfig::default(); let response = self - .block_on(client.consensus_state(req)) - .map_err(|e| Error::grpc_status(e, "query_consensus_state".to_owned()))? + .block_on(grpc_retry::retry_grpc_call( + &retry_config, + "query_consensus_state", + || async { + let mut client = self.ibc_client_grpc_client.clone(); + + let mut req = ibc_proto::ibc::core::client::v1::QueryConsensusStateRequest { + client_id: request.client_id.to_string(), + revision_height: request.consensus_height.revision_height(), + revision_number: request.consensus_height.revision_number(), + latest_height: false, + } + .into_request(); + + let map = req.metadata_mut(); + let height_str: String = match request.query_height { + QueryHeight::Latest => 0.to_string(), + QueryHeight::Specific(h) => h.to_string(), + }; + map.insert("height", height_str.parse().expect("valid ascii string")); + + client + .consensus_state(req) + .await + .map_err(|e| Error::grpc_status(e, "query_consensus_state".to_owned())) + }, + ))? .into_inner(); let Some(consensus_state) = response.consensus_state else { @@ -780,16 +810,25 @@ impl ChainEndpoint for AstriaChain { &self, request: QueryConsensusStateHeightsRequest, ) -> Result, Error> { - let mut client = self.ibc_client_grpc_client.clone(); - - let req = ibc_proto::ibc::core::client::v1::QueryConsensusStateHeightsRequest { - client_id: request.client_id.to_string(), - pagination: Default::default(), - }; + let retry_config = grpc_retry::GrpcRetryConfig::default(); let response = self - .block_on(client.consensus_state_heights(req)) - .map_err(|e| Error::grpc_status(e, "query_consensus_state_heights".to_owned()))? + .block_on(grpc_retry::retry_grpc_call( + &retry_config, + "query_consensus_state_heights", + || async { + let mut client = self.ibc_client_grpc_client.clone(); + + let req = ibc_proto::ibc::core::client::v1::QueryConsensusStateHeightsRequest { + client_id: request.client_id.to_string(), + pagination: Default::default(), + }; + + client.consensus_state_heights(req).await.map_err(|e| { + Error::grpc_status(e, "query_consensus_state_heights".to_owned()) + }) + }, + ))? .into_inner(); let heights = response @@ -804,22 +843,32 @@ impl ChainEndpoint for AstriaChain { &self, request: QueryUpgradedClientStateRequest, ) -> Result<(AnyClientState, MerkleProof), Error> { - let mut client = self.ibc_client_grpc_client.clone(); - let mut req = - ibc_proto::ibc::core::client::v1::QueryUpgradedClientStateRequest {}.into_request(); - let map = req.metadata_mut(); - map.insert( - "height", - request - .upgrade_height - .to_string() - .parse() - .expect("valid ascii string"), - ); + let retry_config = grpc_retry::GrpcRetryConfig::default(); let response = self - .block_on(client.upgraded_client_state(req)) - .map_err(|e| Error::grpc_status(e, "query_upgraded_client_state".to_owned()))? + .block_on(grpc_retry::retry_grpc_call( + &retry_config, + "query_upgraded_client_state", + || async { + let mut client = self.ibc_client_grpc_client.clone(); + let mut req = + ibc_proto::ibc::core::client::v1::QueryUpgradedClientStateRequest {} + .into_request(); + let map = req.metadata_mut(); + map.insert( + "height", + request + .upgrade_height + .to_string() + .parse() + .expect("valid ascii string"), + ); + + client.upgraded_client_state(req).await.map_err(|e| { + Error::grpc_status(e, "query_upgraded_client_state".to_owned()) + }) + }, + ))? .into_inner(); let client_state = response diff --git a/crates/relayer/src/chain/cosmos/query/account.rs b/crates/relayer/src/chain/cosmos/query/account.rs index a483e7e268..9e1a9c587f 100644 --- a/crates/relayer/src/chain/cosmos/query/account.rs +++ b/crates/relayer/src/chain/cosmos/query/account.rs @@ -7,7 +7,7 @@ use tracing::info; use crate::chain::cosmos::types::account::Account; use crate::config::default::max_grpc_decoding_size; use crate::error::Error; -use crate::util::create_grpc_client; +use crate::util::{create_grpc_client, grpc_retry}; /// EthAccount defines an Ethermint account. /// TODO: remove when/if a canonical `EthAccount` @@ -68,35 +68,39 @@ pub async fn query_account( grpc_address: &Uri, account_address: &str, ) -> Result { - let mut client = create_grpc_client(grpc_address, QueryClient::new).await?; + let retry_config = grpc_retry::GrpcRetryConfig::default(); - client = client.max_decoding_message_size(max_grpc_decoding_size().get_bytes() as usize); + grpc_retry::retry_grpc_call(&retry_config, "query_account", || async { + let mut client = create_grpc_client(grpc_address, QueryClient::new).await?; - let request = tonic::Request::new(QueryAccountRequest { - address: account_address.to_string(), - }); + client = client.max_decoding_message_size(max_grpc_decoding_size().get_bytes() as usize); - let response = client.account(request).await; + let request = tonic::Request::new(QueryAccountRequest { + address: account_address.to_string(), + }); - // Querying for an account might fail, i.e. if the account doesn't actually exist - let resp_account = match response - .map_err(|e| Error::grpc_status(e, "query_account".to_owned()))? - .into_inner() - .account - { - Some(account) => account, - None => return Err(Error::empty_query_account(account_address.to_string())), - }; + let response = client + .account(request) + .await + .map_err(|e| Error::grpc_status(e, "query_account".to_owned()))?; - if resp_account.type_url == "/cosmos.auth.v1beta1.BaseAccount" { - Ok(BaseAccount::decode(resp_account.value.as_slice()) - .map_err(|e| Error::protobuf_decode("BaseAccount".to_string(), e))?) - } else if resp_account.type_url.ends_with(".EthAccount") { - Ok(EthAccount::decode(resp_account.value.as_slice()) - .map_err(|e| Error::protobuf_decode("EthAccount".to_string(), e))? - .base_account - .ok_or_else(Error::empty_base_account)?) - } else { - Err(Error::unknown_account_type(resp_account.type_url)) - } + // Querying for an account might fail, i.e. if the account doesn't actually exist + let resp_account = match response.into_inner().account { + Some(account) => account, + None => return Err(Error::empty_query_account(account_address.to_string())), + }; + + if resp_account.type_url == "/cosmos.auth.v1beta1.BaseAccount" { + Ok(BaseAccount::decode(resp_account.value.as_slice()) + .map_err(|e| Error::protobuf_decode("BaseAccount".to_string(), e))?) + } else if resp_account.type_url.ends_with(".EthAccount") { + Ok(EthAccount::decode(resp_account.value.as_slice()) + .map_err(|e| Error::protobuf_decode("EthAccount".to_string(), e))? + .base_account + .ok_or_else(Error::empty_base_account)?) + } else { + Err(Error::unknown_account_type(resp_account.type_url)) + } + }) + .await } diff --git a/crates/relayer/src/chain/cosmos/query/balance.rs b/crates/relayer/src/chain/cosmos/query/balance.rs index 23053fd807..ca9dd6f572 100644 --- a/crates/relayer/src/chain/cosmos/query/balance.rs +++ b/crates/relayer/src/chain/cosmos/query/balance.rs @@ -7,7 +7,7 @@ use ibc_proto::cosmos::bank::v1beta1::{ use crate::account::Balance; use crate::config::default::max_grpc_decoding_size; use crate::error::Error; -use crate::util::create_grpc_client; +use crate::util::{create_grpc_client, grpc_retry}; /// Uses the GRPC client to retrieve the account balance for a specific denom pub async fn query_balance( @@ -15,30 +15,35 @@ pub async fn query_balance( account_address: &str, denom: &str, ) -> Result { - let mut client = create_grpc_client(grpc_address, QueryClient::new).await?; + let retry_config = grpc_retry::GrpcRetryConfig::default(); - client = client.max_decoding_message_size(max_grpc_decoding_size().get_bytes() as usize); + grpc_retry::retry_grpc_call(&retry_config, "query_balance", || async { + let mut client = create_grpc_client(grpc_address, QueryClient::new).await?; - let request = tonic::Request::new(QueryBalanceRequest { - address: account_address.to_string(), - denom: denom.to_string(), - }); + client = client.max_decoding_message_size(max_grpc_decoding_size().get_bytes() as usize); - let response = client - .balance(request) - .await - .map(|r| r.into_inner()) - .map_err(|e| Error::grpc_status(e, "query_balance".to_owned()))?; + let request = tonic::Request::new(QueryBalanceRequest { + address: account_address.to_string(), + denom: denom.to_string(), + }); - // Querying for a balance might fail, i.e. if the account doesn't actually exist - let balance = response - .balance - .ok_or_else(|| Error::empty_query_account(account_address.to_string()))?; + let response = client + .balance(request) + .await + .map(|r| r.into_inner()) + .map_err(|e| Error::grpc_status(e, "query_balance".to_owned()))?; - Ok(Balance { - amount: balance.amount, - denom: balance.denom, + // Querying for a balance might fail, i.e. if the account doesn't actually exist + let balance = response + .balance + .ok_or_else(|| Error::empty_query_account(account_address.to_string()))?; + + Ok(Balance { + amount: balance.amount, + denom: balance.denom, + }) }) + .await } /// Uses the GRPC client to retrieve the account balance for all denom @@ -46,30 +51,35 @@ pub async fn query_all_balances( grpc_address: &Uri, account_address: &str, ) -> Result, Error> { - let mut client = create_grpc_client(grpc_address, QueryClient::new).await?; - - client = client.max_decoding_message_size(max_grpc_decoding_size().get_bytes() as usize); - - let request = tonic::Request::new(QueryAllBalancesRequest { - address: account_address.to_string(), - pagination: None, - resolve_denom: false, // TODO: Correctly handle resolve_denom argument - }); - - let response = client - .all_balances(request) - .await - .map(|r| r.into_inner()) - .map_err(|e| Error::grpc_status(e, "query_all_balances".to_owned()))?; - - let balances = response - .balances - .into_iter() - .map(|balance| Balance { - amount: balance.amount, - denom: balance.denom, - }) - .collect(); + let retry_config = grpc_retry::GrpcRetryConfig::default(); - Ok(balances) + grpc_retry::retry_grpc_call(&retry_config, "query_all_balances", || async { + let mut client = create_grpc_client(grpc_address, QueryClient::new).await?; + + client = client.max_decoding_message_size(max_grpc_decoding_size().get_bytes() as usize); + + let request = tonic::Request::new(QueryAllBalancesRequest { + address: account_address.to_string(), + pagination: None, + resolve_denom: false, // TODO: Correctly handle resolve_denom argument + }); + + let response = client + .all_balances(request) + .await + .map(|r| r.into_inner()) + .map_err(|e| Error::grpc_status(e, "query_all_balances".to_owned()))?; + + let balances = response + .balances + .into_iter() + .map(|balance| Balance { + amount: balance.amount, + denom: balance.denom, + }) + .collect(); + + Ok(balances) + }) + .await } diff --git a/crates/relayer/src/util.rs b/crates/relayer/src/util.rs index 9799e6cc4e..d035368fdd 100644 --- a/crates/relayer/src/util.rs +++ b/crates/relayer/src/util.rs @@ -6,6 +6,7 @@ pub mod compat_mode; pub mod debug_section; pub mod diff; pub mod excluded_sequences; +pub mod grpc_retry; pub mod iter; pub mod lock; pub mod pretty; diff --git a/crates/relayer/src/util/grpc_retry.rs b/crates/relayer/src/util/grpc_retry.rs new file mode 100644 index 0000000000..ebdf8ec376 --- /dev/null +++ b/crates/relayer/src/util/grpc_retry.rs @@ -0,0 +1,115 @@ +use core::time::Duration; +use std::future::Future; + +use tracing::{debug, warn}; + +use crate::error::Error; + +/// Default retry configuration for gRPC calls +#[derive(Debug, Clone)] +pub struct GrpcRetryConfig { + pub max_attempts: usize, + pub initial_delay: Duration, + pub max_delay: Duration, +} + +impl Default for GrpcRetryConfig { + fn default() -> Self { + Self { + max_attempts: 3, + initial_delay: Duration::from_millis(100), + max_delay: Duration::from_secs(5), + } + } +} + +/// Retry a gRPC call with fixed delay +pub async fn retry_grpc_call( + config: &GrpcRetryConfig, + operation_name: &str, + operation: F, +) -> Result +where + F: Fn() -> Fut, + Fut: Future>, +{ + let result = tryhard::retry_fn(|| async { + let result = operation().await; + + match &result { + Ok(_) => { + debug!(operation = operation_name, "gRPC call succeeded"); + } + Err(e) => { + warn!( + operation = operation_name, + error = %e, + "gRPC call failed, will retry" + ); + } + } + + result + }) + .retries(config.max_attempts.saturating_sub(1) as u32) + .await; + + result.map_err(|e| { + Error::grpc_status( + tonic::Status::unavailable(format!("gRPC retries exhausted: {}", e)), + operation_name.to_string(), + ) + }) +} + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::atomic::{AtomicUsize, Ordering}; + use std::sync::Arc; + + #[tokio::test] + async fn test_retry_success_on_first_attempt() { + let config = GrpcRetryConfig::default(); + let counter = Arc::new(AtomicUsize::new(0)); + let counter_clone = Arc::clone(&counter); + + let result = retry_grpc_call(&config, "test_operation", || async { + counter_clone.fetch_add(1, Ordering::SeqCst); + Ok::(42) + }) + .await; + + assert!(result.is_ok()); + assert_eq!(result.unwrap(), 42); + assert_eq!(counter.load(Ordering::SeqCst), 1); + } + + #[tokio::test] + async fn test_retry_success_after_failure() { + let config = GrpcRetryConfig { + max_attempts: 3, + initial_delay: Duration::from_millis(1), + max_delay: Duration::from_millis(10), + }; + let counter = Arc::new(AtomicUsize::new(0)); + let counter_clone = Arc::clone(&counter); + + let result = retry_grpc_call(&config, "test_operation", || async { + let count = counter_clone.fetch_add(1, Ordering::SeqCst); + if count < 2 { + Err(Error::grpc_status( + tonic::Status::unavailable("connection timeout"), + "test".to_string(), + )) + } else { + Ok::(42) + } + }) + .await; + + assert!(result.is_ok()); + assert_eq!(result.unwrap(), 42); + assert_eq!(counter.load(Ordering::SeqCst), 3); + } +} From 2001e03dd9bc5479d7b8e9e2fb138d5cba6ebcd0 Mon Sep 17 00:00:00 2001 From: ethanoroshiba Date: Tue, 3 Jun 2025 16:01:08 -0500 Subject: [PATCH 2/2] trigger rerun of jobs --- crates/relayer/src/util/grpc_retry.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/crates/relayer/src/util/grpc_retry.rs b/crates/relayer/src/util/grpc_retry.rs index ebdf8ec376..0b77c1fe16 100644 --- a/crates/relayer/src/util/grpc_retry.rs +++ b/crates/relayer/src/util/grpc_retry.rs @@ -44,7 +44,7 @@ where warn!( operation = operation_name, error = %e, - "gRPC call failed, will retry" + "gRPC call failed, retrying" ); } }