Fix bug where only first CCQ event was extracted (#3955)

* Fix create_fork.sh used in misbehaviour test

* Fix client upgrade test

* Fix 'interchainquery' event extraction to extract all events instead of only the first one

* Add changelog entry

* Update .changelog/unreleased/bug-fixes/ibc-relayer/3954-interchainquery-missed-events.md

Co-authored-by: Romain Ruetschi <[email protected]>
Signed-off-by: Luca Joss <[email protected]>

* ICS 031 event extraction perf improvement and refactor (#3956)

* Small performance optimizations

* Re-use existing `append_events` infrastructure

* Cleanup

* Add the flag '--reject-config-defaults' to the start command for Osmosis

---------

Co-authored-by: Luca Joss <[email protected]>
Co-authored-by: Luca Joss <[email protected]>

---------

Signed-off-by: Luca Joss <[email protected]>
Co-authored-by: Romain Ruetschi <[email protected]>

.changelog/unreleased/bug-fixes/ibc-relayer/3954-interchainquery-missed-events.md

@@ -0,0 +1,2 @@
+- Fix a bug where Hermes would only ever extract the first emitted ICS 031 CrossChain Query event, which would cause it to miss the other CCQ events.
+  ([\#3954](https://github.com/informalsystems/hermes/issues/3954))

crates/relayer-types/src/applications/ics31_icq/events.rs

@@ -1,4 +1,3 @@
-use std::collections::BTreeMap;
 use std::str::FromStr;
 
 use serde::{Deserialize, Serialize};
@@ -9,7 +8,7 @@ use crate::events::IbcEvent;
 
 use super::error::Error;
 
-const EVENT_TYPE_PREFIX: &str = "query_request";
+pub const EVENT_TYPE_PREFIX: &str = "query_request";
 
 #[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq)]
 pub struct CrossChainQueryPacket {
@@ -23,6 +22,12 @@ pub struct CrossChainQueryPacket {
     pub request: String,
 }
 
+impl From<CrossChainQueryPacket> for IbcEvent {
+    fn from(packet: CrossChainQueryPacket) -> Self {
+        IbcEvent::CrossChainQueryPacket(packet)
+    }
+}
+
 fn find_value<'a>(key: &str, entries: &'a [abci::EventAttribute]) -> Result<&'a str, Error> {
     entries
         .iter()
@@ -93,66 +98,3 @@ impl<'a> TryFrom<&'a [abci::EventAttribute]> for CrossChainQueryPacket {
         })
     }
 }
-
-fn fetch_first_element_from_events(
-    block_events: &BTreeMap<String, Vec<String>>,
-    key: &str,
-) -> Result<String, Error> {
-    let res = block_events
-        .get(key)
-        .ok_or_else(|| Error::event(format!("attribute not found for key: {key}")))?
-        .first()
-        .ok_or_else(|| {
-            Error::event(format!(
-                "element at position 0, of attribute with key `{key}`, not found"
-            ))
-        })?;
-
-    Ok(res.clone())
-}
-
-impl CrossChainQueryPacket {
-    pub fn extract_query_event(
-        block_events: &BTreeMap<String, Vec<String>>,
-    ) -> Result<IbcEvent, Error> {
-        let chain_id_str = fetch_first_element_from_events(
-            block_events,
-            &format!("{}.{}", EVENT_TYPE_PREFIX, "chain_id"),
-        )?;
-        let connection_id_str = fetch_first_element_from_events(
-            block_events,
-            &format!("{}.{}", EVENT_TYPE_PREFIX, "connection_id"),
-        )?;
-        let query_type = fetch_first_element_from_events(
-            block_events,
-            &format!("{}.{}", EVENT_TYPE_PREFIX, "type"),
-        )?;
-        let height_str = fetch_first_element_from_events(
-            block_events,
-            &format!("{}.{}", EVENT_TYPE_PREFIX, "height"),
-        )?;
-
-        Ok(IbcEvent::CrossChainQueryPacket(CrossChainQueryPacket {
-            module: fetch_first_element_from_events(
-                block_events,
-                &format!("{}.{}", EVENT_TYPE_PREFIX, "module"),
-            )?,
-            action: fetch_first_element_from_events(
-                block_events,
-                &format!("{}.{}", EVENT_TYPE_PREFIX, "action"),
-            )?,
-            query_id: fetch_first_element_from_events(
-                block_events,
-                &format!("{}.{}", EVENT_TYPE_PREFIX, "query_id"),
-            )?,
-            chain_id: ChainId::from_string(&chain_id_str),
-            connection_id: ConnectionId::from_str(&connection_id_str)?,
-            query_type,
-            height: Height::from_str(&height_str)?,
-            request: fetch_first_element_from_events(
-                block_events,
-                &format!("{}.{}", EVENT_TYPE_PREFIX, "request"),
-            )?,
-        }))
-    }
-}

crates/relayer/src/chain/cosmos/types/events/channel.rs

@@ -1,5 +1,6 @@
 use alloc::collections::btree_map::BTreeMap as HashMap;
 
+use ibc_relayer_types::applications::ics31_icq::events::CrossChainQueryPacket;
 use ibc_relayer_types::core::ics02_client::height::HeightErrorDetail;
 use ibc_relayer_types::core::ics04_channel::error::Error;
 use ibc_relayer_types::core::ics04_channel::events::{
@@ -12,6 +13,7 @@ use ibc_relayer_types::core::ics04_channel::events::{
 use ibc_relayer_types::core::ics04_channel::events::{ReceivePacket, TimeoutOnClosePacket};
 use ibc_relayer_types::core::ics04_channel::packet::Packet;
 use ibc_relayer_types::core::ics04_channel::timeout::TimeoutHeight;
+use ibc_relayer_types::core::ics24_host::identifier::ChainId;
 use ibc_relayer_types::events::Error as EventError;
 use ibc_relayer_types::Height;
 
@@ -176,6 +178,28 @@ impl TryFrom<RawObject<'_>> for Packet {
     }
 }
 
+impl TryFrom<RawObject<'_>> for CrossChainQueryPacket {
+    type Error = EventError;
+
+    fn try_from(obj: RawObject<'_>) -> Result<Self, Self::Error> {
+        Ok(Self {
+            module: extract_attribute(&obj, &format!("{}.module", obj.action))?,
+            action: extract_attribute(&obj, &format!("{}.action", obj.action))?,
+            query_id: extract_attribute(&obj, &format!("{}.query_id", obj.action))?,
+            chain_id: extract_attribute(&obj, &format!("{}.chain_id", obj.action))
+                .map(|s| ChainId::from_string(&s))?,
+            connection_id: extract_attribute(&obj, &format!("{}.connection_id", obj.action))?
+                .parse()
+                .map_err(EventError::parse)?,
+            query_type: extract_attribute(&obj, &format!("{}.type", obj.action))?,
+            request: extract_attribute(&obj, &format!("{}.request", obj.action))?,
+            height: extract_attribute(&obj, &format!("{}.height", obj.action))?
+                .parse()
+                .map_err(|_| EventError::height())?,
+        })
+    }
+}
+
 #[derive(Debug, Clone)]
 pub struct RawObject<'a> {
     pub height: Height,

