Add configuration to skip packet sequences when clearing (#3862)

* Implement packet clearing filtering logic

* Add tests for packet clearing filter

* Add documentation

* Add changelog entry

* Add excluded sequences to `LinkParameters` struct

* Fix sequence filter by adding setting it to be per-channel

* Skip sequence filter test with Celestia due to the token filter module

* Small refactor

* Small cleanup

---------

Co-authored-by: Romain Ruetschi <romain@informal.systems>

.changelog/unreleased/features/ibc-relayer/3754-filter-packet-clearing.md

@@ -0,0 +1,5 @@
+- Add a per-chain configuration `excluded_sequences` allowing
+  users to specify a list of packet sequences which will not be
+  cleared.
+  This configuration has no impact on standard packet relaying.
+  ([\#3754](https://github.com/informalsystems/hermes/issues/3754))

config.toml

@@ -440,6 +440,18 @@ memo_prefix = ''
 # This will override the global clear interval for this chain only, allowing different intervals for each chain.
 # clear_interval = 50
 
+# Specify packet sequences which should not be cleared, per channel.
+#
+# For each channel, specify a list of sequences which should not be cleared, eg.
+#
+#   excluded_sequences = [
+#     ['channel-0', [1, 2, 3]],
+#     ['channel-1', [4, 5, 6]]
+#  ]
+#
+# Default: No filter
+# excluded_sequences = []
+
 [[chains]]
 id = 'ibc-1'
 rpc_addr = 'http://127.0.0.1:26557'

crates/relayer-cli/src/chain_registry.rs

@@ -1,12 +1,12 @@
 //! Contains functions to generate a relayer config for a given chain
 
-use std::collections::HashMap;
-use std::fmt::Display;
-use std::marker::Send;
-
 use futures::future::join_all;
 use http::Uri;
 use ibc_relayer::config::dynamic_gas::DynamicGasPrice;
+use std::collections::BTreeMap;
+use std::collections::HashMap;
+use std::fmt::Display;
+use std::marker::Send;
 use tokio::task::{JoinError, JoinHandle};
 use tracing::{error, trace};
 
@@ -173,6 +173,7 @@ where
         extension_options: Vec::new(),
         compat_mode: None,
         clear_interval: None,
+        excluded_sequences: BTreeMap::new(),
     }))
 }
 

crates/relayer-cli/src/commands/clear.rs

@@ -5,6 +5,7 @@ use abscissa_core::config::Override;
 use abscissa_core::{Command, FrameworkErrorKind, Runnable};
 
 use ibc_relayer::chain::handle::{BaseChainHandle, ChainHandle};
+use ibc_relayer::chain::requests::{IncludeProof, QueryChannelRequest, QueryHeight};
 use ibc_relayer::config::Config;
 use ibc_relayer::link::error::LinkError;
 use ibc_relayer::link::{Link, LinkParameters};
@@ -146,27 +147,74 @@ impl Runnable for ClearPacketsCmd {
             }
         }
 
-        let mut ev_list = vec![];
+        let (channel, _) = match chains.src.query_channel(
+            QueryChannelRequest {
+                port_id: self.port_id.clone(),
+                channel_id: self.channel_id.clone(),
+                height: QueryHeight::Latest,
+            },
+            IncludeProof::No,
+        ) {
+            Ok(channel) => channel,
+            Err(e) => Output::error(e).exit(),
+        };
+
+        let exclude_src_sequences = config
+            .find_chain(&chains.src.id())
+            .map(|chain_config| chain_config.excluded_sequences(&self.channel_id).to_vec())
+            .unwrap_or_default();
+
+        let exclude_dst_sequences =
+            if let Some(counterparty_channel_id) = channel.counterparty().channel_id() {
+                config
+                    .find_chain(&chains.dst.id())
+                    .map(|chain_config| {
+                        chain_config
+                            .excluded_sequences(counterparty_channel_id)
+                            .to_vec()
+                    })
+                    .unwrap_or_default()
+            } else {
+                Vec::new()
+            };
 
         // Construct links in both directions.
