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 {
+    let expected_time_per_block = 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
+}

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

@@ -211,7 +211,7 @@ impl<ChainA: ChainHandle, ChainB: ChainHandle> Link<ChainA, ChainB> {
         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() {
+        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)?;
@@ -237,7 +237,7 @@ impl<ChainA: ChainHandle, ChainB: ChainHandle> Link<ChainA, ChainB> {
         let mut results = vec![];
 
         // Block waiting for all of the scheduled data
-        while let Some(odata) = self.a_to_b.fetch_scheduled_operational_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)?;

relayer/src/link/error.rs

@@ -143,7 +143,9 @@ define_error! {
                     e.channel_id, e.chain_id)
             },
 
-    }
+        UpdateClientFailed
+             |_| { "failed to update client" },
+   }
 }
 
 impl HasExpiredOrFrozenError for LinkErrorDetail {

relayer/src/link/operational_data.rs

@@ -1,12 +1,14 @@
 use alloc::borrow::Cow;
 use core::fmt;
 use core::iter;
-use std::time::Instant;
+use std::time::{Duration, Instant};
 
 use ibc_proto::google::protobuf::Any;
 use nanoid::nanoid;
 use tracing::{debug, info};
 
+use ibc::core::ics02_client::client_state::ClientState;
+use ibc::core::ics04_channel::context::calculate_block_delay;
 use ibc::events::IbcEvent;
 use ibc::Height;
 
@@ -89,23 +91,28 @@ pub struct OperationalData {
     pub proofs_height: Height,
     pub batch: Vec<TransitMessage>,
     pub target: OperationalDataTarget,
-    /// Stores the time when the clients on the target chain has been updated, i.e., when this data
-    /// was scheduled. Necessary for packet delays.
-    pub scheduled_time: Instant,
     pub tracking_id: String,
+    /// Stores `Some(ConnectionDelay)` if the delay is non-zero and `None` otherwise
+    connection_delay: Option<ConnectionDelay>,
 }
 
 impl OperationalData {
     pub fn new(
         proofs_height: Height,
         target: OperationalDataTarget,
         tracking_id: impl Into<String>,
+        connection_delay: Duration,
     ) -> Self {
+        let connection_delay = if !connection_delay.is_zero() {
+            Some(ConnectionDelay::new(connection_delay))
+        } else {
+            None
+        };
         OperationalData {
             proofs_height,
             batch: vec![],
             target,
-            scheduled_time: Instant::now(),
+            connection_delay,
             tracking_id: tracking_id.into(),
         }
     }
@@ -141,7 +148,7 @@ impl OperationalData {
         relay_path: &RelayPath<ChainA, ChainB>,
     ) -> Result<TrackedMsgs, LinkError> {
         // For zero delay we prepend the client update msgs.
-        let client_update_msg = if relay_path.zero_delay() {
+        let client_update_msg = if !self.conn_delay_needed() {
             let update_height = self.proofs_height.increment();
 
             debug!(
@@ -162,7 +169,22 @@ impl OperationalData {
 
             client_update_opt.pop()
         } else {
-            None
+            let client_state = match self.target {
+                OperationalDataTarget::Source => relay_path
+                    .src_chain()
+                    .query_client_state(relay_path.src_client_id(), Height::zero())
+                    .map_err(|e| LinkError::query(relay_path.src_chain().id(), e))?,
+                OperationalDataTarget::Destination => relay_path
+                    .dst_chain()
+                    .query_client_state(relay_path.dst_client_id(), Height::zero())
+                    .map_err(|e| LinkError::query(relay_path.dst_chain().id(), e))?,
+            };
+
+            if client_state.is_frozen() {
+                return Ok(TrackedMsgs::new(vec![], &self.tracking_id));
+            } else {
+                None
+            }
         };
 
         let msgs: Vec<Any> = match client_update_msg {
@@ -178,6 +200,113 @@ impl OperationalData {
 
         Ok(tm)
     }
+
+    /// Returns true iff the batch contains a packet event
+    fn has_packet_msgs(&self) -> bool {
+        self.batch.iter().any(|msg| msg.event.packet().is_some())
+    }
+
+    /// Returns the `connection_delay` iff the connection delay for this relaying path is non-zero
+    /// and the `batch` contains packet messages.
+    fn get_delay_if_needed(&self) -> Option<&ConnectionDelay> {
+        self.connection_delay
+            .as_ref()
+            .filter(|_| self.has_packet_msgs())
+    }
+
+    /// Returns `true` iff the connection delay for this relaying path is non-zero and `op_data`
+    /// contains packet messages.
+    pub fn conn_delay_needed(&self) -> bool {
+        self.get_delay_if_needed().is_some()
+    }
+
+    /// Sets the scheduled time that is used for connection-delay calculations
+    pub fn set_scheduled_time(&mut self, scheduled_time: Instant) {
+        if let Some(mut delay) = self.connection_delay.as_mut() {
+            delay.scheduled_time = scheduled_time;
+        }
+    }
+
+    /// Sets the update height that is used for connection-delay calculations
+    pub fn set_update_height(&mut self, update_height: Height) {
+        if let Some(mut delay) = self.connection_delay.as_mut() {
+            delay.update_height = Some(update_height);
+        }
+    }
+
+    /// Returns `Ok(())` if the connection-delay has elapsed and `Err(remaining-delay)` otherwise.
+    pub fn conn_time_delay_elapsed(&self, chain_time: Instant) -> Result<(), Duration> {
+        if let Some(delay) = self.get_delay_if_needed() {
+            delay.conn_time_delay_elapsed(chain_time)
+        } else {
+            Ok(())
+        }
+    }
+
+    /// Returns `Ok(())` if the connection-delay has elapsed and `Err(remaining-delay)` otherwise.
+    pub fn conn_block_delay_elapsed(
+        &self,
+        max_expected_time_per_block: Duration,
+        latest_height: Height,
+    ) -> Result<(), u64> {
+        if let Some(delay) = self.get_delay_if_needed() {
+            let block_delay = delay.conn_block_delay(max_expected_time_per_block);
+            delay.conn_block_delay_elapsed(block_delay, latest_height)
+        } else {
+            Ok(())
+        }
+    }
+}
+
+/// A struct that holds everything that is required to calculate and deal with the connection-delay
+/// feature.
+#[derive(Clone)]
+struct ConnectionDelay {
+    delay: Duration,
+    scheduled_time: Instant,
+    update_height: Option<Height>,
+}
+
+impl ConnectionDelay {
+    fn new(delay: Duration) -> Self {
+        Self {
+            delay,
+            scheduled_time: Instant::now(),
+            update_height: None,
+        }
+    }
+
+    /// Returns `Ok(())` if the connection-delay has elapsed and `Err(remaining-delay)` otherwise.
+    fn conn_time_delay_elapsed(&self, chain_time: Instant) -> Result<(), Duration> {
+        // since chain time instant is relative to relayer's current time, it is possible that
+        // `scheduled_time` is later (by nano secs) than `chain_time`, hence the call to
+        // `saturating_duration_since()`.
+        let elapsed = chain_time.saturating_duration_since(self.scheduled_time);
+        if elapsed > self.delay {
+            Ok(())
+        } else {
+            Err(elapsed)
+        }
+    }
+
+    /// Returns `Ok(())` if the connection-delay has elapsed and `Err(remaining-delay)` otherwise.
+    fn conn_block_delay_elapsed(&self, block_delay: u64, latest_height: Height) -> Result<(), u64> {
+        let acceptable_height = self
+            .update_height
+            .expect("processed height not set")
+            .add(block_delay);
+        if latest_height >= acceptable_height {
+            Ok(())
+        } else {
+            debug_assert!(acceptable_height.revision_number == latest_height.revision_number);
+            Err(acceptable_height.revision_height - latest_height.revision_height)
+        }
+    }
+
+    /// Calculates and returns the block-delay based on the `max_expected_time_per_block`
+    fn conn_block_delay(&self, max_expected_time_per_block: Duration) -> u64 {
+        calculate_block_delay(self.delay, max_expected_time_per_block)
+    }
 }
 
 /// A lightweight informational data structure that can be extracted