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

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,
+        }
+    }
+}

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<C> NymdClient<C> {
         &self.client_address.as_ref().unwrap()[0]
     }
 
+    pub async fn account_sequence(&self) -> Result<SequenceResponse, NymdError>
+    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<C> NymdClient<C> {
         Ok(self.client.get_block(None).await?.block.header.time)
     }
 
+    pub async fn get_current_block_height(&self) -> Result<Height, NymdError>
+    where
+        C: CosmWasmClient + Sync,
+    {
+        self.client.get_height().await
+    }
+
     pub async fn get_balance(&self, address: &AccountId) -> Result<Option<CosmosCoin>, NymdError>
     where
         C: CosmWasmClient + Sync,

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<C> Client<C> {
             .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<C> Client<C> {
             .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())
+                }
+            }
+        }
     }
 }

validator-api/src/rewarding/error.rs

@@ -45,3 +45,14 @@ impl From<ValidatorClientError> 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,
+        }
+    }
+}

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();