Fix the connection delay logic to use the header update block time

.changelog/unreleased/bug-fixes/ibc-relayer/1772-fix-conn-delay-check.md

@@ -0,0 +1,2 @@
+- Fix the connection delay logic to use the timestamp of the host block when the client update header was installed.
+  ([#1772](https://github.com/informalsystems/ibc-rs/issues/1772))

config.toml

@@ -177,6 +177,7 @@ clock_drift = '5s'
 # The block time together with the clock drift are added to the source drift to estimate
 # the maximum clock drift when creating a client on this chain. Default: 10s
 # For cosmos-SDK chains a good approximation is `timeout_propose` + `timeout_commit`
+# Note: This MUST be the same as the `max_expected_time_per_block` genesis parameter for Tendermint chains.
 max_block_time = '10s'
 
 # Specify the amount of time to be used as the light client trusting period.

modules/src/core/ics04_channel/context.rs

@@ -99,13 +99,7 @@ pub trait ChannelReader {
     fn max_expected_time_per_block(&self) -> Duration;
 
     fn block_delay(&self, delay_period_time: Duration) -> u64 {
-        let expected_time_per_block = self.max_expected_time_per_block();
-        if expected_time_per_block.is_zero() {
-            return 0;
-        }
-
-        FloatCore::ceil(delay_period_time.as_secs_f64() / expected_time_per_block.as_secs_f64())
-            as u64
+        calculate_block_delay(delay_period_time, self.max_expected_time_per_block())
     }
 }
 
@@ -279,3 +273,15 @@ pub trait ChannelKeeper {
     /// Should never fail.
     fn increase_channel_counter(&mut self);
 }
+
+pub fn calculate_block_delay(
+    delay_period_time: Duration,
+    max_expected_time_per_block: Duration,
+) -> u64 {
+    if max_expected_time_per_block.is_zero() {
+        return 0;
+    }
+
+    FloatCore::ceil(delay_period_time.as_secs_f64() / max_expected_time_per_block.as_secs_f64())
+        as u64
+}

relayer/src/chain.rs

@@ -270,6 +270,8 @@ pub trait ChainEndpoint: Sized {
         request: QueryBlockRequest,
     ) -> Result<(Vec<IbcEvent>, Vec<IbcEvent>), Error>;
 
+    fn query_host_consensus_state(&self, height: ICSHeight) -> Result<Self::ConsensusState, Error>;
+
     // Provable queries
     fn proven_client_state(
         &self,

relayer/src/chain/cosmos.rs

@@ -1948,6 +1948,20 @@ impl ChainEndpoint for CosmosSdkChain {
         }
     }
 
+    fn query_host_consensus_state(&self, height: ICSHeight) -> Result<Self::ConsensusState, Error> {
+        let height = Height::try_from(height.revision_height).map_err(Error::invalid_height)?;
+
+        // TODO(hu55a1n1): use the `/header` RPC endpoint instead when we move to tendermint v0.35.x
+        let rpc_call = match height.value() {
+            0 => self.rpc_client.latest_block(),
+            _ => self.rpc_client.block(height),
+        };
+        let response = self
+            .block_on(rpc_call)
+            .map_err(|e| Error::rpc(self.config.rpc_addr.clone(), e))?;
+        Ok(response.block.header.into())
+    }
+
     fn proven_client_state(
         &self,
         client_id: &ClientId,

relayer/src/chain/handle.rs

@@ -333,6 +333,11 @@ pub enum ChainRequest {
         request: QueryBlockRequest,
         reply_to: ReplyTo<(Vec<IbcEvent>, Vec<IbcEvent>)>,
     },
+
+    QueryHostConsensusState {
+        height: Height,
+        reply_to: ReplyTo<AnyConsensusState>,
+    },
 }
 
 pub trait ChainHandle: Clone + Send + Sync + Serialize + Debug + 'static {
@@ -560,4 +565,6 @@ pub trait ChainHandle: Clone + Send + Sync + Serialize + Debug + 'static {
         &self,
         request: QueryBlockRequest,
     ) -> Result<(Vec<IbcEvent>, Vec<IbcEvent>), Error>;
+
+    fn query_host_consensus_state(&self, height: Height) -> Result<AnyConsensusState, Error>;
 }

relayer/src/chain/handle/base.rs

@@ -454,6 +454,10 @@ impl ChainHandle for BaseChainHandle {
     ) -> Result<(Vec<IbcEvent>, Vec<IbcEvent>), Error> {
         self.send(|reply_to| ChainRequest::QueryPacketEventDataFromBlocks { request, reply_to })
     }
