Fix Hermes retrying mechanism not regenerating messages

.changelog/unreleased/bug-fixes/ibc-relayer/1792-fix-hermes-retrying-not-regenerating-msgs.md

@@ -0,0 +1 @@
+- Fixed Hermes retrying mechanism not regenerating operational data for messages ([#1792](https://github.com/informalsystems/ibc-rs/pull/1951))

relayer/src/link/pending.rs

@@ -9,7 +9,7 @@ use ibc::events::IbcEvent;
 use ibc::query::{QueryTxHash, QueryTxRequest};
 
 use crate::error::Error as RelayerError;
-use crate::link::error::LinkError;
+use crate::link::{error::LinkError, RelayPath};
 use crate::util::queue::Queue;
 use crate::{
     chain::handle::ChainHandle,
@@ -134,10 +134,11 @@ impl<Chain: ChainHandle> PendingTxs<Chain> {
     }
 
     /// Try and process one pending transaction if available.
-    pub fn process_pending(
+    pub fn process_pending<ChainA: ChainHandle, ChainB: ChainHandle>(
         &self,
         timeout: Duration,
-        resubmit: impl FnOnce(OperationalData) -> Result<AsyncReply, LinkError>,
+        relay_path: &RelayPath<ChainA, ChainB>,
+        resubmit: Option<impl FnOnce(OperationalData) -> Result<AsyncReply, LinkError>>,
     ) -> Result<Option<RelaySummary>, LinkError> {
         // We process pending transactions in a FIFO manner, so take from
         // the front of the queue.
@@ -177,16 +178,28 @@ impl<Chain: ChainHandle> PendingTxs<Chain> {
                         // relayer to resubmit the transaction to the chain again.
                         error!("timed out while confirming {}", tx_hashes);
 
-                        let resubmit_res = resubmit(pending.original_od.clone());
-
-                        match resubmit_res {
-                            Ok(reply) => {
-                                self.insert_new_pending_tx(reply, pending.original_od);
-                                Ok(None)
+                        match resubmit {
+                            Some(f) => {
+                                let new_od = relay_path
+                                    .regenerate_operational_data(pending.original_od.clone());
+                                match new_od.map(f) {
+                                    Some(Ok(reply)) => {
+                                        self.insert_new_pending_tx(reply, pending.original_od);
+                                        Ok(None)
+                                    }
+                                    Some(Err(e)) => {
+                                        self.pending_queue.push_back(pending);
+                                        Err(e)
+                                    }
+                                    None => {
+                                        // No operational data was regenerated; nothing to resubmit
+                                        Ok(None)
+                                    }
+                                }
                             }
-                            Err(e) => {
-                                self.pending_queue.push_back(pending);
-                                Err(e)
+                            None => {
+                                // The clear packet interval is 0 such that no resubmit logic was received
+                                Ok(None)
                             }
                         }
                     } else {

relayer/src/link/relay_path.rs

@@ -527,7 +527,7 @@ impl<ChainA: ChainHandle, ChainB: ChainHandle> RelayPath<ChainA, ChainB> {
     ///
     /// Side effects: may schedule a new operational data targeting the source chain, comprising
     /// new timeout messages.
-    fn regenerate_operational_data(
+    pub(crate) fn regenerate_operational_data(
         &self,
         initial_odata: OperationalData,
     ) -> Option<OperationalData> {
@@ -1227,45 +1227,55 @@ impl<ChainA: ChainHandle, ChainB: ChainHandle> RelayPath<ChainA, ChainB> {
         Ok(())
     }
 
-    pub fn process_pending_txs(&self) -> RelaySummary {
+    pub fn process_pending_txs(&self, clear_interval: u64) -> RelaySummary {
         if !self.confirm_txes {
             return RelaySummary::empty();
         }
 
-        let mut summary_src = self.process_pending_txs_src().unwrap_or_else(|e| {
-            error!("error processing pending events in source chain: {}", e);
-            RelaySummary::empty()
-        });
+        let mut summary_src = self
+            .process_pending_txs_src(clear_interval)
+            .unwrap_or_else(|e| {
+                error!("error processing pending events in source chain: {}", e);
+                RelaySummary::empty()
+            });
 
-        let summary_dst = self.process_pending_txs_dst().unwrap_or_else(|e| {
-            error!(
-                "error processing pending events in destination chain: {}",
-                e
-            );
-            RelaySummary::empty()
-        });
+        let summary_dst = self
+            .process_pending_txs_dst(clear_interval)
+            .unwrap_or_else(|e| {
+                error!(
+                    "error processing pending events in destination chain: {}",
+                    e
+                );
+                RelaySummary::empty()
+            });
 
         summary_src.extend(summary_dst);
         summary_src
     }
 
-    fn process_pending_txs_src(&self) -> Result<RelaySummary, LinkError> {
+    fn process_pending_txs_src(&self, clear_interval: u64) -> Result<RelaySummary, LinkError> {
+        let resubmit = if clear_interval == 0 {
+            Some(|odata| self.relay_from_operational_data::<relay_sender::AsyncSender>(odata))
+        } else {
+            None
+        };
         let res = self
             .pending_txs_src
-            .process_pending(pending::TIMEOUT, |odata| {
-                self.relay_from_operational_data::<relay_sender::AsyncSender>(odata)
-            })?
+            .process_pending(pending::TIMEOUT, self, resubmit)?
             .unwrap_or_else(RelaySummary::empty);
 
         Ok(res)
     }
 
-    fn process_pending_txs_dst(&self) -> Result<RelaySummary, LinkError> {
+    fn process_pending_txs_dst(&self, clear_interval: u64) -> Result<RelaySummary, LinkError> {
+        let resubmit = if clear_interval == 0 {
+            Some(|odata| self.relay_from_operational_data::<relay_sender::AsyncSender>(odata))
+        } else {
+            None
+        };
         let res = self
             .pending_txs_dst
-            .process_pending(pending::TIMEOUT, |odata| {
-                self.relay_from_operational_data::<relay_sender::AsyncSender>(odata)
-            })?
+            .process_pending(pending::TIMEOUT, self, resubmit)?
             .unwrap_or_else(RelaySummary::empty);
 
         Ok(res)

relayer/src/worker.rs

@@ -120,7 +120,11 @@ pub fn spawn_worker_tasks<ChainA: ChainHandle, ChainB: ChainHandle>(
                     );
                     task_handles.push(packet_task);
 
-                    let link_task = packet::spawn_packet_worker(path.clone(), link);
+                    let link_task = packet::spawn_packet_worker(
+                        path.clone(),
+                        link,
+                        packets_config.clear_interval,
+                    );
                     task_handles.push(link_task);
                     Some(cmd_tx)
                 }

relayer/src/worker/packet.rs

@@ -46,6 +46,7 @@ pub fn spawn_packet_worker<ChainA: ChainHandle, ChainB: ChainHandle>(
     path: Packet,
     // Mutex is used to prevent race condition between the packet workers
     link: Arc<Mutex<Link<ChainA, ChainB>>>,
+    clear_interval: u64,
 ) -> TaskHandle {
     let span = {
         let relay_path = &link.lock().unwrap().a_to_b;
@@ -69,7 +70,7 @@ pub fn spawn_packet_worker<ChainA: ChainHandle, ChainB: ChainHandle>(
             .execute_schedule()
             .map_err(handle_link_error_in_task)?;
 
-        let summary = relay_path.process_pending_txs();
+        let summary = relay_path.process_pending_txs(clear_interval);
 
         if !summary.is_empty() {
             trace!("Packet worker produced relay summary: {:?}", summary);
@@ -202,7 +203,7 @@ fn handle_packet_cmd<ChainA: ChainHandle, ChainB: ChainHandle>(
         }
     }
 
-    let summary = link.a_to_b.process_pending_txs();
+    let summary = link.a_to_b.process_pending_txs(clear_interval);
 
     if !summary.is_empty() {
         trace!("produced relay summary: {:?}", summary);