Hermes retrying mechanism for packet clearing

relayer-cli/src/commands/tx/packet.rs

@@ -1,11 +1,13 @@
-use abscissa_core::{Command, Options, Runnable};
+use abscissa_core::config::Override;
+use abscissa_core::{Command, FrameworkErrorKind, Options, Runnable};
 
 use ibc::events::IbcEvent;
 use ibc::ics24_host::identifier::{ChainId, ChannelId, PortId};
 use ibc_relayer::link::{Link, LinkParameters};
 
 use crate::cli_utils::ChainHandlePair;
 use crate::conclude::Output;
+use crate::config::Config;
 use crate::error::{Error, Kind};
 use crate::prelude::*;
 
@@ -22,6 +24,26 @@ pub struct TxRawPacketRecvCmd {
 
     #[options(free, required, help = "identifier of the source channel")]
     src_channel_id: ChannelId,
+
+    #[options(
+        help = "use the given signing key (default: `key_name` config)",
+        short = "k"
+    )]
+    key: Option<String>,
+}
+
+impl Override<Config> for TxRawPacketRecvCmd {
+    fn override_config(&self, mut config: Config) -> Result<Config, abscissa_core::FrameworkError> {
+        let src_chain_config = config.find_chain_mut(&self.src_chain_id).ok_or_else(|| {
+            FrameworkErrorKind::ComponentError.context("missing src chain configuration")
+        })?;
+
+        if let Some(ref key_name) = self.key {
+            src_chain_config.key_name = key_name.to_string();
+        }
+
+        Ok(config)
+    }
 }
 
 impl Runnable for TxRawPacketRecvCmd {

relayer/src/chain/cosmos.rs

@@ -92,9 +92,10 @@ mod retry_strategy {
     use crate::util::retry::Fixed;
     use std::time::Duration;
 
-    pub fn wait_for_block_commits() -> impl Iterator<Item = Duration> {
-        // The total time should be higher than the full node timeout which defaults to 10sec.
-        Fixed::from_millis(300).take(40) // 12 seconds
+    pub fn wait_for_block_commits(max_total_wait: Duration) -> impl Iterator<Item = Duration> {
+        let backoff_millis = 300; // The periodic backoff
+        let count: usize = (max_total_wait.as_millis() / backoff_millis as u128) as usize;
+        Fixed::from_millis(backoff_millis).take(count)
     }
 }
 
@@ -495,53 +496,55 @@ impl CosmosSdkChain {
         thread::sleep(Duration::from_millis(200));
 
         let start = Instant::now();
+        let result = retry_with_index(
+            retry_strategy::wait_for_block_commits(self.config.rpc_timeout),
+            |index| {
+                if all_tx_results_found(&tx_sync_results) {
+                    trace!(
+                        "wait_for_block_commits: retrieved {} tx results after {} tries ({}ms)",
+                        tx_sync_results.len(),
+                        index,
+                        start.elapsed().as_millis()
+                    );
 
-        let result = retry_with_index(retry_strategy::wait_for_block_commits(), |index| {
-            if all_tx_results_found(&tx_sync_results) {
-                trace!(
-                    "wait_for_block_commits: retrieved {} tx results after {} tries ({}ms)",
-                    tx_sync_results.len(),
-                    index,
-                    start.elapsed().as_millis()
-                );
-
-                // All transactions confirmed
-                return RetryResult::Ok(());
-            }
+                    // All transactions confirmed
+                    return RetryResult::Ok(());
+                }
 
-            for TxSyncResult { response, events } in tx_sync_results.iter_mut() {
-                // If this transaction was not committed, determine whether it was because it failed
-                // or because it hasn't been committed yet.
-                if empty_event_present(&events) {
-                    // If the transaction failed, replace the events with an error,
-                    // so that we don't attempt to resolve the transaction later on.
-                    if response.code.value() != 0 {
-                        *events = vec![IbcEvent::ChainError(format!(
-                            "deliver_tx for Tx hash {} reports error: code={:?}, log={:?}",
-                            response.hash, response.code, response.log
+                for TxSyncResult { response, events } in tx_sync_results.iter_mut() {
+                    // If this transaction was not committed, determine whether it was because it failed
+                    // or because it hasn't been committed yet.
+                    if empty_event_present(&events) {
+                        // If the transaction failed, replace the events with an error,
+                        // so that we don't attempt to resolve the transaction later on.
+                        if response.code.value() != 0 {
+                            *events = vec![IbcEvent::ChainError(format!(
+                            "deliver_tx on chain {} for Tx hash {} reports error: code={:?}, log={:?}",
+                            self.id(), response.hash, response.code, response.log
                         ))];
 
-                    // Otherwise, try to resolve transaction hash to the corresponding events.
-                    } else if let Ok(events_per_tx) =
-                        self.query_txs(QueryTxRequest::Transaction(QueryTxHash(response.hash)))
-                    {
-                        // If we get events back, progress was made, so we replace the events
-                        // with the new ones. in both cases we will check in the next iteration
-                        // whether or not the transaction was fully committed.
-                        if !events_per_tx.is_empty() {
-                            *events = events_per_tx;
+                        // Otherwise, try to resolve transaction hash to the corresponding events.
+                        } else if let Ok(events_per_tx) =
+                            self.query_txs(QueryTxRequest::Transaction(QueryTxHash(response.hash)))
+                        {
+                            // If we get events back, progress was made, so we replace the events
+                            // with the new ones. in both cases we will check in the next iteration
+                            // whether or not the transaction was fully committed.
+                            if !events_per_tx.is_empty() {
+                                *events = events_per_tx;
+                            }
                         }
                     }
                 }
-            }
-            RetryResult::Retry(index)
-        });
+                RetryResult::Retry(index)
+            },
+        );
 
         match result {
             // All transactions confirmed
             Ok(()) => Ok(tx_sync_results),
             // Did not find confirmation
-            Err(_) => Err(Kind::TxNoConfirmation.into()),
+            Err(_) => Err(Kind::TxNoConfirmation(format!("from chain {}", self.id())).into()),
         }
     }
 }

relayer/src/error.rs

@@ -78,6 +78,10 @@ pub enum Kind {
     #[error("Failed to create client state")]
     BuildClientStateFailure,
 
+    /// Tx failure due to lack of confirmation
+    #[error("Failed Tx: no confirmation {0}")]
+    TxNoConfirmation(String),
+
     /// Gas estimate from simulated Tx exceeds the maximum configured
     #[error("{chain_id} gas estimate {estimated_gas} from simulated Tx exceeds the maximum configured {max_gas}")]
     TxSimulateGasEstimateExceeded {
@@ -86,10 +90,6 @@ pub enum Kind {
         max_gas: u64,
     },
 
-    /// Tx failure for lack of confirmation
-    #[error("Failed Tx: no confirmation")]
-    TxNoConfirmation,
-
     /// Create client failure
     #[error("Failed to create client {0}")]
     CreateClient(String),

relayer/src/link.rs

@@ -5,6 +5,7 @@ use std::fmt;
 use std::thread;
 use std::time::Instant;
 
+use itertools::Itertools;
 use prost_types::Any;
 use thiserror::Error;
 use tracing::{debug, error, info, trace, warn};
@@ -397,7 +398,6 @@ impl RelayPath {
     }
 
     fn relay_pending_packets(&mut self, height: Height) -> Result<(), LinkError> {
-        info!("[{}] clearing old packets", self);
         for _ in 0..MAX_RETRIES {
             if self
                 .build_recv_packet_and_timeout_msgs(Some(height))
@@ -412,7 +412,7 @@ impl RelayPath {
 
     /// Should not run more than once per execution.
     pub fn clear_packets(&mut self, above_height: Height) -> Result<(), LinkError> {
-        if self.clear_packets {
+        if self.clear_packets || above_height.revision_height % 100 == 0 {
             info!(
                 "[{}] clearing pending packets from events before height {}",
                 self, above_height
@@ -423,7 +423,10 @@ impl RelayPath {
 
             self.relay_pending_packets(clear_height)?;
 
-            info!("[{}] finished clearing pending packets", self);
+            info!(
+                "[{}] finished scheduling the clearing of pending packets",
+                self
+            );
 
             self.clear_packets = false;
         }
@@ -721,7 +724,8 @@ impl RelayPath {
     }
 
     /// Sends a transaction to the chain targeted by the operational data `odata`.
-    /// If the transaction generates an error, returns the error as well as  `LinkError::SendError` if  input events if a sending failure occurs.
+    /// If the transaction generates an error, returns the error as well as  `LinkError::SendError`
+    /// if  input events if a sending failure occurs.
     /// Returns the events generated by the target chain upon success.
     fn send_from_operational_data(
         &mut self,
@@ -921,19 +925,13 @@ impl RelayPath {
         if packet_commitments.is_empty() {
             return Ok((events_result, query_height));
         }
-        let commit_sequences = packet_commitments.iter().map(|p| p.sequence).collect();
-        debug!(
-            "[{}] packets that still have commitments on {}: {:?}",
-            self,
-            self.src_chain().id(),
-            commit_sequences
-        );
+        let commit_sequences: Vec<u64> = packet_commitments.iter().map(|p| p.sequence).collect();
 
         // Get the packets that have not been received on destination chain
         let request = QueryUnreceivedPacketsRequest {
             port_id: self.dst_port_id().to_string(),
             channel_id: self.dst_channel_id()?.to_string(),
-            packet_commitment_sequences: commit_sequences,
+            packet_commitment_sequences: commit_sequences.clone(),
         };
 
         let sequences: Vec<Sequence> = self
@@ -943,17 +941,25 @@ impl RelayPath {
             .map(From::from)
             .collect();
 
+        if sequences.is_empty() {
+            return Ok((events_result, query_height));
+        }
+
         debug!(
-            "[{}] recv packets to send out to {} of the ones with commitments on source {}: {:?}",
+            "[{}] packets that still have commitments on {}: {} (first 10 shown here; total={})",
             self,
-            self.dst_chain().id(),
             self.src_chain().id(),
-            sequences
+            commit_sequences.iter().take(10).join(", "),
+            commit_sequences.len()
         );
 
-        if sequences.is_empty() {
-            return Ok((events_result, query_height));
-        }
+        debug!(
+            "[{}] recv packets to send out to {} of the ones with commitments on source {}: {} (first 10 shown here; total={})",
+            self,
+            self.dst_chain().id(),
+            self.src_chain().id(),
+            sequences.iter().take(10).join(", "), sequences.len()
+        );
 
         let query = QueryTxRequest::Packet(QueryPacketEventDataRequest {
             event_id: IbcEventType::SendPacket,
@@ -972,8 +978,17 @@ impl RelayPath {
             let send_event = downcast!(event => IbcEvent::SendPacket)
                 .ok_or_else(|| LinkError::Failed("unexpected query tx response".into()))?;
             packet_sequences.push(send_event.packet.sequence);
+            if packet_sequences.len() > 10 {
+                // Enough to print the first 10
+                break;
+            }
         }
-        debug!("[{}] received from query_txs {:?}", self, packet_sequences);
+        info!(
+            "[{}] found unprocessed SendPacket events for {:?} (first 10 shown here; total={})",
+            self,
+            packet_sequences,
+            events_result.len()
+        );
 
         Ok((events_result, query_height))
     }
@@ -1006,18 +1021,13 @@ impl RelayPath {
             return Ok((events_result, query_height));
         }
 
-        let acked_sequences = acks_on_source.iter().map(|p| p.sequence).collect();
-        debug!(
-            "[{}] packets that have acknowledgments on {} {:?}",
-            self,
-            self.src_chain().id(),
-            acked_sequences
-        );
+        let mut acked_sequences: Vec<u64> = acks_on_source.iter().map(|p| p.sequence).collect();
+        acked_sequences.sort_unstable();
 
         let request = QueryUnreceivedAcksRequest {
             port_id: self.dst_port_id().to_string(),
             channel_id: dst_channel_id.to_string(),
-            packet_ack_sequences: acked_sequences,
+            packet_ack_sequences: acked_sequences.clone(),
         };
 
         let sequences: Vec<Sequence> = self
@@ -1027,17 +1037,27 @@ impl RelayPath {
             .into_iter()
             .map(From::from)
             .collect();
+
+        if sequences.is_empty() {
+            return Ok((events_result, query_height));
+        }
+
         debug!(
-            "[{}] ack packets to send out to {} of the ones with acknowledgments on {}: {:?}",
+            "[{}] packets that have acknowledgments on {}: [{:?}..{:?}] (total={})",
             self,
-            self.dst_chain().id(),
             self.src_chain().id(),
-            sequences
+            acked_sequences.first(),
+            acked_sequences.last(),
+            acked_sequences.len()
         );
 
-        if sequences.is_empty() {
-            return Ok((events_result, query_height));
-        }
+        debug!(
+            "[{}] ack packets to send out to {} of the ones with acknowledgments on {}: {} (first 10 shown here; total={})",
+            self,
+            self.dst_chain().id(),
+            self.src_chain().id(),
+            sequences.iter().take(10).join(", "), sequences.len()
+        );
 
         events_result = self
             .src_chain()
@@ -1057,8 +1077,12 @@ impl RelayPath {
             let write_ack_event = downcast!(event => IbcEvent::WriteAcknowledgement)
                 .ok_or_else(|| LinkError::Failed("unexpected query tx response".into()))?;
             packet_sequences.push(write_ack_event.packet.sequence);
+            if packet_sequences.len() > 10 {
+                // Enough to print the first 10
+                break;
+            }
         }
-        info!("[{}] received from query_txs {:?}", self, packet_sequences);
+        info!("[{}] found unprocessed WriteAcknowledgement events for {:?} (first 10 shown here; total={})", self, packet_sequences, events_result.len());
 
         Ok((events_result, query_height))
     }