+
+    fn query_host_consensus_state(&self, height: Height) -> Result<AnyConsensusState, Error> {
+        self.send(|reply_to| ChainRequest::QueryHostConsensusState { height, reply_to })
+    }
 }
 
 impl Serialize for BaseChainHandle {

relayer/src/chain/handle/cache.rs

@@ -412,4 +412,8 @@ impl<Handle: ChainHandle> ChainHandle for CachingChainHandle<Handle> {
     ) -> Result<(Vec<IbcEvent>, Vec<IbcEvent>), Error> {
         self.inner().query_blocks(request)
     }
+
+    fn query_host_consensus_state(&self, height: Height) -> Result<AnyConsensusState, Error> {
+        self.inner.query_host_consensus_state(height)
+    }
 }

relayer/src/chain/handle/counting.rs

@@ -448,4 +448,8 @@ impl<Handle: ChainHandle> ChainHandle for CountingChainHandle<Handle> {
         self.inc_metric("query_blocks");
         self.inner().query_blocks(request)
     }
+
+    fn query_host_consensus_state(&self, height: Height) -> Result<AnyConsensusState, Error> {
+        self.inner.query_host_consensus_state(height)
+    }
 }

relayer/src/chain/mock.rs

@@ -302,6 +302,10 @@ impl ChainEndpoint for MockChain {
         unimplemented!()
     }
 
+    fn query_host_consensus_state(&self, _height: Height) -> Result<Self::ConsensusState, Error> {
+        unimplemented!()
+    }
+
     fn proven_client_state(
         &self,
         _client_id: &ClientId,

relayer/src/chain/runtime.rs

@@ -411,6 +411,10 @@ where
                             self.query_blocks(request, reply_to)?
                         },
 
+                        Ok(ChainRequest::QueryHostConsensusState { height, reply_to }) => {
+                            self.query_host_consensus_state(height, reply_to)?
+                        },
+
                         Err(e) => error!("received error via chain request channel: {}", e),
                     }
                 },
@@ -870,4 +874,19 @@ where
 
         Ok(())
     }
+
+    fn query_host_consensus_state(
+        &self,
+        height: Height,
+        reply_to: ReplyTo<AnyConsensusState>,
+    ) -> Result<(), Error> {
+        let result = self
+            .chain
+            .query_host_consensus_state(height)
+            .map(|h| h.wrap_any());
+
+        reply_to.send(result).map_err(Error::send)?;
+
+        Ok(())
+    }
 }

relayer/src/link.rs

@@ -4,17 +4,16 @@ use ibc::{
         ics04_channel::channel::State as ChannelState,
         ics24_host::identifier::{ChannelId, PortChannelId, PortId},
     },
-    events::IbcEvent,
     Height,
 };
-use tracing::error_span;
 
 use crate::chain::counterparty::check_channel_counterparty;
 use crate::chain::handle::ChainHandle;
 use crate::channel::{Channel, ChannelSide};
 use crate::link::error::LinkError;
 use crate::link::relay_path::RelayPath;
 
+pub mod cli;
 pub mod error;
 pub mod operational_data;
 mod pending;
@@ -194,56 +193,4 @@ impl<ChainA: ChainHandle, ChainB: ChainHandle> Link<ChainA, ChainB> {
         // going slowly, but reliably.
         Link::new_from_opts(chain_b, chain_a, opts, with_tx_confirmation)
     }
-
-    /// Implements the `packet-recv` CLI
-    pub fn build_and_send_recv_packet_messages(&self) -> Result<Vec<IbcEvent>, LinkError> {
-        let _span = error_span!(
-            "PacketRecvCmd",
-            src_chain = %self.a_to_b.src_chain().id(),
-            src_port = %self.a_to_b.src_port_id(),
-            src_channel = %self.a_to_b.src_channel_id(),
-            dst_chain = %self.a_to_b.dst_chain().id(),
-        )
-        .entered();
-
-        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::<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(&self) -> Result<Vec<IbcEvent>, LinkError> {
-        let _span = error_span!(
-            "PacketAckCmd",
-            src_chain = %self.a_to_b.src_chain().id(),
-            src_port = %self.a_to_b.src_port_id(),
-            src_channel = %self.a_to_b.src_channel_id(),
-            dst_chain = %self.a_to_b.dst_chain().id(),
-        )
-        .entered();
-
-        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::<relay_sender::SyncSender>(odata)?;
-            results.append(&mut last_res.events);
-        }
-
-        Ok(results)
-    }
 }

relayer/src/link/cli.rs

