From 7caac334f4273689c3e12f27648bdbba57c107cb Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?J=C4=99drzej=20Stuczy=C5=84ski?= Date: Mon, 11 Oct 2021 10:48:59 +0100 Subject: [PATCH] Feature/resending rewards on timeout (#810) * Additional methods on nymd client * Checking for time out errors * Attempting retransmission on suspected time out * Don't report error if tx is a mempool duplicate --- .../validator-client/src/nymd/error.rs | 51 +++++++++- .../validator-client/src/nymd/mod.rs | 18 +++- validator-api/src/nymd_client.rs | 98 +++++++++++++++---- validator-api/src/rewarding/error.rs | 11 +++ validator-api/src/rewarding/mod.rs | 32 ++++-- 5 files changed, 178 insertions(+), 32 deletions(-) diff --git a/common/client-libs/validator-client/src/nymd/error.rs b/common/client-libs/validator-client/src/nymd/error.rs index 01f730f477..a9a709b955 100644 --- a/common/client-libs/validator-client/src/nymd/error.rs +++ b/common/client-libs/validator-client/src/nymd/error.rs @@ -3,10 +3,12 @@ use crate::nymd::cosmwasm_client::types::ContractCodeId; use cosmrs::tendermint::block; -use cosmrs::{bip32, rpc, tx, AccountId}; +use cosmrs::{bip32, tx, AccountId}; use std::io; use thiserror::Error; +pub use cosmrs::rpc::error::{Code, Error as TendermintRpcError}; + #[derive(Debug, Error)] pub enum NymdError { #[error("No contract address is available to perform the call")] @@ -31,7 +33,7 @@ pub enum NymdError { InvalidTxHash(String), #[error("There was an issue with a tendermint RPC request - {0}")] - TendermintError(#[from] rpc::Error), + TendermintError(#[from] TendermintRpcError), #[error("There was an issue when attempting to serialize data")] SerializationError(String), @@ -98,3 +100,48 @@ pub enum NymdError { #[error("The provided gas price is malformed")] MalformedGasPrice, } + +impl NymdError { + pub fn is_tendermint_timeout(&self) -> bool { + match &self { + NymdError::TendermintError(tm_err) => { + if tm_err.code() == Code::InternalError { + // 0.34 (and earlier) versions of tendermint seemed to be using phrase "timed out waiting ..." + // (https://github.com/tendermint/tendermint/blob/v0.34.13/rpc/core/mempool.go#L124) + // while 0.35+ has "timeout waiting for ..." + // https://github.com/tendermint/tendermint/blob/v0.35.0-rc3/internal/rpc/core/mempool.go#L99 + // note that as of the time of writing this comment (08.10.2021), the most recent version + // of cosmos-sdk (v0.44.1) uses tendermint 0.34.13 + if let Some(data) = tm_err.data() { + data.contains("timed out") || data.contains("timeout") + } else { + false + } + } else { + false + } + } + _ => false, + } + } + + pub fn is_tendermint_duplicate(&self) -> bool { + match &self { + NymdError::TendermintError(tm_err) => { + if tm_err.code() == Code::InternalError { + // this particular error message seems to be unchanged between 0.34 and newer versions + // https://github.com/tendermint/tendermint/blob/v0.34.13/mempool/errors.go#L10 + // https://github.com/tendermint/tendermint/blob/v0.35.0-rc3/types/mempool.go#L10 + if let Some(data) = tm_err.data() { + data.contains("tx already exists in cache") + } else { + false + } + } else { + false + } + } + _ => false, + } + } +} diff --git a/common/client-libs/validator-client/src/nymd/mod.rs b/common/client-libs/validator-client/src/nymd/mod.rs index c05d77a51f..2ec670fb6e 100644 --- a/common/client-libs/validator-client/src/nymd/mod.rs +++ b/common/client-libs/validator-client/src/nymd/mod.rs @@ -4,14 +4,13 @@ use crate::nymd::cosmwasm_client::signing_client; use crate::nymd::cosmwasm_client::types::{ ChangeAdminResult, ContractCodeId, ExecuteResult, InstantiateOptions, InstantiateResult, - MigrateResult, UploadMeta, UploadResult, + MigrateResult, SequenceResponse, UploadMeta, UploadResult, }; use crate::nymd::error::NymdError; use crate::nymd::fee_helpers::Operation; use crate::nymd::wallet::DirectSecp256k1HdWallet; use cosmrs::rpc::endpoint::broadcast; use cosmrs::rpc::{Error as TendermintRpcError, HttpClientUrl}; - use cosmwasm_std::Coin; use mixnet_contract::{ Addr, Delegation, ExecuteMsg, Gateway, GatewayOwnershipResponse, IdentityKey, @@ -28,6 +27,7 @@ pub use crate::nymd::cosmwasm_client::client::CosmWasmClient; pub use crate::nymd::cosmwasm_client::signing_client::SigningCosmWasmClient; pub use crate::nymd::gas_price::GasPrice; pub use cosmrs::rpc::HttpClient as QueryNymdClient; +pub use cosmrs::tendermint::block::Height; pub use cosmrs::tendermint::Time as TendermintTime; pub use cosmrs::tx::{Fee, Gas}; pub use cosmrs::Coin as CosmosCoin; @@ -155,6 +155,13 @@ impl NymdClient { &self.client_address.as_ref().unwrap()[0] } + pub async fn account_sequence(&self) -> Result + where + C: SigningCosmWasmClient + Sync, + { + self.client.get_sequence(self.address()).await + } + pub fn get_fee(&self, operation: Operation) -> Fee { let gas_limit = self.custom_gas_limits.get(&operation).cloned(); operation.determine_fee(&self.gas_price, gas_limit) @@ -171,6 +178,13 @@ impl NymdClient { Ok(self.client.get_block(None).await?.block.header.time) } + pub async fn get_current_block_height(&self) -> Result + where + C: CosmWasmClient + Sync, + { + self.client.get_height().await + } + pub async fn get_balance(&self, address: &AccountId) -> Result, NymdError> where C: CosmWasmClient + Sync, diff --git a/validator-api/src/nymd_client.rs b/validator-api/src/nymd_client.rs index 870b0194ac..1a449abb8d 100644 --- a/validator-api/src/nymd_client.rs +++ b/validator-api/src/nymd_client.rs @@ -10,7 +10,9 @@ use crate::rewarding::{ use config::defaults::DEFAULT_VALIDATOR_API_PORT; use mixnet_contract::{Delegation, ExecuteMsg, GatewayBond, IdentityKey, MixNodeBond, StateParams}; use std::sync::Arc; +use std::time::Duration; use tokio::sync::RwLock; +use tokio::time::sleep; use validator_client::nymd::{ CosmWasmClient, Fee, QueryNymdClient, SigningCosmWasmClient, SigningNymdClient, TendermintTime, }; @@ -193,16 +195,46 @@ impl Client { .get_mixnet_contract_address() .ok_or(RewardingError::UnspecifiedContractAddress)?; - // technically we don't require a write lock here, however, we really don't want to be executing - // multiple blocks concurrently as one of them WILL fail due to incorrect sequence number - self.0 - .write() - .await - .nymd - .execute_multiple(&contract, msgs, fee, memo) - .await?; + // grab the write lock here so we're sure nothing else is executing anything on the contract + // in the meantime + // however, we're not 100% guarded against everything + // for example somebody might have taken the mnemonic used by the validator + // and sent a transaction manually using the same account. The sequence number + // would have gotten incremented, yet the rewarding transaction might have actually not + // been included in the block. sadly we can't do much about that. + let client_guard = self.0.write().await; + let pre_sequence = client_guard.nymd.account_sequence().await?; - Ok(()) + let res = client_guard + .nymd + .execute_multiple(&contract, msgs.clone(), fee.clone(), memo.clone()) + .await; + + match res { + Ok(_) => Ok(()), + Err(err) => { + if err.is_tendermint_timeout() { + // wait until we're sure we're into the next block (remember we're holding the lock) + sleep(Duration::from_secs(11)).await; + let curr_sequence = client_guard.nymd.account_sequence().await?; + if curr_sequence.sequence > pre_sequence.sequence { + // unless somebody was messing around doing stuff manually in that tiny time interval + // we're good. It was a false negative. + Ok(()) + } else { + // the sequence number has not increased, meaning the transaction was not executed + // so attempt to send it again + client_guard + .nymd + .execute_multiple(&contract, msgs, fee, memo) + .await?; + Ok(()) + } + } else { + Err(err.into()) + } + } + } } pub(crate) async fn reward_gateways( @@ -231,15 +263,45 @@ impl Client { .get_mixnet_contract_address() .ok_or(RewardingError::UnspecifiedContractAddress)?; - // technically we don't require a write lock here, however, we really don't want to be executing - // multiple blocks concurrently as one of them WILL fail due to incorrect sequence number - self.0 - .write() - .await - .nymd - .execute_multiple(&contract, msgs, fee, memo) - .await?; + // grab the write lock here so we're sure nothing else is executing anything on the contract + // in the meantime + // however, we're not 100% guarded against everything + // for example somebody might have taken the mnemonic used by the validator + // and sent a transaction manually using the same account. The sequence number + // would have gotten incremented, yet the rewarding transaction might have actually not + // been included in the block. sadly we can't do much about that. + let client_guard = self.0.write().await; + let pre_sequence = client_guard.nymd.account_sequence().await?; - Ok(()) + let res = client_guard + .nymd + .execute_multiple(&contract, msgs.clone(), fee.clone(), memo.clone()) + .await; + + match res { + Ok(_) => Ok(()), + Err(err) => { + if err.is_tendermint_timeout() { + // wait until we're sure we're into the next block (remember we're holding the lock) + sleep(Duration::from_secs(11)).await; + let curr_sequence = client_guard.nymd.account_sequence().await?; + if curr_sequence.sequence > pre_sequence.sequence { + // unless somebody was messing around doing stuff manually in that tiny time interval + // we're good. It was a false negative. + Ok(()) + } else { + // the sequence number has not increased, meaning the transaction was not executed + // so attempt to send it again + client_guard + .nymd + .execute_multiple(&contract, msgs, fee, memo) + .await?; + Ok(()) + } + } else { + Err(err.into()) + } + } + } } } diff --git a/validator-api/src/rewarding/error.rs b/validator-api/src/rewarding/error.rs index e4c10dcff1..ea8597dc81 100644 --- a/validator-api/src/rewarding/error.rs +++ b/validator-api/src/rewarding/error.rs @@ -45,3 +45,14 @@ impl From for RewardingError { RewardingError::ValidatorClientError(err) } } + +impl RewardingError { + pub fn is_tendermint_duplicate(&self) -> bool { + match &self { + RewardingError::ValidatorClientError(ValidatorClientError::NymdError(nymd_err)) => { + nymd_err.is_tendermint_duplicate() + } + _ => false, + } + } +} diff --git a/validator-api/src/rewarding/mod.rs b/validator-api/src/rewarding/mod.rs index e5030567fe..09d56eada3 100644 --- a/validator-api/src/rewarding/mod.rs +++ b/validator-api/src/rewarding/mod.rs @@ -384,11 +384,17 @@ impl Rewarder { for (i, mix_chunk) in eligible_mixnodes.chunks(MAX_TO_REWARD_AT_ONCE).enumerate() { if let Err(err) = self.nymd_client.reward_mixnodes(mix_chunk).await { - error!("failed to reward mixnodes... - {}", err); - failed_chunks.push(FailedMixnodeRewardChunkDetails { - possibly_unrewarded: mix_chunk.to_vec(), - error_message: err.to_string(), - }) + // this is a super weird edge case that we didn't catch change to sequence and + // resent rewards unnecessarily, but the mempool saved us from executing it again + // however, still we want to wait until we're sure we're into the next block + if !err.is_tendermint_duplicate() { + error!("failed to reward mixnodes... - {}", err); + failed_chunks.push(FailedMixnodeRewardChunkDetails { + possibly_unrewarded: mix_chunk.to_vec(), + error_message: err.to_string(), + }); + } + sleep(Duration::from_secs(11)).await; } let rewarded = i * MAX_TO_REWARD_AT_ONCE + mix_chunk.len(); let percentage = rewarded as f32 * 100.0 / eligible_mixnodes.len() as f32; @@ -429,11 +435,17 @@ impl Rewarder { for (i, gateway_chunk) in eligible_gateways.chunks(MAX_TO_REWARD_AT_ONCE).enumerate() { if let Err(err) = self.nymd_client.reward_gateways(gateway_chunk).await { - error!("failed to reward gateways... - {}", err); - failed_chunks.push(FailedGatewayRewardChunkDetails { - possibly_unrewarded: gateway_chunk.to_vec(), - error_message: err.to_string(), - }) + // this is a super weird edge case that we didn't catch change to sequence and + // resent rewards unnecessarily, but the mempool saved us from executing it again + // however, still we want to wait until we're sure we're into the next block + if !err.is_tendermint_duplicate() { + error!("failed to reward gateways... - {}", err); + failed_chunks.push(FailedGatewayRewardChunkDetails { + possibly_unrewarded: gateway_chunk.to_vec(), + error_message: err.to_string(), + }); + } + sleep(Duration::from_secs(11)).await; } let rewarded = i * MAX_TO_REWARD_AT_ONCE + gateway_chunk.len();