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;
@@ -96,7 +96,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;
@@ -1050,13 +1050,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_tx_commit", id = %tracked_msgs.tracking_id()).entered();
+
+ let proto_msgs = tracked_msgs.messages();
 
  if proto_msgs.is_empty() {
  return Ok(vec![]);
@@ -1107,13 +1108,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_tx_check", id = %tracked_msgs.tracking_id());
+ 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,43 @@
+use prost_types::Any;
+
+/// A wrapper over a vector of proto-encoded messages
+/// (`Vec<Any>`), which has an associated tracking
+/// number.
+///
+/// A [`TrackedMsgs`] correlates with a [`TrackedEvents`]
+/// by sharing the same `tracking_id`.
+#[derive(Debug, Clone)]
+pub struct TrackedMsgs {
+ msgs: Vec<Any>,
+ tracking_id: String,
+}
+
+impl TrackedMsgs {
+ pub fn new(msgs: Vec<Any>, tid: impl Into<String>) -> Self {
+ Self {
+ msgs,
+ tracking_id: tid.into(),
+ }
+ }
+
+ pub fn new_single(msg: Any, tid: impl Into<String>) -> Self {
+ Self {
+ msgs: vec![msg],
+ tracking_id: tid.into(),
+ }
+ }
+
+ pub fn messages(&self) -> &Vec<Any> {
+ &self.msgs
+ }
+
+ pub fn tracking_id(&self) -> &str {
+ &self.tracking_id
+ }
+}
+
+impl From<TrackedMsgs> for Vec<Any> {
+ fn from(tm: TrackedMsgs) -> Vec<Any> {
+ tm.msgs
+ }
+}