Forward NewBlock events to all workers to clear pending packets if needed

guide/src/tutorial_raw.md

@@ -33,7 +33,7 @@ Currently, cosmos-SDK implementation uses:
  - `connection-<n>` for connections
      - For example `connection-0` is assigned to the first connection created on `ibc-1`:
          ```shell
-         tx raw conn-init ibc-1 ibc-0 07-tendermint-0 07-tendermint-0 | jq
+         hermes tx raw conn-init ibc-1 ibc-0 07-tendermint-0 07-tendermint-0 | jq
          ```
          ```json
         {

relayer-cli/src/conclude.rs

@@ -35,7 +35,9 @@
 //! #[derive(Debug, Error)]
 //! pub enum Kind {
 //!     #[error("failed with underlying causes: {0}, {1}")]
-//!     Query(String, String),  // ...
+//!     Query(String, String),
+//!     // ...
+//! }
 //! ```
 //!
 //! - Exit from a query/tx with success:

relayer/src/event/monitor.rs

@@ -24,6 +24,12 @@ pub struct EventBatch {
     pub events: Vec<IbcEvent>,
 }
 
+impl EventBatch {
+    pub fn unwrap_or_clone(self: Arc<Self>) -> Self {
+        Arc::try_unwrap(self).unwrap_or_else(|batch| batch.as_ref().clone())
+    }
+}
+
 type SubscriptionResult = Result<tendermint_rpc::event::Event, tendermint_rpc::Error>;
 type SubscriptionStream = dyn Stream<Item = SubscriptionResult> + Send + Sync + Unpin;
 

relayer/src/link.rs

@@ -1,5 +1,5 @@
+use std::thread;
 use std::time::Duration;
-use std::{sync::Arc, thread};
 
 use prost_types::Any;
 use thiserror::Error;
@@ -411,19 +411,24 @@ impl RelayPath {
         Err(LinkError::OldPacketClearingFailed)
     }
 
-    /// Iterate through the IBC Events, build the message for each and collect all at same height.
-    /// Send a multi message transaction with these, prepending the client update
-    pub fn relay_from_events(&mut self, batch: Arc<EventBatch>) -> Result<(), LinkError> {
+    pub fn clear_packets(&mut self, height: Height) -> Result<(), LinkError> {
         if self.clear_packets {
-            self.src_height = batch
-                .height
+            self.src_height = height
                 .decrement()
                 .map_err(|e| LinkError::Failed(e.to_string()))?;
 
             self.relay_pending_packets()?;
             self.clear_packets = false;
         }
 
+        Ok(())
+    }
+
+    /// Iterate through the IBC Events, build the message for each and collect all at same height.
+    /// Send a multi message transaction with these, prepending the client update
+    pub fn relay_from_events(&mut self, batch: EventBatch) -> Result<(), LinkError> {
+        self.clear_packets(batch.height)?;
+
         // collect relevant events in self.all_events
         self.collect_events(&batch.events);
         self.adjust_events_height()?;
@@ -893,12 +898,12 @@ impl Link {
                 return Ok(());
             }
 
-            if let Ok(events) = events_a.try_recv() {
-                self.a_to_b.relay_from_events(events)?;
+            if let Ok(batch) = events_a.try_recv() {
+                self.a_to_b.relay_from_events(batch.unwrap_or_clone())?;
             }
 
-            if let Ok(events) = events_b.try_recv() {
-                self.b_to_a.relay_from_events(events)?;
+            if let Ok(batch) = events_b.try_recv() {
+                self.b_to_a.relay_from_events(batch.unwrap_or_clone())?;
             }
 
             // TODO - select over the two subscriptions