crates/relayer/src/event/source/rpc/extract.rs

@@ -1,6 +1,6 @@
-use ibc_relayer_types::applications::ics29_fee::events::DistributionType;
 use tendermint::abci;
 
+use ibc_relayer_types::applications::ics29_fee::events::DistributionType;
 use ibc_relayer_types::core::ics02_client::height::Height;
 use ibc_relayer_types::core::ics24_host::identifier::ChainId;
 use ibc_relayer_types::events::IbcEvent;

crates/relayer/src/event/source/websocket/extract.rs

@@ -1,6 +1,7 @@
 use alloc::collections::BTreeMap as HashMap;
 use ibc_relayer_types::applications::ics29_fee::events::DistributionType;
 
+use ibc_relayer_types::applications::ics31_icq;
 use tendermint_rpc::{event::Event as RpcEvent, event::EventData as RpcEventData};
 
 use ibc_relayer_types::applications::ics31_icq::events::CrossChainQueryPacket;
@@ -328,10 +329,17 @@ fn extract_block_events(
         extract_events(height, block_events, "channel_close_confirm", "channel_id"),
         height,
     );
-    // extract cross chain query event from block_events
-    if let Ok(ccq) = CrossChainQueryPacket::extract_query_event(block_events) {
-        events.push(IbcEventWithHeight::new(ccq, height));
-    }
+
+    append_events::<CrossChainQueryPacket>(
+        &mut events,
+        extract_events(
+            height,
+            block_events,
+            ics31_icq::events::EVENT_TYPE_PREFIX,
+            "query_id",
+        ),
+        height,
+    );
 
     events
 }

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

