From IbcEvent to IbcEventWithHash

relayer-cli/src/commands/listen.rs

@@ -109,7 +109,7 @@ pub fn listen(config: &ChainConfig, filters: &[EventFilter]) -> Result<(), BoxEr
  let matching_events = batch
  .events
  .into_iter()
- .filter(|e| event_match(e, filters))
+ .filter(|e| event_match(&e.event, filters))
  .collect_vec();
 
  if matching_events.is_empty() {

relayer-cli/src/commands/misbehaviour.rs

@@ -61,7 +61,7 @@ pub fn monitor_misbehaviour(
  match event_batch {
  Ok(event_batch) => {
  for event in event_batch.events {
- match event {
+ match event.event {
  IbcEvent::UpdateClient(update) => {
  debug!("{:?}", update);
  misbehaviour_handling(
@@ -78,7 +78,7 @@ pub fn monitor_misbehaviour(
 
  IbcEvent::ClientMisbehaviour(ref _misbehaviour) => {
  // TODO - submit misbehaviour to the witnesses (our full node)
- return Ok(Some(event));
+ return Ok(Some(event.event));
  }
 
  _ => {}

relayer/src/chain/cosmos.rs

@@ -599,7 +599,7 @@ 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));

relayer/src/event/monitor.rs

@@ -24,6 +24,8 @@ use crate::util::{
  retry::{retry_count, retry_with_index, RetryResult},
  stream::group_while,
 };
+use itertools::Itertools;
+use crate::event::rpc::IbcEventWithHash;
 
 mod retry_strategy {
  use crate::util::retry::clamp_total;
@@ -72,7 +74,7 @@ pub enum Error {
 pub struct EventBatch {
  pub chain_id: ChainId,
  pub height: Height,
- pub events: Vec<IbcEvent>,
+ pub events: Vec<IbcEventWithHash>,
 }
 
 pub trait UnwrapOrClone {
@@ -375,8 +377,16 @@ impl EventMonitor {
 }
 
 /// Collect the IBC events from an RPC event
-fn collect_events(chain_id: &ChainId, event: RpcEvent) -> impl Stream<Item = (Height, IbcEvent)> {
+fn collect_events(chain_id: &ChainId, event: RpcEvent) -> impl Stream<Item = (Height, IbcEventWithHash)> {
+ let q = event.query.clone();
  let events = crate::event::rpc::get_all_events(chain_id, event).unwrap_or_default();
+
+ if chain_id.as_str() == "ibc-1" {
+ let s = events.iter().map(|(_, e)| e).join(",");
+ let hs = events.iter().map(|(h, _)| h).join(",");
+ info!("\t [0--monitor@{}] event.query {} -> vals length: {}, summary: {} <> {}", chain_id, q, events.len(), s, hs);
+ }
+
  stream::iter(events)
 }
 
@@ -406,6 +416,10 @@ fn stream_batches(
  let mut events = events.into_iter().map(|(_, e)| e).collect();
  sort_events(&mut events);
 
+ if chain_id.as_str() == "ibc-1" {
+ info!("\t [1--monitor@{}] event batch: {:?}", chain_id, events);
+ }
+
  EventBatch {
  height,
  events,
@@ -416,8 +430,8 @@ fn stream_batches(
 
 /// Sort the given events by putting the NewBlock event first,
 /// and leaving the other events as is.
-fn sort_events(events: &mut Vec<IbcEvent>) {
- events.sort_by(|a, b| match (a, b) {
+fn sort_events(events: &mut Vec<IbcEventWithHash>) {
+ events.sort_by(|a, b| match (&a.event, &b.event) {
  (IbcEvent::NewBlock(_), _) => Ordering::Less,
  _ => Ordering::Equal,
  })

relayer/src/event/rpc.rs

@@ -2,6 +2,8 @@ use std::{collections::HashMap, convert::TryFrom};
 
 use anomaly::BoxError;
 use tendermint_rpc::event::{Event as RpcEvent, EventData as RpcEventData};
+use tracing::info;
+use itertools::Itertools;
 
 use ibc::ics02_client::events::NewBlock;
 use ibc::ics02_client::height::Height;
@@ -13,14 +15,37 @@ use ibc::{
  ics04_channel::events as ChannelEvents,
 };
 
+#[derive(Debug, Clone)]
+pub struct IbcEventWithHash {
+ pub event: IbcEvent,
+ pub tx_hash: Option<String>
+}
+
+impl From<NewBlock> for IbcEventWithHash {
+ fn from(nb: NewBlock) -> Self {
+ Self {
+ event: nb.into(), // from `NewBlock` into `IbcEvent::NewBlock`
+ tx_hash: None // no hash associated here
+ }
+ }
+}
+
+impl std::fmt::Display for IbcEventWithHash {
+ fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+ write!(f, "{}%{:?}", self.event, self.tx_hash)
+ }
+}
+
 pub fn get_all_events(
  chain_id: &ChainId,
  result: RpcEvent,
-) -> Result<Vec<(Height, IbcEvent)>, String> {
- let mut vals: Vec<(Height, IbcEvent)> = vec![];
+) -> Result<Vec<(Height, IbcEventWithHash)>, String> {
+ let mut vals: Vec<(Height, IbcEventWithHash)> = vec![];
 
  match &result.data {
  RpcEventData::NewBlock { block, .. } => {
+ // info!("\t newblock looks like this: {:?}", result);
+
  let height = Height::new(
  ChainId::chain_version(chain_id.to_string().as_str()),
  u64::from(block.as_ref().ok_or("tx.height")?.header.height),
@@ -38,16 +63,29 @@ pub fn get_all_events(
  ChainId::chain_version(chain_id.to_string().as_str()),
  height_raw,
  );
+ let pre_tx_hash = events.get("tx.hash");
+ if let Some(hs) = pre_tx_hash {
+ let d = hs.iter().join(",");
+ info!("\t\t\t *** found tx hash(es) {}", d);
+ info!("DUMP: {:#?}", events);
+ }
+ let tx_hash_raw = pre_tx_hash.map(|hv| hv.first()).flatten().cloned();
 
  let actions_and_indices = extract_helper(events)?;
+ // info!("\t actions & indices: {:?}", actions_and_indices);
+ // info!("\t events: {:?}", events);
  for action in actions_and_indices {
  if let Ok(event) = build_event(RawObject::new(
  height,
  action.0,
  action.1 as usize,
  events.clone(),
  )) {
- vals.push((height, event));
+
+ vals.push((height, IbcEventWithHash {
+ event,
+ tx_hash: tx_hash_raw.clone(),
+ }));
  }
  }
  }

relayer/src/link.rs

@@ -16,6 +16,7 @@ mod error;
 mod operational_data;
 mod relay_path;
 mod relay_summary;
+mod relay_sender;
 
 // Re-export the telemetries summary
 pub use relay_summary::RelaySummary;
@@ -169,6 +170,7 @@ impl Link {
  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)?;
 
@@ -183,6 +185,7 @@ impl Link {
  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)?;
 

relayer/src/link/operational_data.rs

@@ -9,6 +9,7 @@ use ibc::Height;
 
 use crate::link::error::LinkError;
 use crate::link::RelayPath;
+// use crate::link::relay_sender;
 
 #[derive(Clone, Copy, PartialEq)]
 pub enum OperationalDataTarget {
@@ -64,8 +65,9 @@ impl OperationalData {
  self.batch.iter().map(|gm| gm.event.clone()).collect()
  }
 
- /// Returns all the messages in this operational data, plus prepending the client update message
- /// if necessary.
+ /// Returns all the messages in this operational data,
+ /// also prepending the client update message if
+ /// necessary.
  pub fn assemble_msgs(&self, relay_path: &RelayPath) -> Result<Vec<Any>, LinkError> {
  if self.batch.is_empty() {
  warn!("assemble_msgs() method call on an empty OperationalData!");
@@ -107,6 +109,10 @@ impl OperationalData {
 
  Ok(msgs)
  }
+
+ // pub fn into_sender(self, _mode: relay_sender::Mode) -> Box<dyn relay_sender::Sender> {
+ // todo!()
+ // }
 }
 
 impl fmt::Display for OperationalData {

relayer/src/link/relay_path.rs

@@ -37,6 +37,7 @@ use crate::foreign_client::{ForeignClient, ForeignClientError};
 use crate::link::error::LinkError;
 use crate::link::operational_data::{OperationalData, OperationalDataTarget, TransitMessage};
 use crate::link::relay_summary::RelaySummary;
+use crate::event::rpc::IbcEventWithHash;
 
 const MAX_RETRIES: usize = 5;
 
@@ -203,7 +204,7 @@ impl RelayPath {
 
  // Determines if the events received are relevant and should be processed.
  // Only events for a port/channel matching one of the channel ends should be processed.
- fn filter_events(&self, events: &[IbcEvent]) -> Vec<IbcEvent> {
+ fn filter_events(&self, events: &[IbcEventWithHash]) -> Vec<IbcEvent> {
  let mut result = vec![];
 
  let src_channel_id = if let Ok(some_id) = self.src_channel_id() {
@@ -212,34 +213,34 @@ impl RelayPath {
  return vec![];
  };
 
- for event in events.iter() {
- match event {
+ for ewh in events.iter() {
+ match &ewh.event {
  IbcEvent::SendPacket(send_packet_ev) => {
  if src_channel_id == send_packet_ev.src_channel_id()
  && self.src_port_id() == send_packet_ev.src_port_id()
  {
- result.push(event.clone());
+ result.push(ewh.event.clone());
  }
  }
  IbcEvent::WriteAcknowledgement(write_ack_ev) => {
  if src_channel_id == write_ack_ev.dst_channel_id()
  && self.src_port_id() == write_ack_ev.dst_port_id()
  {
- result.push(event.clone());
+ result.push(ewh.event.clone());
  }
  }
  IbcEvent::CloseInitChannel(chan_close_ev) => {
  if src_channel_id == chan_close_ev.channel_id()
  && self.src_port_id() == chan_close_ev.port_id()
  {
- result.push(event.clone());
+ result.push(ewh.event.clone());
  }
  }
  IbcEvent::TimeoutPacket(timeout_ev) => {
  if src_channel_id == timeout_ev.src_channel_id()
  && self.channel.src_port_id() == timeout_ev.src_port_id()
  {
- result.push(event.clone());
+ result.push(ewh.event.clone());
  }
  }
  _ => {}
@@ -453,6 +454,9 @@ impl RelayPath {
  }
 
  /// Returns the events generated by the target chain
+ /// TODO: Add a sender strategy as a parameter here -- either sync or async.
+ /// Based on the sender, the result of this method will differ.
+ /// call it relay_sender?
  pub(crate) fn relay_from_operational_data(
  &mut self,
  initial_od: OperationalData,
@@ -585,12 +589,11 @@ impl RelayPath {
  None
  }
 
- /// Sends a transaction to the chain targeted by the operational data `odata`.
- /// If the transaction generates an error, returns the error as well as `LinkError::SendError`
- /// if input events if a sending failure occurs.
- /// Returns the events generated by the target chain upon success.
+ /// Sends a transaction to the target chain of [`OperationalData`].
+ /// Returns the tx hashes generated by the target chain upon success.
+ /// Propagates any encountered errors.
  fn send_from_operational_data(
- &mut self,
+ &self,
  odata: OperationalData,
  ) -> Result<RelaySummary, LinkError> {
  if odata.batch.is_empty() {