Skip to content

Commit

Permalink
Address my review comments
Browse files Browse the repository at this point in the history
  • Loading branch information
@ancazamfir
ancazamfir committed May 11, 2022
1 parent 02378dc commit 2775ed0
Show file tree
Hide file tree
Showing 2 changed files with 46 additions and 99 deletions.
55 changes: 27 additions & 28 deletions relayer/src/foreign_client.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -716,23 +716,20 @@ impl<DstChain: ChainHandle, SrcChain: ChainHandle> ForeignClient<DstChain, SrcCh
let client_latest_height = client_state.latest_height();

if client_latest_height < target_height {
// If the latest height of the client is already lower than the
// target height, we can simply use it.
// Use the latest client height if it is lower than the
// target height.
Ok(client_latest_height)
} else {
// We cannot use the client's latest height as trusted height. Instead, we
// need to find a consensus state at height `h` that this client previously
// installed, so that `h` < `target_height`.
// Latest height cannot be used as trusted height.
// Find a consensus state at height `h` so that `h` < `target_height`.
//
// The occasion when we're in this case is when the command line user
// wants to submit a client update at an older height even
// when the relayer already have an up-to-date client at a newer height.
// In production, this should rarely happen, e.g., when there is another
// relayer racing to update the client state, and that we so happen
// to get the the latest client state that was updated between
// the time the target height was determined, and the time
// the client state was fetched.
self.consensus_state_height_bounded(target_height)
// Note: This can happen in the following cases:
// - a client update CLI specifies a target height lower than the
// latest on-chain consensus height
// - multiple workers are relaying between the same two chains using same clients and
// perform client update with the height dictated by the proof heights
// - multiple relayer instances
self.bounded_consensus_state_height(target_height)
}
}

Expand Down Expand Up @@ -1107,47 +1104,49 @@ impl<DstChain: ChainHandle, SrcChain: ChainHandle> ForeignClient<DstChain, SrcCh
Ok(res)
}