@@ -17,6 +17,9 @@ use ibc_test_framework::chain::config::{
 use ibc_test_framework::chain::ext::crosschainquery::CrossChainQueryMethodsExt;
 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, query_identified_channel_ends,
+};
 use ibc_test_framework::util::interchain_security::{
     update_genesis_for_consumer_chain, update_relayer_config_for_consumer_chain,
 };
@@ -138,6 +141,33 @@ impl BinaryChannelTest for InterchainSecurityIcqTest {
             &wallet_b.0.id.to_string(),
         )?;
 
+        // Wait for channel to initialise so that the query can find
+        // all the channels related to registering a host-zone
+        std::thread::sleep(Duration::from_secs(5));
+
+        let channels = query_identified_channel_ends::<ChainA, ChainB>(chains.handle_a())?;
+
+        // Wait for channel created by registering a host-zone to be Open
+        for channel in channels.iter() {
+            let tagged_channel_id: TaggedChannelId<ChainA, ChainB> =
+                DualTagged::new(channel.0.channel_id.clone());
+            let tagged_port_id: TaggedPortId<ChainA, ChainB> =
+                DualTagged::new(channel.0.port_id.clone());
+
+            if channel.0.port_id.as_str() == "icahost" {
+                info!(
+                    "Will assert that channel {}/{} is eventually Open",
+                    channel.0.channel_id, channel.0.port_id
+                );
+                assert_eventually_channel_established(
+                    chains.handle_a(),
+                    chains.handle_b(),
+                    &tagged_channel_id.as_ref(),
+                    &tagged_port_id.as_ref(),
+                )?;
+            }
+        }
+
         // Wait for the cross chain query to be pending.
         chains
             .node_b

tools/test-framework/src/chain/chain_type.rs

