Structured logs for relayer logic

config.toml

@@ -121,6 +121,8 @@ websocket_addr = 'ws://127.0.0.1:26657/websocket'
 
 # Specify the maximum amount of time (duration) that the RPC requests should
 # take before timing out. Default: 10s (10 seconds)
+# Note: Hermes uses this parameter _only_ in `start` mode; for all other CLIs,
+# Hermes uses a large preconfigured timeout (on the order of minutes).
 rpc_timeout = '10s'
 
 # Specify the prefix used by the chain. Required

relayer-cli/src/commands.rs

@@ -5,6 +5,7 @@
 //! See the `impl Configurable` below for how to specify the path to the
 //! application's configuration file.
 
+use core::time::Duration;
 use std::path::PathBuf;
 
 use abscissa_core::{config::Override, Clap, Command, Configurable, FrameworkError, Runnable};
@@ -138,6 +139,14 @@ impl Configurable<Config> for CliCmd {
             ccfg.memo_prefix.apply_suffix(&suffix);
         }
 
+        // For all commands except for `start` Hermes retries
+        // for a prolonged period of time.
+        if !matches!(self, CliCmd::Start(_)) {
+            for c in config.chains.iter_mut() {
+                c.rpc_timeout = Duration::from_secs(120);
+            }
+        }
+
         match self {
             CliCmd::Tx(cmd) => cmd.override_config(config),
             // CliCmd::Help(cmd) => cmd.override_config(config),
@@ -146,7 +155,6 @@ impl Configurable<Config> for CliCmd {
             // CliCmd::Update(cmd) => cmd.override_config(config),
             // CliCmd::Upgrade(cmd) => cmd.override_config(config),
             // CliCmd::Start(cmd) => cmd.override_config(config),
-            // CliCmd::StartMulti(cmd) => cmd.override_config(config),
             // CliCmd::Query(cmd) => cmd.override_config(config),
             // CliCmd::Listen(cmd) => cmd.override_config(config),
             // CliCmd::Misbehaviour(cmd) => cmd.override_config(config),

relayer/Cargo.toml

@@ -60,6 +60,7 @@ anyhow = "1.0"
 semver = "1.0"
 uint = "0.9"
 humantime = "2.1.0"
+nanoid = "0.4.0"
 
 [dependencies.num-bigint]
 version = "0.4"

relayer/src/chain.rs

@@ -1,5 +1,4 @@
 use alloc::sync::Arc;
-use prost_types::Any;
 use tendermint::block::Height;
 use tokio::runtime::Runtime as TokioRuntime;
 
@@ -46,10 +45,13 @@ use crate::keyring::{KeyEntry, KeyRing};
 use crate::light_client::LightClient;
 use crate::{config::ChainConfig, event::monitor::EventReceiver};
 
+use self::tx::TrackedMsgs;
+
 pub mod cosmos;
 pub mod counterparty;
 pub mod handle;
 pub mod runtime;
+pub mod tx;
 
 #[cfg(test)]
 pub mod mock;
@@ -125,14 +127,14 @@ pub trait ChainEndpoint: Sized {
     // synchronously wait for it to be committed.
     fn send_messages_and_wait_commit(
         &mut self,
-        proto_msgs: Vec<Any>,
+        tracked_msgs: TrackedMsgs,
     ) -> Result<Vec<IbcEvent>, Error>;
 
     /// Sends one or more transactions with `msgs` to chain.
     /// Non-blocking alternative to `send_messages_and_wait_commit` interface.
     fn send_messages_and_wait_check_tx(
         &mut self,
-        proto_msgs: Vec<Any>,
+        tracked_msgs: TrackedMsgs,
     ) -> Result<Vec<TxResponse>, Error>;
 
     fn get_signer(&mut self) -> Result<Signer, Error>;

relayer/src/chain/cosmos.rs

@@ -29,7 +29,7 @@ use tendermint_rpc::{
 };
 use tokio::runtime::Runtime as TokioRuntime;
 use tonic::codegen::http::Uri;
-use tracing::{debug, error, info, trace, warn};
+use tracing::{debug, error, info, span, trace, warn, Level};
 
 use ibc::clients::ics07_tendermint::client_state::{AllowUpdate, ClientState};
 use ibc::clients::ics07_tendermint::consensus_state::ConsensusState as TMConsensusState;
@@ -95,7 +95,7 @@ use crate::{
     sdk_error::sdk_error_from_tx_sync_error_code,
 };
 
-use super::{ChainEndpoint, HealthCheck};
+use super::{tx::TrackedMsgs, ChainEndpoint, HealthCheck};
 
 mod compatibility;
 pub mod version;
@@ -952,13 +952,14 @@ impl ChainEndpoint for CosmosSdkChain {
     /// msgs in a Tx until any of the max size, max num msgs, max fee are exceeded.
     fn send_messages_and_wait_commit(
         &mut self,
-        proto_msgs: Vec<Any>,
+        tracked_msgs: TrackedMsgs,
     ) -> Result<Vec<IbcEvent>, Error> {
         crate::time!("send_messages_and_wait_commit");
-        debug!(
-            "send_messages_and_wait_commit with {} messages",
-            proto_msgs.len()
-        );
+
+        let span = span!(Level::DEBUG, "send_w", id = %tracked_msgs);
+        let _enter = span.enter();
+
+        let proto_msgs = tracked_msgs.messages();
 
         if proto_msgs.is_empty() {
             return Ok(vec![]);
@@ -1009,13 +1010,14 @@ impl ChainEndpoint for CosmosSdkChain {
 
     fn send_messages_and_wait_check_tx(
         &mut self,
-        proto_msgs: Vec<Any>,
+        tracked_msgs: TrackedMsgs,
     ) -> Result<Vec<Response>, Error> {
         crate::time!("send_messages_and_wait_check_tx");
-        debug!(
-            "send_messages_and_wait_check_tx with {} messages",
-            proto_msgs.len()
-        );
+
+        let span = span!(Level::DEBUG, "send", id = %tracked_msgs);
+        let _enter = span.enter();
+
+        let proto_msgs = tracked_msgs.messages();
 
         if proto_msgs.is_empty() {
             return Ok(vec![]);

relayer/src/chain/handle.rs

@@ -54,7 +54,7 @@ use crate::{
     keyring::KeyEntry,
 };
 
-use super::HealthCheck;
+use super::{tx::TrackedMsgs, HealthCheck};
 
 mod prod;
 pub mod requests;
@@ -111,12 +111,12 @@ pub enum ChainRequest {
     },
 
     SendMessagesAndWaitCommit {
-        proto_msgs: Vec<prost_types::Any>,
+        tracked_msgs: TrackedMsgs,
         reply_to: ReplyTo<Vec<IbcEvent>>,
     },
 
     SendMessagesAndWaitCheckTx {
-        proto_msgs: Vec<prost_types::Any>,
+        tracked_msgs: TrackedMsgs,
         reply_to: ReplyTo<Vec<tendermint_rpc::endpoint::broadcast::tx_sync::Response>>,
     },
 
@@ -350,7 +350,7 @@ 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_messages_and_wait_commit(
         &self,
-        proto_msgs: Vec<prost_types::Any>,
+        tracked_msgs: TrackedMsgs,
     ) -> Result<Vec<IbcEvent>, Error>;
 
     /// Submit messages asynchronously.
@@ -359,7 +359,7 @@ pub trait ChainHandle: Clone + Send + Sync + Serialize + Debug {
     /// returns a set of transaction hashes.
     fn send_messages_and_wait_check_tx(
         &self,
-        proto_msgs: Vec<prost_types::Any>,
+        tracked_msgs: TrackedMsgs,
     ) -> Result<Vec<tendermint_rpc::endpoint::broadcast::tx_sync::Response>, Error>;
 
     fn get_signer(&self) -> Result<Signer, Error>;

relayer/src/chain/handle/prod.rs

@@ -37,6 +37,7 @@ use ibc_proto::ibc::core::commitment::v1::MerkleProof;
 use ibc_proto::ibc::core::connection::v1::QueryClientConnectionsRequest;
 use ibc_proto::ibc::core::connection::v1::QueryConnectionsRequest;
 
+use crate::chain::tx::TrackedMsgs;
 use crate::{
     chain::handle::requests::AppVersion, chain::StatusResponse, config::ChainConfig,
     connection::ConnectionMsgType, error::Error, keyring::KeyEntry,
@@ -98,20 +99,20 @@ impl ChainHandle for ProdChainHandle {
 
     fn send_messages_and_wait_commit(
         &self,
-        proto_msgs: Vec<prost_types::Any>,
+        tracked_msgs: TrackedMsgs,
     ) -> Result<Vec<IbcEvent>, Error> {
         self.send(|reply_to| ChainRequest::SendMessagesAndWaitCommit {
-            proto_msgs,
+            tracked_msgs,
             reply_to,
         })
     }
 
     fn send_messages_and_wait_check_tx(
         &self,
-        proto_msgs: Vec<prost_types::Any>,
+        tracked_msgs: TrackedMsgs,
     ) -> Result<Vec<tendermint_rpc::endpoint::broadcast::tx_sync::Response>, Error> {
         self.send(|reply_to| ChainRequest::SendMessagesAndWaitCheckTx {
-            proto_msgs,
+            tracked_msgs,
             reply_to,
         })
     }

relayer/src/chain/mock.rs

@@ -3,7 +3,6 @@ use core::ops::Add;
 use core::time::Duration;
 
 use crossbeam_channel as channel;
-use prost_types::Any;
 use tendermint_testgen::light_block::TmLightBlock;
 use tokio::runtime::Runtime;
 
@@ -50,6 +49,7 @@ use crate::keyring::{KeyEntry, KeyRing};
 use crate::light_client::Verified;
 use crate::light_client::{mock::LightClient as MockLightClient, LightClient};
 
+use super::tx::TrackedMsgs;
 use super::HealthCheck;
 
 /// The representation of a mocked chain as the relayer sees it.
@@ -129,17 +129,20 @@ impl ChainEndpoint for MockChain {
 
     fn send_messages_and_wait_commit(
         &mut self,
-        proto_msgs: Vec<Any>,
+        tracked_msgs: TrackedMsgs,
     ) -> Result<Vec<IbcEvent>, Error> {
         // Use the ICS18Context interface to submit the set of messages.
-        let events = self.context.send(proto_msgs).map_err(Error::ics18)?;
+        let events = self
+            .context
+            .send(tracked_msgs.into())
+            .map_err(Error::ics18)?;
 
         Ok(events)
     }
 
     fn send_messages_and_wait_check_tx(
         &mut self,
-        _proto_msgs: Vec<Any>,
+        _tracked_msgs: TrackedMsgs,
     ) -> Result<Vec<tendermint_rpc::endpoint::broadcast::tx_sync::Response>, Error> {
         todo!()
     }

relayer/src/chain/runtime.rs

@@ -60,6 +60,7 @@ use crate::{
 
 use super::{
     handle::{ChainHandle, ChainRequest, ReplyTo, Subscription},
+    tx::TrackedMsgs,
     ChainEndpoint, HealthCheck,
 };
 
@@ -251,12 +252,12 @@ where
                             self.subscribe(reply_to)?
                         },
 
-                        Ok(ChainRequest::SendMessagesAndWaitCommit { proto_msgs, reply_to }) => {
-                            self.send_messages_and_wait_commit(proto_msgs, reply_to)?
+                        Ok(ChainRequest::SendMessagesAndWaitCommit { tracked_msgs, reply_to }) => {
+                            self.send_messages_and_wait_commit(tracked_msgs, reply_to)?
                         },
 
-                        Ok(ChainRequest::SendMessagesAndWaitCheckTx { proto_msgs, reply_to }) => {
-                            self.send_messages_and_wait_check_tx(proto_msgs, reply_to)?
+                        Ok(ChainRequest::SendMessagesAndWaitCheckTx { tracked_msgs, reply_to }) => {
+                            self.send_messages_and_wait_check_tx(tracked_msgs, reply_to)?
                         },
 
                         Ok(ChainRequest::Signer { reply_to }) => {
@@ -446,19 +447,19 @@ where
 
     fn send_messages_and_wait_commit(
         &mut self,
-        proto_msgs: Vec<prost_types::Any>,
+        tracked_msgs: TrackedMsgs,
         reply_to: ReplyTo<Vec<IbcEvent>>,
     ) -> Result<(), Error> {
-        let result = self.chain.send_messages_and_wait_commit(proto_msgs);
+        let result = self.chain.send_messages_and_wait_commit(tracked_msgs);
         reply_to.send(result).map_err(Error::send)
     }
 
     fn send_messages_and_wait_check_tx(
         &mut self,
-        proto_msgs: Vec<prost_types::Any>,
+        tracked_msgs: TrackedMsgs,
         reply_to: ReplyTo<Vec<tendermint_rpc::endpoint::broadcast::tx_sync::Response>>,
     ) -> Result<(), Error> {
-        let result = self.chain.send_messages_and_wait_check_tx(proto_msgs);
+        let result = self.chain.send_messages_and_wait_check_tx(tracked_msgs);
         reply_to.send(result).map_err(Error::send)
     }
 

relayer/src/chain/tx.rs

@@ -0,0 +1,41 @@
+use core::fmt;
+
+use prost_types::Any;
+
+#[derive(Debug, Clone)]
+pub struct TrackedMsgs {
+    msgs: Vec<Any>,
+    tracking_id: String,
+}
+
+impl TrackedMsgs {
+    pub fn new(msgs: Vec<Any>, tid: &str) -> Self {
+        Self {
+            msgs,
+            tracking_id: tid.into(),
+        }
+    }
+
+    pub fn messages(&self) -> &Vec<Any> {
+        &self.msgs
+    }
+
+    pub fn new_single(msg: Any, tid: &str) -> Self {
+        Self {
+            msgs: vec![msg],
+            tracking_id: tid.into(),
+        }
+    }
+}
+
+impl From<TrackedMsgs> for Vec<Any> {
+    fn from(tm: TrackedMsgs) -> Vec<Any> {
+        tm.msgs
+    }
+}
+
+impl fmt::Display for TrackedMsgs {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "{}; len={}", self.tracking_id, self.msgs.len())
+    }
+}