-        let opts = LinkParameters {
+        let fwd_opts = LinkParameters {
             src_port_id: self.port_id.clone(),
             src_channel_id: self.channel_id.clone(),
             max_memo_size: config.mode.packets.ics20_max_memo_size,
             max_receiver_size: config.mode.packets.ics20_max_receiver_size,
+            exclude_src_sequences,
         };
 
-        let fwd_link = match Link::new_from_opts(chains.src.clone(), chains.dst, opts, false, false)
-        {
+        // Construct links in both directions.
+        let reverse_opts = LinkParameters {
+            src_port_id: self.port_id.clone(),
+            src_channel_id: self.channel_id.clone(),
+            max_memo_size: config.mode.packets.ics20_max_memo_size,
+            max_receiver_size: config.mode.packets.ics20_max_receiver_size,
+            exclude_src_sequences: exclude_dst_sequences,
+        };
+
+        let fwd_link = match Link::new_from_opts(
+            chains.src.clone(),
+            chains.dst.clone(),
+            fwd_opts,
+            false,
+            false,
+        ) {
             Ok(link) => link,
             Err(e) => Output::error(e).exit(),
         };
 
-        let rev_link = match fwd_link.reverse(false, false) {
+        let rev_link = match Link::new_from_opts(chains.dst, chains.src, reverse_opts, false, false)
+        {
             Ok(link) => link,
             Err(e) => Output::error(e).exit(),
         };
 
+        let mut ev_list = vec![];
+
         // Schedule RecvPacket messages for pending packets in both directions or,
         // if packet sequences are provided, only on the specified chain.
         // This may produce pending acks which will be processed in the next phase.

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

@@ -88,7 +88,11 @@ impl Runnable for TxPacketRecvCmd {
             src_channel_id: self.src_channel_id.clone(),
             max_memo_size: config.mode.packets.ics20_max_memo_size,
             max_receiver_size: config.mode.packets.ics20_max_receiver_size,
+
+            // Packets are only excluded when clearing
+            exclude_src_sequences: vec![],
         };
+
         let link = match Link::new_from_opts(chains.src, chains.dst, opts, false, false) {
             Ok(link) => link,
             Err(e) => Output::error(e).exit(),
@@ -185,7 +189,11 @@ impl Runnable for TxPacketAckCmd {
             src_channel_id: self.src_channel_id.clone(),
             max_memo_size: config.mode.packets.ics20_max_memo_size,
             max_receiver_size: config.mode.packets.ics20_max_receiver_size,
+
+            // Packets are only excluded when clearing
+            exclude_src_sequences: vec![],
         };
+
         let link = match Link::new_from_opts(chains.src, chains.dst, opts, false, false) {
             Ok(link) => link,
             Err(e) => Output::error(e).exit(),

crates/relayer/src/chain/cosmos.rs

@@ -63,6 +63,7 @@ use tendermint_rpc::endpoint::status;
 use tendermint_rpc::{Client, HttpClient, Order};
 
 use crate::account::Balance;
+use crate::chain::client::ClientSettings;
 use crate::chain::cosmos::batch::{
     send_batched_messages_and_wait_check_tx, send_batched_messages_and_wait_commit,
     sequential_send_batched_messages_and_wait_commit,
@@ -91,6 +92,7 @@ use crate::chain::handle::Subscription;
 use crate::chain::requests::*;
 use crate::chain::tracking::TrackedMsgs;
 use crate::client_state::{AnyClientState, IdentifiedAnyClientState};
+use crate::config::Error as ConfigError;
 use crate::config::{parse_gas_prices, ChainConfig, GasPrice};
 use crate::consensus_state::AnyConsensusState;
 use crate::denom::DenomTrace;
@@ -102,10 +104,10 @@ use crate::light_client::tendermint::LightClient as TmLightClient;
 use crate::light_client::{LightClient, Verified};
 use crate::misbehaviour::MisbehaviourEvidence;
 use crate::util::compat_mode::compat_mode_from_version;
+use crate::util::pretty::PrettySlice;
 use crate::util::pretty::{
     PrettyIdentifiedChannel, PrettyIdentifiedClientState, PrettyIdentifiedConnection,
 };
-use crate::{chain::client::ClientSettings, config::Error as ConfigError};
 
 use self::gas::dynamic_gas_price;
 use self::types::app_state::GenesisAppState;
@@ -671,8 +673,7 @@ impl CosmosSdkChain {
 
         let mut client = self
             .block_on(ServiceClient::connect(grpc_addr.clone()))
-            .map_err(Error::grpc_transport)
-            .unwrap();
+            .map_err(Error::grpc_transport)?;
 
         let request = tonic::Request::new(GetSyncingRequest {});
 
@@ -2430,6 +2431,17 @@ fn do_health_check(chain: &CosmosSdkChain) -> Result<(), Error> {
     let grpc_address = chain.grpc_addr.to_string();
     let rpc_address = chain.config.rpc_addr.to_string();
 
+    if !chain.config.excluded_sequences.is_empty() {
+        for (channel_id, seqs) in chain.config.excluded_sequences.iter() {
+            if !seqs.is_empty() {
+                warn!(
+                    "chain '{chain_id}' will not clear packets on channel '{channel_id}' with sequences: {}. \
+                    Ignore this warning if this configuration is correct.", PrettySlice(seqs)
+                );
+            }
+        }
+    }
+
     chain.block_on(chain.rpc_client.health()).map_err(|e| {
         Error::health_check_json_rpc(
             chain_id.clone(),

crates/relayer/src/chain/cosmos/config.rs

@@ -1,12 +1,14 @@
 use core::time::Duration;
+use std::collections::BTreeMap;
 use std::path::PathBuf;
 
 use byte_unit::Byte;
+use ibc_relayer_types::core::ics04_channel::packet::Sequence;
 use serde_derive::{Deserialize, Serialize};
 use tendermint_rpc::Url;
 
 use ibc_relayer_types::core::ics23_commitment::specs::ProofSpecs;
-use ibc_relayer_types::core::ics24_host::identifier::ChainId;
+use ibc_relayer_types::core::ics24_host::identifier::{ChainId, ChannelId};
 
 use crate::chain::cosmos::config::error::Error as ConfigError;
 use crate::config::compat_mode::CompatMode;
@@ -146,6 +148,8 @@ pub struct CosmosSdkConfig {
     pub extension_options: Vec<ExtensionOption>,
     pub compat_mode: Option<CompatMode>,
     pub clear_interval: Option<u64>,
+    #[serde(default)]
+    pub excluded_sequences: BTreeMap<ChannelId, Vec<Sequence>>,
 }
 
 impl CosmosSdkConfig {

crates/relayer/src/config.rs

@@ -14,6 +14,8 @@ use core::cmp::Ordering;
 use core::fmt::{Display, Error as FmtError, Formatter};
 use core::str::FromStr;
 use core::time::Duration;
+use ibc_relayer_types::core::ics04_channel::packet::Sequence;
+use std::borrow::Cow;
 use std::{fs, fs::File, io::Write, ops::Range, path::Path};
 
 use byte_unit::Byte;
@@ -703,6 +705,16 @@ impl ChainConfig {
             Self::CosmosSdk(config) => config.query_packets_chunk_size = query_packets_chunk_size,
         }
     }
+
+    pub fn excluded_sequences(&self, channel_id: &ChannelId) -> Cow<'_, [Sequence]> {
+        match self {
+            Self::CosmosSdk(config) => config
+                .excluded_sequences
+                .get(channel_id)
+                .map(|seqs| Cow::Borrowed(seqs.as_slice()))
+                .unwrap_or_else(|| Cow::Owned(Vec::new())),
+        }
+    }
 }
 
 // /!\ Update me when adding a new chain type!

crates/relayer/src/link.rs

@@ -1,6 +1,7 @@
 use ibc_relayer_types::core::{
     ics03_connection::connection::State as ConnectionState,
     ics04_channel::channel::State as ChannelState,
+    ics04_channel::packet::Sequence,
     ics24_host::identifier::{ChannelId, PortChannelId, PortId},
 };
 use tracing::info;
@@ -38,6 +39,7 @@ pub struct LinkParameters {
     pub src_channel_id: ChannelId,
     pub max_memo_size: Ics20FieldSizeLimit,
     pub max_receiver_size: Ics20FieldSizeLimit,
+    pub exclude_src_sequences: Vec<Sequence>,
 }
 
 pub struct Link<ChainA: ChainHandle, ChainB: ChainHandle> {
@@ -177,31 +179,4 @@ impl<ChainA: ChainHandle, ChainB: ChainHandle> Link<ChainA, ChainB> {
 
         Link::new(channel, with_tx_confirmation, opts)
     }
-
-    /// Constructs a link around the channel that is reverse to the channel
-    /// in this link.
-    pub fn reverse(
-        &self,
-        with_tx_confirmation: bool,
-        auto_register_counterparty_payee: 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(),
-            max_memo_size: self.a_to_b.max_memo_size,
-            max_receiver_size: self.a_to_b.max_receiver_size,
-        };
-        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,
-            auto_register_counterparty_payee,
-        )
-    }
 }

crates/relayer/src/link/relay_path.rs

@@ -1,5 +1,6 @@
 use alloc::collections::BTreeMap as HashMap;
 use alloc::collections::VecDeque;
+use ibc_relayer_types::core::ics04_channel::packet::Sequence;
 use std::ops::Sub;
 use std::time::{Duration, Instant};
 
@@ -115,6 +116,7 @@ pub struct RelayPath<ChainA: ChainHandle, ChainB: ChainHandle> {
 
     pub max_memo_size: Ics20FieldSizeLimit,
     pub max_receiver_size: Ics20FieldSizeLimit,
+    pub exclude_src_sequences: Vec<Sequence>,
 }
 
 impl<ChainA: ChainHandle, ChainB: ChainHandle> RelayPath<ChainA, ChainB> {
@@ -163,6 +165,8 @@ impl<ChainA: ChainHandle, ChainB: ChainHandle> RelayPath<ChainA, ChainB> {
 
             max_memo_size: link_parameters.max_memo_size,
             max_receiver_size: link_parameters.max_receiver_size,
+
+            exclude_src_sequences: link_parameters.exclude_src_sequences,
         })
     }
 
@@ -1155,6 +1159,12 @@ impl<ChainA: ChainHandle, ChainB: ChainHandle> RelayPath<ChainA, ChainB> {
             return Ok(());
         }
 
+        // Retain only sequences which should not be filtered out
+        let sequences: Vec<Sequence> = sequences
+            .into_iter()
+            .filter(|sequence| !self.exclude_src_sequences.contains(sequence))
+            .collect();
+
         debug!(
             dst_chain = %self.dst_chain().id(),
             src_chain = %self.src_chain().id(),
@@ -1219,6 +1229,12 @@ impl<ChainA: ChainHandle, ChainB: ChainHandle> RelayPath<ChainA, ChainB> {
             return Ok(());
         }
 
+        // Retain only sequences which should not be filtered out
+        let sequences: Vec<Sequence> = sequences
+            .into_iter()
+            .filter(|sequence| !self.exclude_src_sequences.contains(sequence))
+            .collect();
+
         debug!(
             dst_chain = %self.dst_chain().id(),
             src_chain = %self.src_chain().id(),

crates/relayer/src/worker.rs

@@ -110,6 +110,12 @@ pub fn spawn_worker_tasks<ChainA: ChainHandle, ChainB: ChainHandle>(
             (Some(cmd_tx), None)
         }
         Object::Packet(path) => {
+            let exclude_src_sequences = config
+                .find_chain(&chains.a.id())
+                .map(|chain_config| chain_config.excluded_sequences(&path.src_channel_id))
+                .unwrap_or_default()
+                .to_vec();
+
             let packets_config = config.mode.packets;
             let link_res = Link::new_from_opts(
                 chains.a.clone(),
@@ -119,6 +125,7 @@ pub fn spawn_worker_tasks<ChainA: ChainHandle, ChainB: ChainHandle>(
                     src_channel_id: path.src_channel_id.clone(),
                     max_memo_size: packets_config.ics20_max_memo_size,
                     max_receiver_size: packets_config.ics20_max_receiver_size,
+                    exclude_src_sequences,
                 },
                 packets_config.tx_confirmation,
                 packets_config.auto_register_counterparty_payee,