Asynchronous tx confirmations

relayer/src/chain.rs

@@ -109,6 +109,13 @@ pub trait ChainEndpoint: Sized {
     /// Sends one or more transactions with `msgs` to chain.
     fn send_msgs(&mut self, proto_msgs: Vec<Any>) -> Result<Vec<IbcEvent>, Error>;
 
+    /// Sends one or more transactions with `msgs` to chain.
+    /// Non-blocking alternative to `send_msgs` interface.
+    fn submit_msgs(
+        &mut self,
+        proto_msgs: Vec<Any>,
+    ) -> Result<Vec<tendermint_rpc::endpoint::broadcast::tx_sync::Response>, Error>;
+
     fn get_signer(&mut self) -> Result<Signer, Error>;
 
     fn get_key(&mut self) -> Result<KeyEntry, Error>;

relayer/src/chain/cosmos.rs

@@ -627,7 +627,11 @@ impl CosmosSdkChain {
             .map(|res| res.response.hash.to_string())
             .join(", ");
 
-        debug!("[{}] waiting for commit of block(s) {}", self.id(), hashes);
+        warn!(
+            "[{}] waiting for commit of tx hashes(s) {}",
+            self.id(),
+            hashes
+        );
 
         // Wait a little bit initially
         thread::sleep(Duration::from_millis(200));
@@ -655,6 +659,7 @@ impl CosmosSdkChain {
                     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.
+                        // !! TODO(Adi): This check must be part of `submit_msgs` method!!
                         if response.code.value() != 0 {
                             *events = vec![IbcEvent::ChainError(format!(
                             "deliver_tx on chain {} for Tx hash {} reports error: code={:?}, log={:?}",
@@ -839,6 +844,38 @@ impl ChainEndpoint for CosmosSdkChain {
         Ok(events)
     }
 
+    fn submit_msgs(&mut self, proto_msgs: Vec<Any>) -> Result<Vec<Response>, Error> {
+        crate::time!("submit_msgs");
+
+        if proto_msgs.is_empty() {
+            return Ok(vec![]);
+        }
+        let mut responses = vec![];
+
+        let mut n = 0;
+        let mut size = 0;
+        let mut msg_batch = vec![];
+        for msg in proto_msgs.iter() {
+            msg_batch.push(msg.clone());
+            let mut buf = Vec::new();
+            prost::Message::encode(msg, &mut buf).unwrap();
+            n += 1;
+            size += buf.len();
+            if n >= self.max_msg_num() || size >= self.max_tx_size() {
+                // Send the tx and enqueue the resulting response
+                responses.push(self.send_tx(msg_batch)?);
+                n = 0;
+                size = 0;
+                msg_batch = vec![];
+            }
+        }
+        if !msg_batch.is_empty() {
+            responses.push(self.send_tx(msg_batch)?);
+        }
+
+        Ok(responses)
+    }
+
     /// Get the account for the signer
     fn get_signer(&mut self) -> Result<Signer, Error> {
         crate::time!("get_signer");

relayer/src/chain/handle.rs

@@ -105,6 +105,11 @@ pub enum ChainRequest {
         reply_to: ReplyTo<Vec<IbcEvent>>,
     },
 
+    SubmitMsgs {
+        proto_msgs: Vec<prost_types::Any>,
+        reply_to: ReplyTo<Vec<tendermint_rpc::endpoint::broadcast::tx_sync::Response>>,
+    },
+
     Signer {
         reply_to: ReplyTo<Signer>,
     },
@@ -316,6 +321,15 @@ pub trait ChainHandle: Clone + Send + Sync + Serialize + Debug {
     /// and return the list of events emitted by the chain after the transaction was committed.
     fn send_msgs(&self, proto_msgs: Vec<prost_types::Any>) -> Result<Vec<IbcEvent>, Error>;
 
+    /// Submit messages asynchronously.
+    /// Does not block waiting on the chain to produce the
+    /// resulting events. Instead of events, this method
+    /// returns a set of transaction hashes.
+    fn submit_msgs(
+        &self,
+        proto_msgs: Vec<prost_types::Any>,
+    ) -> Result<Vec<tendermint_rpc::endpoint::broadcast::tx_sync::Response>, Error>;
+
     fn get_signer(&self) -> Result<Signer, Error>;
 
     fn get_key(&self) -> Result<KeyEntry, Error>;

relayer/src/chain/handle/prod.rs

@@ -95,6 +95,16 @@ impl ChainHandle for ProdChainHandle {
         })
     }
 
+    fn submit_msgs(
+        &self,
+        proto_msgs: Vec<prost_types::Any>,
+    ) -> Result<Vec<tendermint_rpc::endpoint::broadcast::tx_sync::Response>, Error> {
+        self.send(|reply_to| ChainRequest::SubmitMsgs {
+            proto_msgs,
+            reply_to,
+        })
+    }
+
     fn get_signer(&self) -> Result<Signer, Error> {
         self.send(|reply_to| ChainRequest::Signer { reply_to })
     }

relayer/src/chain/mock.rs

@@ -116,6 +116,13 @@ impl ChainEndpoint for MockChain {
         Ok(events)
     }
 
+    fn submit_msgs(
+        &mut self,
+        _proto_msgs: Vec<Any>,
+    ) -> Result<Vec<tendermint_rpc::endpoint::broadcast::tx_sync::Response>, Error> {
+        todo!()
+    }
+
     fn get_signer(&mut self) -> Result<Signer, Error> {
         Ok(get_dummy_account_id())
     }

relayer/src/chain/runtime.rs

@@ -203,6 +203,10 @@ where
                             self.send_msgs(proto_msgs, reply_to)?
                         },
 
+                        Ok(ChainRequest::SubmitMsgs { proto_msgs, reply_to }) => {
+                            self.submit_msgs(proto_msgs, reply_to)?
+                        },
+
                         Ok(ChainRequest::Signer { reply_to }) => {
                             self.get_signer(reply_to)?
                         }
@@ -372,6 +376,18 @@ where
         Ok(())
     }
 
+    fn submit_msgs(
+        &mut self,
+        proto_msgs: Vec<prost_types::Any>,
+        reply_to: ReplyTo<Vec<tendermint_rpc::endpoint::broadcast::tx_sync::Response>>,
+    ) -> Result<(), Error> {
+        let result = self.chain.submit_msgs(proto_msgs);
+
+        reply_to.send(result).map_err(Error::send)?;
+
+        Ok(())
+    }
+
     fn query_latest_height(&self, reply_to: ReplyTo<Height>) -> Result<(), Error> {
         let latest_height = self.chain.query_latest_height();
 

relayer/src/link.rs

@@ -15,7 +15,11 @@ use crate::link::relay_path::RelayPath;
 pub mod error;
 mod operational_data;
 mod relay_path;
+mod relay_sender;
 mod relay_summary;
+mod tx_hashes;
+mod unconfirmed;
+use tx_hashes::TxHashes;
 
 // Re-export the telemetries summary
 pub use relay_summary::RelaySummary;
@@ -150,28 +154,34 @@ impl<ChainA: ChainHandle, ChainB: ChainHandle> Link<ChainA, ChainB> {
         Ok(Link::new(channel))
     }
 
+    /// Implements the `packet-recv` CLI
     pub fn build_and_send_recv_packet_messages(&mut self) -> Result<Vec<IbcEvent>, LinkError> {
         self.a_to_b.build_recv_packet_and_timeout_msgs(None)?;
 
         let mut results = vec![];
 
         // Block waiting for all of the scheduled data (until `None` is returned)
         while let Some(odata) = self.a_to_b.fetch_scheduled_operational_data() {
-            let mut last_res = self.a_to_b.relay_from_operational_data(odata)?;
+            let mut last_res = self
+                .a_to_b
+                .relay_from_operational_data::<relay_sender::SyncSender>(odata)?;
             results.append(&mut last_res.events);
         }
 
         Ok(results)
     }
 
+    /// Implements the `packet-ack` CLI
     pub fn build_and_send_ack_packet_messages(&mut self) -> Result<Vec<IbcEvent>, LinkError> {
         self.a_to_b.build_packet_ack_msgs(None)?;
 
         let mut results = vec![];
 
         // Block waiting for all of the scheduled data
         while let Some(odata) = self.a_to_b.fetch_scheduled_operational_data() {
-            let mut last_res = self.a_to_b.relay_from_operational_data(odata)?;
+            let mut last_res = self
+                .a_to_b
+                .relay_from_operational_data::<relay_sender::SyncSender>(odata)?;
             results.append(&mut last_res.events);
         }
 

relayer/src/link/operational_data.rs

@@ -76,10 +76,8 @@ impl OperationalData {
             return Ok(vec![]);
         }
 
-        let mut msgs: Vec<Any> = self.batch.iter().map(|gm| gm.msg.clone()).collect();
-
         // For zero delay we prepend the client update msgs.
-        if relay_path.zero_delay() {
+        let client_update_msg = if relay_path.zero_delay() {
             let update_height = self.proofs_height.increment();
 
             info!(
@@ -98,9 +96,16 @@ impl OperationalData {
                 }
             };
 
-            if let Some(client_update) = client_update_opt.pop() {
-                msgs.insert(0, client_update);
-            }
+            client_update_opt.pop()
+        } else {
+            None
+        };
+
+        let mut msgs: Vec<Any> = self.batch.iter().map(|gm| gm.msg.clone()).collect();
+        if let Some(client_update) = client_update_msg {
+            // SAFETY: inserting at position `0` should not panic because
+            // we already checked that the `batch` vector is non-empty.
+            msgs.insert(0, client_update);
         }
 
         info!(