/// Retrieves all consensus heights for this client sorted in descending
/// order.
fn consensus_state_heights(&self) -> Result<Vec<Height>, ForeignClientError> {
/// Retrieves some or all consensus heights for this client as specified by
/// `result_limit`. Heights are sorted in descending order.
fn consensus_state_heights(
&self,
result_limit: Option<u64>,
) -> Result<Vec<Height>, ForeignClientError> {
// [TODO] Utilize query that only fetches consensus state heights
// https://github.com/cosmos/ibc-go/issues/798
let consensus_state_heights: Vec<Height> = self
.consensus_states_with_limit(None)?
.consensus_states_with_limit(result_limit)?
.iter()
.map(|cs| cs.height)
.collect();

Ok(consensus_state_heights)
}

/// Search the consensus heights of this client to find one that is
/// Find the highest client consensus state with a height
/// smaller than the provided [`Height`] upper bound.
///
/// Utility method for use when solving a trusted height for this client.
fn consensus_state_height_bounded(
fn bounded_consensus_state_height(
&self,
upper_bound: Height,
) -> Result<Height, ForeignClientError> {
// Optimistically fetch only the last few consensus states.
// Iterate through the available consensus heights and find one
// that is lower than the target height.
if let Some(res) = self
.consensus_states_with_limit(Some(OPTIMISTIC_CONSENSUS_STATES_QUERY_LIMIT))?
.consensus_state_heights(Some(OPTIMISTIC_CONSENSUS_STATES_QUERY_LIMIT))?
.iter()
.map(|cs| cs.height)
.find(|h| h < &upper_bound)
.find(|h| h < &&upper_bound)
{
info!(upper_bound = %upper_bound, result = %res, "optimistically resolved consensus_state_height");
Ok(res)
Ok(*res)
} else {
// The optimistic query was not enough. We'll pull _all_ the consensus states
// and pick an appropriate height among those.
warn!("resolving trusted height from the full list of consensus state heights; this may take a while");
self.consensus_states_with_limit(None)?
self.consensus_state_heights(None)?
.iter()
.map(|cs| cs.height)
.find(|h| h < &upper_bound)
.find(|h| h < &&upper_bound)
.cloned()
.ok_or_else(|| {
ForeignClientError::missing_smaller_trusted_height(
self.dst_chain().id(),
Expand Down Expand Up @@ -1216,7 +1215,7 @@ impl<DstChain: ChainHandle, SrcChain: ChainHandle> ForeignClient<DstChain, SrcCh
// the one installed by the `CreateClient` which does not include a header.
// For chains that do support pruning, it is possible that the last consensus state
// was installed by an `UpdateClient` and an event and header will be found.
self.consensus_state_heights()?
self.consensus_state_heights(None)?
};

trace!(
Expand Down
90 changes: 19 additions & 71 deletions relayer/src/link/cli.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -2,18 +2,17 @@ use std::convert::TryInto;
use std::thread;
use std::time::{Duration, Instant};

use tracing::{error, error_span, info};
use tracing::{error_span, info};

use ibc::events::IbcEvent;
use ibc::Height;

use crate::chain::counterparty::{unreceived_acknowledgements, unreceived_packets};
use crate::chain::handle::ChainHandle;
use crate::link::error::LinkError;
use crate::link::operational_data::OperationalData;
use crate::link::packet_events::{query_packet_events_with, query_send_packet_events};

use crate::link::relay_path::RelayPath;
use crate::link::Link;
use crate::link::{relay_sender, Link};

// TODO(Adi): Open an issue or discussion. Options are:
// a. We remove this code and deprecate relaying on paths with non-zero delay.
Expand Down Expand Up @@ -58,41 +57,16 @@ impl<ChainA: ChainHandle, ChainB: ChainHandle> Link<ChainA, ChainB> {
)
.entered();

// Relaying on a non-zero connection delay requires (indefinite) blocking
// to wait for the connection delay to pass.
// We do not support this in interactive mode.
if !self.a_to_b.channel().connection_delay.is_zero() {
error!(
"relaying on a non-zero connection delay path is not supported in interactive mode"
);
panic!("please use the passive relaying mode (`hermes start`)");
}

// Find the sequence numbers of unreceived packets
let (sequences, src_response_height) = unreceived_packets(
self.a_to_b.dst_chain(),
self.a_to_b.src_chain(),
&self.a_to_b.path_id,
)
.map_err(LinkError::supervisor)?;
self.a_to_b.schedule_recv_packet_and_timeout_msgs(None)?;

if sequences.is_empty() {
return Ok(vec![]);
}

info!("unreceived packets found: {} ", sequences.len());

// Relay
let mut results = vec![];
for events_chunk in query_packet_events_with(
&sequences,
src_response_height,
self.a_to_b.src_chain(),
&self.a_to_b.path_id,
query_send_packet_events,
) {
let mut last_events = self.a_to_b.relay_from_events(events_chunk)?;
results.append(&mut last_events.events);

// 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::<relay_sender::SyncSender>(odata)?;
results.append(&mut last_res.events);
}

Ok(results)
Expand All @@ -109,42 +83,16 @@ impl<ChainA: ChainHandle, ChainB: ChainHandle> Link<ChainA, ChainB> {
)
.entered();

// Relaying on a non-zero connection delay requires (indefinite) blocking
// to wait for the connection delay to pass.
// We do not support this in interactive mode.
if !self.a_to_b.channel().connection_delay.is_zero() {
error!(
"relaying on a non-zero connection delay path is not supported in interactive mode"
);
panic!("please use the passive relaying mode (`hermes start`)");
}

// Find the sequence numbers of unreceived acknowledgements
let (sequences, src_response_height) = unreceived_acknowledgements(
self.a_to_b.dst_chain(),
self.a_to_b.src_chain(),
&self.a_to_b.path_id,
)
.map_err(LinkError::supervisor)?;
self.a_to_b.schedule_packet_ack_msgs(None)?;

if sequences.is_empty() {
return Ok(vec![]);
}

info!("unreceived acknowledgements found: {} ", sequences.len());

// Relay
let mut results = vec![];
for events_chunk in query_packet_events_with(
&sequences,
src_response_height,
self.a_to_b.src_chain(),
&self.a_to_b.path_id,
query_send_packet_events,
) {
// Bypass scheduling and waiting on operational data, relay directly.
let mut last_events = self.a_to_b.relay_from_events(events_chunk)?;
results.append(&mut last_events.events);

// 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::<relay_sender::SyncSender>(odata)?;
results.append(&mut last_res.events);
}

Ok(results)
Expand Down

0 comments on commit 2775ed0

Please sign in to comment.