Implement clear packets command

.changelog/unreleased/improvements/ibc-relayer-cli/1834-clear-packets-cmd.md

@@ -0,0 +1,3 @@
+- Added `clear packets` command, combining the effects of
+  `tx raw packet-recv` and `tx raw packet-ack`.
+  ([#1834](https://github.com/informalsystems/ibc-rs/pull/1834))

relayer-cli/src/commands.rs

@@ -5,6 +5,7 @@
 //! See the `impl Configurable` below for how to specify the path to the
 //! application's configuration file.
 
+mod clear;
 mod completions;
 mod config;
 mod create;
@@ -20,9 +21,10 @@ mod upgrade;
 mod version;
 
 use self::{
-    completions::CompletionsCmd, config::ConfigCmd, create::CreateCmds, health::HealthCheckCmd,
-    keys::KeysCmd, listen::ListenCmd, misbehaviour::MisbehaviourCmd, query::QueryCmd,
-    start::StartCmd, tx::TxCmd, update::UpdateCmds, upgrade::UpgradeCmds, version::VersionCmd,
+    clear::ClearCmds, completions::CompletionsCmd, config::ConfigCmd, create::CreateCmds,
+    health::HealthCheckCmd, keys::KeysCmd, listen::ListenCmd, misbehaviour::MisbehaviourCmd,
+    query::QueryCmd, start::StartCmd, tx::TxCmd, update::UpdateCmds, upgrade::UpgradeCmds,
+    version::VersionCmd,
 };
 
 use core::time::Duration;
@@ -63,6 +65,10 @@ pub enum CliCmd {
     #[clap(subcommand)]
     Upgrade(UpgradeCmds),
 
+    /// Clear objects, such as outstanding packets on a channel.
+    #[clap(subcommand)]
+    Clear(ClearCmds),
+
     /// Start the relayer in multi-chain mode.
     ///
     /// Relays packets and open handshake messages between all chains in the config.

relayer-cli/src/commands/clear.rs

@@ -0,0 +1,95 @@
+use abscissa_core::clap::Parser;
+use abscissa_core::{Command, Runnable};
+
+use ibc::core::ics24_host::identifier::{ChainId, ChannelId, PortId};
+use ibc::events::IbcEvent;
+use ibc_relayer::chain::handle::ProdChainHandle;
+use ibc_relayer::link::error::LinkError;
+use ibc_relayer::link::{Link, LinkParameters};
+
+use crate::application::app_config;
+use crate::cli_utils::spawn_chain_counterparty;
+use crate::conclude::Output;
+use crate::error::Error;
+
+/// `clear` subcommands
+#[derive(Command, Debug, Parser, Runnable)]
+pub enum ClearCmds {
+    /// Clear outstanding packets (i.e., packet-recv and packet-ack)
+    /// on a given channel in both directions. The channel is identified
+    /// by the chain, port, and channel IDs at one of its ends.
+    Packets(ClearPacketsCmd),
+}
+
+#[derive(Debug, Parser)]
+pub struct ClearPacketsCmd {
+    #[clap(required = true, help = "identifier of the chain")]
+    chain_id: ChainId,
+
+    #[clap(required = true, help = "identifier of the port")]
+    port_id: PortId,
+
+    #[clap(required = true, help = "identifier of the channel")]
+    channel_id: ChannelId,
+}
+
+impl Runnable for ClearPacketsCmd {
+    fn run(&self) {
+        let config = app_config();
+
+        let chains = match spawn_chain_counterparty::<ProdChainHandle>(
+            &config,
+            &self.chain_id,
+            &self.port_id,
+            &self.channel_id,
+        ) {
+            Ok((chains, _)) => chains,
+            Err(e) => Output::error(format!("{}", e)).exit(),
+        };
+
+        let mut ev_list = vec![];
+
+        // Construct links in both directions.
+        let opts = LinkParameters {
+            src_port_id: self.port_id.clone(),
+            src_channel_id: self.channel_id.clone(),
+        };
+        let fwd_link = match Link::new_from_opts(chains.src.clone(), chains.dst, opts, false) {
+            Ok(link) => link,
+            Err(e) => Output::error(format!("{}", e)).exit(),
+        };
+        let rev_link = match fwd_link.reverse(false) {
+            Ok(link) => link,
+            Err(e) => Output::error(format!("{}", e)).exit(),
+        };
+
+        // Schedule RecvPacket messages for pending packets in both directions.
+        // This may produce pending acks which will be processed in the next phase.
+        run_and_collect_events(&mut ev_list, || {
+            fwd_link.build_and_send_recv_packet_messages()
+        });
+        run_and_collect_events(&mut ev_list, || {
+            rev_link.build_and_send_recv_packet_messages()
+        });
+
+        // Schedule AckPacket messages in both directions.
+        run_and_collect_events(&mut ev_list, || {
+            fwd_link.build_and_send_ack_packet_messages()
+        });
+        run_and_collect_events(&mut ev_list, || {
+            rev_link.build_and_send_ack_packet_messages()
+        });
+
+        Output::success(ev_list).exit()
+    }
+}
+
+fn run_and_collect_events<F>(ev_list: &mut Vec<IbcEvent>, f: F)
+where
+    F: FnOnce() -> Result<Vec<IbcEvent>, LinkError>,
+{
+    match f() {
+        Ok(mut ev) => ev_list.append(&mut ev),
+        Err(e) => Output::error(Error::link(e)).exit(),
+    };
+}

relayer-cli/src/commands/tx/packet.rs

@@ -38,7 +38,7 @@ impl Runnable for TxRawPacketRecvCmd {
             src_port_id: self.src_port_id.clone(),
             src_channel_id: self.src_channel_id.clone(),
         };
-        let mut link = match Link::new_from_opts(chains.src, chains.dst, opts, false) {
+        let link = match Link::new_from_opts(chains.src, chains.dst, opts, false) {
             Ok(link) => link,
             Err(e) => Output::error(format!("{}", e)).exit(),
         };
@@ -82,7 +82,7 @@ impl Runnable for TxRawPacketAckCmd {
             src_port_id: self.src_port_id.clone(),
             src_channel_id: self.src_channel_id.clone(),
         };
-        let mut link = match Link::new_from_opts(chains.src, chains.dst, opts, false) {
+        let link = match Link::new_from_opts(chains.src, chains.dst, opts, false) {
             Ok(link) => link,
             Err(e) => Output::error(format!("{}", e)).exit(),
         };

relayer/src/link.rs

@@ -55,7 +55,12 @@ impl<ChainA: ChainHandle, ChainB: ChainHandle> Link<ChainA, ChainB> {
             .src_chain()
             .query_channel(self.a_to_b.src_port_id(), a_channel_id, Height::default())
             .map_err(|e| {
-                LinkError::channel_not_found(a_channel_id.clone(), self.a_to_b.src_chain().id(), e)
+                LinkError::channel_not_found(
+                    self.a_to_b.src_port_id().clone(),
+                    a_channel_id.clone(),
+                    self.a_to_b.src_chain().id(),
+                    e,
+                )
             })?;
 
         let b_channel_id = self.a_to_b.dst_channel_id();
@@ -65,7 +70,12 @@ impl<ChainA: ChainHandle, ChainB: ChainHandle> Link<ChainA, ChainB> {
             .dst_chain()
             .query_channel(self.a_to_b.dst_port_id(), b_channel_id, Height::default())
             .map_err(|e| {
-                LinkError::channel_not_found(b_channel_id.clone(), self.a_to_b.dst_chain().id(), e)
+                LinkError::channel_not_found(
+                    self.a_to_b.dst_port_id().clone(),
+                    b_channel_id.clone(),
+                    self.a_to_b.dst_chain().id(),
+                    e,
+                )
             })?;
 
         if a_channel.state_matches(&ChannelState::Closed)
@@ -85,9 +95,17 @@ impl<ChainA: ChainHandle, ChainB: ChainHandle> Link<ChainA, ChainB> {
     ) -> Result<Link<ChainA, ChainB>, LinkError> {
         // Check that the packet's channel on source chain is Open
         let a_channel_id = &opts.src_channel_id;
+        let a_port_id = &opts.src_port_id;
         let a_channel = a_chain
-            .query_channel(&opts.src_port_id, a_channel_id, Height::default())
-            .map_err(|e| LinkError::channel_not_found(a_channel_id.clone(), a_chain.id(), e))?;
+            .query_channel(a_port_id, a_channel_id, Height::default())
+            .map_err(|e| {
+                LinkError::channel_not_found(
+                    a_port_id.clone(),
+                    a_channel_id.clone(),
+                    a_chain.id(),
+                    e,
+                )
+            })?;
 
         if !a_channel.state_matches(&ChannelState::Open)
             && !a_channel.state_matches(&ChannelState::Closed)
@@ -162,8 +180,23 @@ impl<ChainA: ChainHandle, ChainB: ChainHandle> Link<ChainA, ChainB> {
         Link::new(channel, with_tx_confirmation)
     }
 
+    /// Constructs a link around the channel that is reverse to the channel
+    /// in this link.
+    pub fn reverse(&self, with_tx_confirmation: bool) -> Result<Link<ChainB, ChainA>, LinkError> {
+        let opts = LinkParameters {
+            src_port_id: self.a_to_b.dst_port_id().clone(),
+            src_channel_id: self.a_to_b.dst_channel_id().clone(),
+        };
+        let chain_b = self.a_to_b.dst_chain().clone();
+        let chain_a = self.a_to_b.src_chain().clone();
+
+        // Some of the checks and initializations may be redundant;
+        // 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(&mut self) -> Result<Vec<IbcEvent>, LinkError> {
+    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(),
@@ -189,7 +222,7 @@ impl<ChainA: ChainHandle, ChainB: ChainHandle> Link<ChainA, ChainB> {
     }
 
     /// Implements the `packet-ack` CLI
-    pub fn build_and_send_ack_packet_messages(&mut self) -> Result<Vec<IbcEvent>, LinkError> {
+    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(),

relayer/src/link/error.rs

@@ -1,6 +1,6 @@
 use flex_error::define_error;
 use ibc::core::ics02_client::error::Error as Ics02Error;
-use ibc::core::ics24_host::identifier::{ChainId, ChannelId};
+use ibc::core::ics24_host::identifier::{ChainId, ChannelId, PortId};
 use ibc::events::IbcEvent;
 use ibc::Height;
 
@@ -45,13 +45,14 @@ define_error! {
 
         ChannelNotFound
             {
+                port_id: PortId,
                 channel_id: ChannelId,
                 chain_id: ChainId,
             }
             [ Error ]
             |e| {
-                format!("channel {} does not exist on chain {}",
-                    e.channel_id, e.chain_id)
+                format!("channel {}/{} does not exist on chain {}",
+                    e.port_id, e.channel_id, e.chain_id)
             },
 
         Connection

relayer/src/link/relay_path.rs

@@ -1121,6 +1121,7 @@ impl<ChainA: ChainHandle, ChainB: ChainHandle> RelayPath<ChainA, ChainB> {
     ) -> Result<Option<Any>, LinkError> {
         let dst_channel_id = self.dst_channel_id();
 
+        debug!("build timeout for channel");
         let (packet_type, next_sequence_received) = if self.ordered_channel() {
             let next_seq = self
                 .dst_chain()

relayer/src/supervisor/error.rs

@@ -16,8 +16,8 @@ define_error! {
                 chain_id: ChainId,
             }
             |e| {
-                format_args!("channel {0} on chain {1} is not open",
-                    e.channel_id, e.chain_id)
+                format_args!("channel {0}/{1} on chain {2} is not open",
+                    e.port_id, e.channel_id, e.chain_id)
             },
 
         ChannelConnectionUninitialized