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 a packet
+  fails to be relayed on an ordered channel, this used to block the
+  channel until either another relayer relayed the packet successfully or
+  until packet clearing kicked off. Hermes will now detect the failure
+  and attempt to clear packets on the channel in order to unblock it.
+  ([\#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/link/relay_path.rs

@@ -415,7 +415,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)?;
@@ -676,12 +676,22 @@ 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 on an ordered channel, perform a packet clearing, but only if we
+                    // are not in the middle of another packet clearing process
+                    if self.ordered_channel() && i == 0 && !odata.tracking_id.is_clearing() {
+                        warn!("Failed to relay to ordered channel, attempting to recover by clearing packets");
+
+                        // We do need to specify the height for the packet clearing,
+                        // since no progress will have been made on the clearing process
+                        self.relay_pending_packets(None)?;
+                    }
+
                     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

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_established, query_channel_end,
 };
+use ibc_test_framework::util::interchain_security::interchain_send_tx;
 
 #[test]
 fn test_ica_filter_default() -> Result<(), Error> {
@@ -178,29 +176,6 @@ impl BinaryConnectionTest for IcaFilterTestAllow {
     }
 }
 
-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)
-}
-
 #[test]
 fn test_ica_filter_deny() -> Result<(), Error> {
     run_binary_connection_test(&IcaFilterTestDeny)

tools/integration-test/src/tests/interchain_security/ica_ordered_channel.rs

@@ -0,0 +1,217 @@
+//! Verifies the behaviour of ordered channels with Interchain Accounts.
+//!
+//! In order to ensure that ordered channels correctly clear packets on ICA
+//! channels, this test sends some sequential packets with the supervisor enabled,
+//! sends the next packet *without* the supervisor enabled, then sends additional
+//! packets with the supervisor enabled again. The pending packet that was sent
+//! without the supervisor enabled should be relayed in order along with the
+//! other packets, as expected of ordered channel behaviour.
+
+use std::str::FromStr;
+
+use ibc_relayer_types::applications::ics27_ica::cosmos_tx::CosmosTx;
+use ibc_relayer_types::applications::ics27_ica::packet_data::InterchainAccountPacketData;
+use ibc_relayer_types::applications::transfer::msgs::send::MsgSend;
+use ibc_relayer_types::applications::transfer::{Amount, Coin};
+use ibc_relayer_types::bigint::U256;
+use ibc_relayer_types::signer::Signer;
+use ibc_relayer_types::timestamp::Timestamp;
+use ibc_relayer_types::tx_msg::Msg;
+use ibc_test_framework::chain::ext::ica::register_interchain_account;
+use ibc_test_framework::framework::binary::channel::run_binary_interchain_security_channel_test;
+use ibc_test_framework::prelude::*;
+use ibc_test_framework::relayer::channel::assert_eventually_channel_established;
+use ibc_test_framework::relayer::channel::query_channel_end;
+use ibc_test_framework::util::interchain_security::{
+    interchain_send_tx, update_genesis_for_consumer_chain, update_relayer_config_for_consumer_chain,
+};
+
+#[test]
+fn test_ica_ordered_channel() -> Result<(), Error> {
+    run_binary_interchain_security_channel_test(&IcaOrderedChannelTest)
+}
+
+struct IcaOrderedChannelTest;
+
+impl TestOverrides for IcaOrderedChannelTest {
+    fn modify_genesis_file(&self, genesis: &mut serde_json::Value) -> Result<(), Error> {
+        use serde_json::Value;
+
+        // Allow MsgSend messages over ICA
+        let allow_messages = genesis
+            .get_mut("app_state")
+            .and_then(|app_state| app_state.get_mut("interchainaccounts"))
+            .and_then(|ica| ica.get_mut("host_genesis_state"))
+            .and_then(|state| state.get_mut("params"))
+            .and_then(|params| params.get_mut("allow_messages"))
+            .and_then(|allow_messages| allow_messages.as_array_mut());
+
+        if let Some(allow_messages) = allow_messages {
+            allow_messages.push(Value::String("/cosmos.bank.v1beta1.MsgSend".to_string()));
+        } else {
+            return Err(Error::generic(eyre!("failed to update genesis file")));
+        }
+
+        update_genesis_for_consumer_chain(genesis)?;
+
+        Ok(())
+    }
+
+    fn modify_relayer_config(&self, config: &mut Config) {
+        config.mode.channels.enabled = true;
+
+        // Disable packet clearing so that packets sent without the supervisor
+        // enabled enter a pending state.
+        config.mode.packets.enabled = true;
+        config.mode.packets.clear_on_start = false;
+        config.mode.packets.clear_interval = 0;
+
+        update_relayer_config_for_consumer_chain(config);
+    }
+
+    fn should_spawn_supervisor(&self) -> bool {
+        false
+    }
+}
+
+impl BinaryChannelTest for IcaOrderedChannelTest {
+    fn run<ChainA: ChainHandle, ChainB: ChainHandle>(
+        &self,
+        _config: &TestConfig,
+        relayer: RelayerDriver,
+        chains: ConnectedChains<ChainA, ChainB>,
+        channel: ConnectedChannel<ChainA, ChainB>,
+    ) -> Result<(), Error> {
+        let connection_b_to_a = channel.connection.clone().flip();
+        let (wallet, channel_id, port_id) =
+            register_interchain_account(&chains.node_b, chains.handle_b(), &connection_b_to_a)?;
+
+        relayer.with_supervisor(|| {
+            // Check that the corresponding ICA channel is eventually established.
+            let _counterparty_channel_id = assert_eventually_channel_established(
+                chains.handle_b(),
+                chains.handle_a(),
+                &channel_id.as_ref(),
+                &port_id.as_ref(),
+            )?;
+
+            Ok(())
+        })?;
+
+        // Assert that the channel returned by `register_interchain_account` is an ordered channel
+        let channel_end =
+            query_channel_end(chains.handle_b(), &channel_id.as_ref(), &port_id.as_ref())?;
+
+        assert_eq!(channel_end.value().ordering(), &Ordering::Ordered);
+
+        // Query the controller chain for the address of the ICA wallet on the host chain.
+        let ica_address = chains.node_b.chain_driver().query_interchain_account(
+            &wallet.address(),
+            &channel.connection.connection_id_b.as_ref(),
+        )?;
+
+        let stake_denom: MonoTagged<ChainA, Denom> = MonoTagged::new(Denom::base("stake"));
+
+        chains.node_a.chain_driver().assert_eventual_wallet_amount(
+            &ica_address.as_ref(),
+            &stake_denom.with_amount(0u64).as_ref(),
+        )?;
+
+        // Send funds to the interchain account.
+        let ica_fund = 42000u64;
+
+        chains.node_a.chain_driver().local_transfer_token(
+            &chains.node_a.wallets().user1(),
+            &ica_address.as_ref(),
+            &stake_denom.with_amount(ica_fund).as_ref(),
+        )?;
+
+        chains.node_a.chain_driver().assert_eventual_wallet_amount(
+            &ica_address.as_ref(),
+            &stake_denom.with_amount(ica_fund).as_ref(),
+        )?;
+
+        let amount = 1200;
+
+        let msg = MsgSend {
+            from_address: ica_address.to_string(),
+            to_address: chains.node_a.wallets().user2().address().to_string(),
+            amount: vec![Coin {
+                denom: stake_denom.to_string(),
+                amount: Amount(U256::from(amount)),
+            }],
+        };
+
+        let raw_msg = msg.to_any();
+
+        let cosmos_tx = CosmosTx {
+            messages: vec![raw_msg],
+        };
+
+        let raw_cosmos_tx = cosmos_tx.to_any();
+
+        let interchain_account_packet_data = InterchainAccountPacketData::new(raw_cosmos_tx.value);
+
+        let signer = Signer::from_str(&wallet.address().to_string()).unwrap();
+
+        sleep(Duration::from_secs(5));
+
+        relayer.with_supervisor(|| {
+            sleep(Duration::from_secs(1));
+
+            let ica_events = interchain_send_tx(
+                chains.handle_b(),
+                &signer,
+                &channel.connection.connection_id_b.0,
+                interchain_account_packet_data.clone(),
+                Timestamp::from_nanoseconds(120000000000).unwrap(),
+            )?;
+
+            info!("First ICA transfer made with supervisor: {ica_events:#?}");
+
+            sleep(Duration::from_secs(5));
+
+            Ok(())
+        })?;
+
+        sleep(Duration::from_secs(1));
+
+        let ica_events = interchain_send_tx(
+            chains.handle_b(),
+            &signer,
+            &channel.connection.connection_id_b.0,
+            interchain_account_packet_data.clone(),
+            Timestamp::from_nanoseconds(120000000000).unwrap(),
+        )?;
+
+        info!("Second ICA transfer made without supervisor: {ica_events:#?}");
+
+        sleep(Duration::from_secs(5));
+
+        relayer.with_supervisor(|| {
+            sleep(Duration::from_secs(1));
+
+            let ica_events = interchain_send_tx(
+                chains.handle_b(),
+                &signer,
+                &channel.connection.connection_id_b.0,
+                interchain_account_packet_data,
+                Timestamp::from_nanoseconds(120000000000).unwrap(),
+            )?;
+
+            info!("Third ICA transfer made with supervisor: {ica_events:#?}");
+
+            sleep(Duration::from_secs(5));
+
+            Ok(())
+        })?;
+
+        // Check that the ICA account's balance has been debited the sent amount.
+        chains.node_a.chain_driver().assert_eventual_wallet_amount(
+            &ica_address.as_ref(),
+            &stake_denom.with_amount(ica_fund - 3 * amount).as_ref(),
+        )?;
+
+        Ok(())
+    }
+}

tools/integration-test/src/tests/interchain_security/ica_transfer.rs

@@ -3,10 +3,7 @@
 //! the second chain a Consumer chain.
 use std::str::FromStr;
 
-use ibc_relayer::chain::tracking::TrackedMsgs;
-use ibc_relayer::event::IbcEventWithHeight;
 use ibc_relayer_types::applications::ics27_ica::cosmos_tx::CosmosTx;
-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::transfer::msgs::send::MsgSend;
 use ibc_relayer_types::applications::transfer::{Amount, Coin};
@@ -19,7 +16,7 @@ use ibc_test_framework::framework::binary::channel::run_binary_interchain_securi
 use ibc_test_framework::prelude::*;
 use ibc_test_framework::relayer::channel::assert_eventually_channel_established;
 use ibc_test_framework::util::interchain_security::{
-    update_genesis_for_consumer_chain, update_relayer_config_for_consumer_chain,
+    interchain_send_tx, update_genesis_for_consumer_chain, update_relayer_config_for_consumer_chain,
 };
 
 #[test]
@@ -151,29 +148,7 @@ impl BinaryChannelTest for InterchainSecurityIcaTransferTest {
             &ica_address.as_ref(),
             &stake_denom.with_amount(ica_fund - amount).as_ref(),
         )?;
+
         Ok(())
     }
 }
-
-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)
-}