@@ -0,0 +1,154 @@
+use std::convert::TryInto;
+use std::thread;
+use std::time::{Duration, Instant};
+
+use ibc::events::IbcEvent;
+use ibc::Height;
+use tracing::{error_span, info};
+
+use crate::chain::handle::ChainHandle;
+use crate::link::error::LinkError;
+use crate::link::operational_data::OperationalData;
+use crate::link::relay_path::RelayPath;
+use crate::link::{relay_sender, Link};
+
+impl<ChainA: ChainHandle, ChainB: ChainHandle> RelayPath<ChainA, ChainB> {
+    /// Fetches an operational data that has fulfilled its predefined delay period. May _block_
+    /// waiting for the delay period to pass.
+    /// Returns `Ok(None)` if there is no operational data scheduled.
+    pub(crate) fn fetch_scheduled_operational_data(
+        &self,
+    ) -> Result<Option<OperationalData>, LinkError> {
+        if let Some(odata) = self.src_operational_data.pop_front() {
+            Ok(Some(wait_for_conn_delay(
+                odata,
+                &|| self.src_time_latest(),
+                &|| self.src_max_block_time(),
+                &|| self.src_latest_height(),
+            )?))
+        } else if let Some(odata) = self.dst_operational_data.pop_front() {
+            Ok(Some(wait_for_conn_delay(
+                odata,
+                &|| self.dst_time_latest(),
+                &|| self.dst_max_block_time(),
+                &|| self.dst_latest_height(),
+            )?))
+        } else {
+            Ok(None)
+        }
+    }
+}
+
+impl<ChainA: ChainHandle, ChainB: ChainHandle> Link<ChainA, ChainB> {
+    /// Implements the `packet-recv` CLI
+    pub fn build_and_send_recv_packet_messages(&self) -> Result<Vec<IbcEvent>, LinkError> {
+        let _span = error_span!(
+            "PacketRecvCmd",
+            src_chain = %self.a_to_b.src_chain().id(),
+            src_port = %self.a_to_b.src_port_id(),
+            src_channel = %self.a_to_b.src_channel_id(),
+            dst_chain = %self.a_to_b.dst_chain().id(),
+        )
+        .entered();
+
+        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::<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(&self) -> Result<Vec<IbcEvent>, LinkError> {
+        let _span = error_span!(
+            "PacketAckCmd",
+            src_chain = %self.a_to_b.src_chain().id(),
+            src_port = %self.a_to_b.src_port_id(),
+            src_channel = %self.a_to_b.src_channel_id(),
+            dst_chain = %self.a_to_b.dst_chain().id(),
+        )
+        .entered();
+
+        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::<relay_sender::SyncSender>(odata)?;
+            results.append(&mut last_res.events);
+        }
+
+        Ok(results)
+    }
+}
+
+fn wait_for_conn_delay<ChainTime, MaxBlockTime, LatestHeight>(
+    odata: OperationalData,
+    chain_time: &ChainTime,
+    max_expected_time_per_block: &MaxBlockTime,
+    latest_height: &LatestHeight,
+) -> Result<OperationalData, LinkError>
+where
+    ChainTime: Fn() -> Result<Instant, LinkError>,
+    MaxBlockTime: Fn() -> Result<Duration, LinkError>,
+    LatestHeight: Fn() -> Result<Height, LinkError>,
+{
+    let (time_left, blocks_left) =
+        odata.conn_delay_remaining(chain_time, max_expected_time_per_block, latest_height)?;
+
+    match (time_left, blocks_left) {
+        (Duration::ZERO, 0) => {
+            info!(
+                "ready to fetch a scheduled op. data with batch of size {} targeting {}",
+                odata.batch.len(),
+                odata.target,
+            );
+            Ok(odata)
+        }
+        (Duration::ZERO, blocks_left) => {
+            info!(
+                    "waiting ({:?} blocks left) for a scheduled op. data with batch of size {} targeting {}",
+                    blocks_left,
+                    odata.batch.len(),
+                    odata.target,
+                );
+
+            let blocks_left: u32 = blocks_left.try_into().expect("blocks_left > u32::MAX");
+
+            // Wait until the delay period passes
+            thread::sleep(blocks_left * max_expected_time_per_block()?);
+
+            Ok(odata)
+        }
+        (time_left, _) => {
+            info!(
+                "waiting ({:?} left) for a scheduled op. data with batch of size {} targeting {}",
+                time_left,
+                odata.batch.len(),
+                odata.target,
+            );
+
+            // Wait until the delay period passes
+            thread::sleep(time_left);
+
+            // `blocks_left` maybe non-zero, so recurse to recheck that all delays are handled.
+            wait_for_conn_delay(
+                odata,
+                chain_time,
+                max_expected_time_per_block,
+                latest_height,
+            )
+        }
+    }
+}