@@ -12,6 +12,7 @@ const PROVENANCE_HD_PATH: &str = "m/44'/505'/0'/0/0";
 #[derive(Clone, Debug)]
 pub enum ChainType {
     Cosmos,
+    Osmosis,
     Evmos,
     Provenance,
     Injective,
@@ -20,15 +21,15 @@ pub enum ChainType {
 impl ChainType {
     pub fn hd_path(&self) -> &str {
         match self {
-            Self::Cosmos => COSMOS_HD_PATH,
+            Self::Cosmos | Self::Osmosis => COSMOS_HD_PATH,
             Self::Evmos | Self::Injective => EVMOS_HD_PATH,
             Self::Provenance => PROVENANCE_HD_PATH,
         }
     }
 
     pub fn chain_id(&self, prefix: &str, use_random_id: bool) -> ChainId {
         match self {
-            Self::Cosmos => {
+            Self::Cosmos | Self::Osmosis => {
                 if use_random_id {
                     ChainId::from_string(&format!("ibc-{}-{:x}", prefix, random_u32()))
                 } else {
@@ -47,6 +48,9 @@ impl ChainType {
         let json_rpc_port = random_unused_tcp_port();
         match self {
             Self::Cosmos | Self::Injective | Self::Provenance => {}
+            Self::Osmosis => {
+                res.push("--reject-config-defaults".to_owned());
+            }
             Self::Evmos => {
                 res.push("--json-rpc.address".to_owned());
                 res.push(format!("localhost:{json_rpc_port}"));
@@ -59,7 +63,7 @@ impl ChainType {
     pub fn extra_add_genesis_account_args(&self, chain_id: &ChainId) -> Vec<String> {
         let mut res = vec![];
         match self {
-            Self::Cosmos | Self::Evmos | Self::Provenance => {}
+            Self::Cosmos | Self::Osmosis | Self::Evmos | Self::Provenance => {}
             Self::Injective => {
                 res.push("--chain-id".to_owned());
                 res.push(format!("{chain_id}"));
@@ -70,7 +74,7 @@ impl ChainType {
 
     pub fn address_type(&self) -> AddressType {
         match self {
-            Self::Cosmos | Self::Provenance => AddressType::default(),
+            Self::Cosmos | Self::Osmosis | Self::Provenance => AddressType::default(),
             Self::Evmos => AddressType::Ethermint {
                 pk_type: "/ethermint.crypto.v1.ethsecp256k1.PubKey".to_string(),
             },
@@ -89,6 +93,7 @@ impl FromStr for ChainType {
             name if name.contains("evmosd") => Ok(ChainType::Evmos),
             name if name.contains("injectived") => Ok(ChainType::Injective),
             name if name.contains("provenanced") => Ok(ChainType::Provenance),
+            name if name.contains("osmosisd") => Ok(ChainType::Osmosis),
             _ => Ok(ChainType::Cosmos),
         }
     }

tools/test-framework/src/chain/cli/host_zone.rs

@@ -31,7 +31,7 @@ pub fn register_host_zone(
             bech32_prefix,
             ibc_denom,
             channel_id,
-            "5",
+            "10",
             "false",
             "--from",
             sender,

tools/test-framework/src/chain/ext/crosschainquery.rs

@@ -72,14 +72,13 @@ impl<'a, Chain: Send> CrossChainQueryMethodsExt<Chain> for MonoTagged<Chain, &'a
                 )?;
 
                 // Verify that the there are no more pending Cross Chain Queries.
-                if json::from_str::<json::Value>(&output)
+                if !json::from_str::<json::Value>(&output)
                     .map_err(handle_generic_error)?
                     .get("pending_queries")
                     .ok_or_else(|| eyre!("no pending cross chain queries"))?
                     .as_array()
                     .ok_or_else(|| eyre!("pending cross chain queries is not an array"))?
-                    .first()
-                    .is_some()
+                    .is_empty()
                 {
                     return Err(Error::generic(eyre!(
                         "Pending query has not been processed"

tools/test-framework/src/chain/ext/proposal.rs

@@ -70,10 +70,10 @@ pub async fn query_upgrade_proposal_height(
         .map(|r| r.into_inner())
         .map_err(|e| RelayerError::grpc_status(e, "query_upgrade_proposal_height".to_owned()))?;
 
-    // Querying for a balance might fail, i.e. if the account doesn't actually exist
+    // Querying for proposal might not exist if the proposal id is incorrect
     let proposal = response
         .proposal
-        .ok_or_else(|| RelayerError::empty_query_account(proposal_id.to_string()))?;
+        .ok_or_else(|| RelayerError::empty_proposal(proposal_id.to_string()))?;
 
     let proposal_content = proposal
         .content

tools/test-framework/src/relayer/channel.rs

@@ -1,7 +1,9 @@
 use core::time::Duration;
 use eyre::eyre;
 use ibc_relayer::chain::handle::ChainHandle;
-use ibc_relayer::chain::requests::{IncludeProof, QueryChannelRequest, QueryHeight};
+use ibc_relayer::chain::requests::{
+    IncludeProof, QueryChannelRequest, QueryChannelsRequest, QueryHeight,
+};
 use ibc_relayer::channel::version::Version;
 use ibc_relayer::channel::{extract_channel_id, Channel, ChannelSide};
 use ibc_relayer_types::core::ics04_channel::channel::State as ChannelState;
@@ -207,6 +209,23 @@ pub fn query_identified_channel_end<ChainA: ChainHandle, ChainB>(
     )))
 }
 
+pub fn query_identified_channel_ends<ChainA: ChainHandle, ChainB>(
+    handle: &ChainA,
+) -> Result<Vec<DualTagged<ChainA, ChainB, IdentifiedChannelEnd>>, Error> {
+    let channel_ends = handle.query_channels(QueryChannelsRequest { pagination: None })?;
+    let identified_channels = channel_ends
+        .iter()
+        .map(|c| {
+            DualTagged::new(IdentifiedChannelEnd::new(
+                c.port_id.clone(),
+                c.channel_id.clone(),
+                c.channel_end.clone(),
+            ))
+        })
+        .collect();
+    Ok(identified_channels)
+}
+
 pub fn assert_eventually_channel_established<ChainA: ChainHandle, ChainB: ChainHandle>(
     handle_a: &ChainA,
     handle_b: &ChainB,