Make ordered channels more resilient in the face of failing packets

.changelog/unreleased/improvements/ibc-relayer/3540-ordered-channels-resilience.md

@@ -0,0 +1,6 @@
+- Improve resilience when relaying on ordered channels. 
+  When relaying packets on an ordered channel, Hermes will now attempt
+  to detect whether the next message to send has the sequence number
+  expected on that channel. If there is a mismatch, then Hermes will trigger a packet
+  clear on the channel to unblock it before resuming operations on that channel.
+  ([\#3540](https://github.com/informalsystems/hermes/issues/3540))

crates/relayer/src/chain/tracking.rs

@@ -15,7 +15,7 @@ pub enum TrackingId {
     /// the CLI or during packet clearing.
     Static(&'static str),
     /// Random identifier used to track latency of packet clearing.
-    ClearedUuid(Uuid),
+    PacketClearing(Uuid),
 }
 
 impl TrackingId {
@@ -29,8 +29,13 @@ impl TrackingId {
         Self::Static(s)
     }
 
-    pub fn new_cleared_uuid() -> Self {
-        Self::ClearedUuid(Uuid::new_v4())
+    pub fn new_packet_clearing() -> Self {
+        Self::PacketClearing(Uuid::new_v4())
+    }
+
+    /// Indicates whether a packet clearing process is currently in-progress.
+    pub fn is_clearing(&self) -> bool {
+        matches!(self, Self::PacketClearing(_))
     }
 }
 
@@ -43,7 +48,7 @@ impl Display for TrackingId {
                 s.fmt(f)
             }
             TrackingId::Static(s) => s.fmt(f),
-            TrackingId::ClearedUuid(u) => {
+            TrackingId::PacketClearing(u) => {
                 let mut uuid = "cleared/".to_owned();
                 let mut s = u.to_string();
                 s.truncate(8);

crates/relayer/src/config.rs

@@ -414,6 +414,9 @@ pub struct Packets {
     pub ics20_max_memo_size: Ics20FieldSizeLimit,
     #[serde(default = "default::ics20_max_receiver_size")]
     pub ics20_max_receiver_size: Ics20FieldSizeLimit,
+
+    #[serde(skip)]
+    pub force_disable_clear_on_start: bool,
 }
 
 impl Default for Packets {
@@ -426,6 +429,7 @@ impl Default for Packets {
             auto_register_counterparty_payee: default::auto_register_counterparty_payee(),
             ics20_max_memo_size: default::ics20_max_memo_size(),
             ics20_max_receiver_size: default::ics20_max_receiver_size(),
+            force_disable_clear_on_start: false,
         }
     }
 }

crates/relayer/src/link/error.rs

@@ -73,7 +73,7 @@ define_error! {
         Send
             { event: IbcEvent }
             |e| {
-                format!("chain error when sending messages: {0}", e.event)
+                format!("failed to send message: {0}", e.event)
             },
 
         MissingChannelId

crates/relayer/src/link/relay_path.rs

@@ -427,7 +427,7 @@ impl<ChainA: ChainHandle, ChainB: ChainHandle> RelayPath<ChainA, ChainB> {
     fn relay_pending_packets(&self, height: Option<Height>) -> Result<(), LinkError> {
         let _span = span!(Level::ERROR, "relay_pending_packets", ?height).entered();
 
-        let tracking_id = TrackingId::new_cleared_uuid();
+        let tracking_id = TrackingId::new_packet_clearing();
         telemetry!(received_event_batch, tracking_id);
 
         let src_config = self.src_chain().config().map_err(LinkError::relayer)?;
@@ -710,12 +710,12 @@ impl<ChainA: ChainHandle, ChainB: ChainHandle> RelayPath<ChainA, ChainB> {
 
                     return Ok(reply);
                 }
-                Err(LinkError(error::LinkErrorDetail::Send(e), _)) => {
-                    // This error means we could retry
-                    error!("error {}", e.event);
+                Err(LinkError(error::LinkErrorDetail::Send(_), _)) => {
                     if i + 1 == MAX_RETRIES {
-                        error!("{}/{} retries exhausted. giving up", i + 1, MAX_RETRIES)
+                        error!("{}/{} retries exhausted, giving up", i + 1, MAX_RETRIES)
                     } else {
+                        debug!("{}/{} retries exhausted, retrying with newly-generated operational data", i + 1, MAX_RETRIES);
+
                         // If we haven't exhausted all retries, regenerate the op. data & retry
                         match self.regenerate_operational_data(odata.clone()) {
                             None => return Ok(S::Reply::empty()), // Nothing to retry

crates/relayer/src/link/relay_sender.rs

@@ -95,10 +95,11 @@ impl Submit for AsyncSender {
     type Reply = AsyncReply;
 
     fn submit(target: &impl ChainHandle, msgs: TrackedMsgs) -> Result<Self::Reply, LinkError> {
-        let a = target
+        let responses = target
             .send_messages_and_wait_check_tx(msgs)
             .map_err(LinkError::relayer)?;
-        let reply = AsyncReply { responses: a };
+
+        let reply = AsyncReply { responses };
 
         // Note: There may be errors in the reply, for example:
         // `Response { code: Err(11), data: Data([]), log: Log("Too much gas wanted: 35000000, maximum is 25000000: out of gas")`

crates/relayer/src/worker.rs

@@ -128,7 +128,7 @@ pub fn spawn_worker_tasks<ChainA: ChainHandle, ChainB: ChainHandle>(
                 Ok(link) => {
                     let channel_ordering = link.a_to_b.channel().ordering;
                     let should_clear_on_start =
-                        packets_config.clear_on_start || channel_ordering == Ordering::Ordered;
+                        should_clear_on_start(&packets_config, channel_ordering);
 
                     let (cmd_tx, cmd_rx) = crossbeam_channel::unbounded();
                     let link = Arc::new(Mutex::new(link));
@@ -217,3 +217,11 @@ pub fn spawn_worker_tasks<ChainA: ChainHandle, ChainB: ChainHandle>(
 
     WorkerHandle::new(id, object, data, cmd_tx, task_handles)
 }
+
+fn should_clear_on_start(config: &crate::config::Packets, channel_ordering: Ordering) -> bool {
+    if config.force_disable_clear_on_start {
+        false
+    } else {
+        config.clear_on_start || channel_ordering == Ordering::Ordered
+    }
+}

crates/relayer/src/worker/packet.rs

@@ -17,14 +17,17 @@ use ibc_proto::ibc::apps::fee::v1::{IdentifiedPacketFees, QueryIncentivizedPacke
 use ibc_proto::ibc::core::channel::v1::PacketId;
 use ibc_relayer_types::applications::ics29_fee::events::IncentivizedPacket;
 use ibc_relayer_types::applications::transfer::{Amount, Coin, RawCoin};
+use ibc_relayer_types::core::ics04_channel::channel::Ordering;
 use ibc_relayer_types::core::ics04_channel::events::WriteAcknowledgement;
 use ibc_relayer_types::core::ics04_channel::packet::Sequence;
 use ibc_relayer_types::events::{IbcEvent, IbcEventType};
 use ibc_relayer_types::Height;
 
 use crate::chain::handle::ChainHandle;
+use crate::chain::requests::QueryHeight;
 use crate::config::filter::FeePolicy;
 use crate::event::source::EventBatch;
+use crate::event::IbcEventWithHeight;
 use crate::foreign_client::HasExpiredOrFrozenError;
 use crate::link::Resubmit;
 use crate::link::{error::LinkError, Link};
@@ -204,6 +207,24 @@ fn handle_packet_cmd<ChainA: ChainHandle, ChainB: ChainHandle>(
 ) -> Result<(), TaskError<RunError>> {
     // Handle packet clearing which is triggered from a command
     let (do_clear, maybe_height) = match &cmd {
+        WorkerCmd::IbcEvents { batch } if link.a_to_b.channel().ordering == Ordering::Ordered => {
+            let lowest_sequence = lowest_sequence(&batch.events);
+
+            let next_sequence = query_next_sequence_receive(
+                link.a_to_b.dst_chain(),
+                link.a_to_b.dst_port_id(),
+                link.a_to_b.dst_channel_id(),
+                QueryHeight::Specific(batch.height),
+            )
+            .ok();
+
+            if *should_clear_on_start || next_sequence < lowest_sequence {
+                (true, Some(batch.height))
+            } else {
+                (false, None)
+            }
+        }
+
         WorkerCmd::IbcEvents { batch } => {
             if *should_clear_on_start {
                 (true, Some(batch.height))
@@ -439,6 +460,34 @@ fn handle_execute_schedule<ChainA: ChainHandle, ChainB: ChainHandle>(
     Ok(())
 }
 
+fn query_next_sequence_receive<Chain: ChainHandle>(
+    chain: &Chain,
+    port_id: &PortId,
+    channel_id: &ChannelId,
+    height: QueryHeight,
+) -> Result<Sequence, LinkError> {
+    use crate::chain::requests::{IncludeProof, QueryNextSequenceReceiveRequest};
+
+    chain
+        .query_next_sequence_receive(
+            QueryNextSequenceReceiveRequest {
+                port_id: port_id.clone(),
+                channel_id: channel_id.clone(),
+                height,
+            },
+            IncludeProof::No,
+        )
+        .map(|(seq, _height)| seq)
+        .map_err(|e| LinkError::query(chain.id(), e))
+}
+
+fn lowest_sequence(events: &[IbcEventWithHeight]) -> Option<Sequence> {
+    events
+        .iter()
+        .flat_map(|event| event.event.packet().map(|p| p.sequence))
+        .min()
+}
+
 #[cfg(feature = "telemetry")]
 use crate::link::RelaySummary;
 

tools/integration-test/src/tests/ica.rs

@@ -2,13 +2,10 @@ use std::collections::HashMap;
 use std::str::FromStr;
 
 use ibc_relayer::chain::handle::ChainHandle;
-use ibc_relayer::chain::tracking::TrackedMsgs;
 use ibc_relayer::config::{
     filter::{ChannelFilters, ChannelPolicy, FilterPattern},
     ChainConfig, PacketFilter,
 };
-use ibc_relayer::event::IbcEventWithHeight;
-use ibc_relayer_types::applications::ics27_ica::msgs::send_tx::MsgSendTx;
 use ibc_relayer_types::applications::ics27_ica::packet_data::InterchainAccountPacketData;
 use ibc_relayer_types::applications::{
     ics27_ica::cosmos_tx::CosmosTx,
@@ -26,6 +23,7 @@ use ibc_test_framework::prelude::*;
 use ibc_test_framework::relayer::channel::{
     assert_eventually_channel_closed, assert_eventually_channel_established, query_channel_end,
 };
+use ibc_test_framework::util::interchain_security::interchain_send_tx;
 
 #[test]
 fn test_ica_filter_default() -> Result<(), Error> {
@@ -187,6 +185,7 @@ impl BinaryConnectionTest for IcaFilterTestAllow {
         Ok(())
     }
 }
+
 pub struct IcaFilterTestDeny;
 
 impl TestOverrides for IcaFilterTestDeny {
@@ -373,26 +372,3 @@ impl BinaryConnectionTest for ICACloseChannelTest {
         })
     }
 }
-
-fn interchain_send_tx<ChainA: ChainHandle>(
-    chain: &ChainA,
-    from: &Signer,
-    connection: &ConnectionId,
-    msg: InterchainAccountPacketData,
-    relative_timeout: Timestamp,
-) -> Result<Vec<IbcEventWithHeight>, Error> {
-    let msg = MsgSendTx {
-        owner: from.clone(),
-        connection_id: connection.clone(),
-        packet_data: msg,
-        relative_timeout,
-    };
-
-    let msg_any = msg.to_any();
-
-    let tm = TrackedMsgs::new_static(vec![msg_any], "SendTx");
-
-    chain
-        .send_messages_and_wait_commit(tm)
-        .map_err(Error::relayer)
-}