Add poll-based event source

.changelog/unreleased/features/ibc-relayer/2850-poll-event-source.md

@@ -0,0 +1,12 @@
+- Add a poll-based event source which fetches events from the chain using
+  the `/block_results` RPC endpoint instead of getting them over WebSocket.
+
+  To use the poll-based event source, set `event_source = 'poll'` in the per-chain configuration.
+
+  **Warning**
+  Only use this if you think Hermes is not getting all
+  the events it should, eg. when relaying for a CosmWasm-enabled blockchain
+  which emits IBC events in a smart contract where the events lack the
+  `message` attribute key. See #3190 and #2809 for more background information.
+
+  ([\#2850](https://github.com/informalsystems/hermes/issues/2850))

ci/misbehaviour/config.toml

@@ -120,7 +120,7 @@ grpc_addr = 'http://127.0.0.1:9090'
 
 # Specify the WebSocket address and port where the chain WebSocket server
 # listens on. Required
-websocket_addr = 'ws://127.0.0.1:26657/websocket'
+event_source = { mode = 'push', url = 'ws://127.0.0.1:26657/websocket', batch_delay = '500ms' }
 
 # Specify the maximum amount of time (duration) that the RPC requests should
 # take before timing out. Default: 10s (10 seconds)
@@ -301,7 +301,7 @@ memo_prefix = ''
 id = 'ibc-1'
 rpc_addr = 'http://127.0.0.1:26557'
 grpc_addr = 'http://127.0.0.1:9091'
-websocket_addr = 'ws://127.0.0.1:26557/websocket'
+event_source = { mode = 'push', url = 'ws://127.0.0.1:26557/websocket', batch_delay = '500ms' }
 rpc_timeout = '10s'
 account_prefix = 'cosmos'
 key_name = 'testkey'

ci/misbehaviour/config_fork.toml

@@ -118,9 +118,8 @@ rpc_addr = 'http://127.0.0.1:26657'
 # Specify the GRPC address and port where the chain GRPC server listens on. Required
 grpc_addr = 'http://127.0.0.1:9090'
 
-# Specify the WebSocket address and port where the chain WebSocket server
-# listens on. Required
-websocket_addr = 'ws://127.0.0.1:26657/websocket'
+# The type of event source to use for getting events from the chain.
+event_source = { mode = 'push', url = 'ws://127.0.0.1:26557/websocket', batch_delay = '500ms' }
 
 # Specify the maximum amount of time (duration) that the RPC requests should
 # take before timing out. Default: 10s (10 seconds)
@@ -301,7 +300,7 @@ memo_prefix = ''
 id = 'ibc-1'
 rpc_addr = 'http://127.0.0.1:26457'
 grpc_addr = 'http://127.0.0.1:9092'
-websocket_addr = 'ws://127.0.0.1:26457/websocket'
+event_source = { mode = 'push', url = 'ws://127.0.0.1:26457/websocket', batch_delay = '500ms' }
 rpc_timeout = '10s'
 account_prefix = 'cosmos'
 key_name = 'testkey'

config.toml

@@ -128,9 +128,34 @@ rpc_addr = 'http://127.0.0.1:26657'
 # Specify the GRPC address and port where the chain GRPC server listens on. Required
 grpc_addr = 'http://127.0.0.1:9090'
 
-# Specify the WebSocket address and port where the chain WebSocket server
-# listens on. Required
-websocket_addr = 'ws://127.0.0.1:26657/websocket'
+# The type of event source to use for getting events from the chain.
+#
+# This setting can take two types of values, as an inline table:
+# 
+# a) Push: for receiving IBC events over WebSocket
+#
+#      `{ mode = 'push', url = 'ws://127.0.0.1:26657/websocket', batch_delay = '500ms' }`
+#
+#    where
+#
+#    - `url` is the WebSocket URL to connect to. Required
+#    - `batch_delay` is the delay until event batch is
+#      emitted in the absence of NewBlock event. Default: 500ms
+#      Lower values will result in faster event processing, improving the latency of Hermes,
+#      but may split the events into more batches than necessary, requiring more client updates
+#      to be submitted, yielding higher costs. Higher values will result in slower event
+#      processing, increasing the latency of Hermes, but are more likely to batch events together.
+#      The default value provides good latency while minimizing the number of client updates needed.
+
+# b) Poll: for polling for IBC events via the `/block_results` RPC endpoint
+#
+#     `{ mode = 'poll', poll_interval = '1s' }`
+#
+#    where
+#
+#    - `poll_interval` is the interval at which to poll for blocks. Default: 1s
+#
+event_source = { mode = 'push', url = 'ws://127.0.0.1:26657/websocket', batch_delay = '500ms' }
 
 # Specify the maximum amount of time (duration) that the RPC requests should
 # take before timing out. Default: 10s (10 seconds)
@@ -149,18 +174,6 @@ rpc_timeout = '10s'
 # Default: false
 trusted_node = false
 
-# Delay until event batch is emitted if no NewBlock events have come yet.
-#
-# Lower values will result in faster event processing, improving the latency of Hermes,
-# but may split the events into more batches than necessary, requiring more client updates
-# to be submitted, yielding higher costs. Higher values will result in slower event
-# processing, increasing the latency of Hermes, but are more likely to batch events together.
-#
-# The default value provides good latency while minimizing the number of client updates needed.
-#
-# Default: 500ms
-batch_delay = '500ms'
-
 # Specify the prefix used by the chain. Required
 account_prefix = 'cosmos'
 
@@ -338,9 +351,8 @@ memo_prefix = ''
 id = 'ibc-1'
 rpc_addr = 'http://127.0.0.1:26557'
 grpc_addr = 'http://127.0.0.1:9091'
-websocket_addr = 'ws://127.0.0.1:26557/websocket'
+event_source = { mode = 'push', url = 'ws://127.0.0.1:26557/websocket', batch_delay = '500ms' }
 rpc_timeout = '10s'
-batch_delay = '500ms'
 trusted_node = false
 account_prefix = 'cosmos'
 key_name = 'testkey'

crates/relayer-cli/src/chain_registry.rs

@@ -19,7 +19,7 @@ use ibc_chain_registry::querier::*;
 use ibc_relayer::config::filter::{FilterPattern, PacketFilter};
 use ibc_relayer::config::gas_multiplier::GasMultiplier;
 use ibc_relayer::config::types::{MaxMsgNum, MaxTxSize, Memo};
-use ibc_relayer::config::{default, AddressType, ChainConfig, GasPrice};
+use ibc_relayer::config::{default, AddressType, ChainConfig, EventSourceMode, GasPrice};
 use ibc_relayer::keyring::Store;
 
 use tendermint_light_client_verifier::types::TrustThreshold;
@@ -107,6 +107,7 @@ where
         MAX_HEALTHY_QUERY_RETRIES,
     )
     .await?;
+
     let websocket_address =
         rpc_data.websocket.clone().try_into().map_err(|e| {
             RegistryError::websocket_url_parse_error(rpc_data.websocket.to_string(), e)
@@ -116,10 +117,12 @@ where
         id: chain_data.chain_id,
         r#type: default::chain_type(),
         rpc_addr: rpc_data.rpc_address,
-        websocket_addr: websocket_address,
         grpc_addr: grpc_address,
+        event_source: EventSourceMode::Push {
+            url: websocket_address,
+            batch_delay: default::batch_delay(),
+        },
         rpc_timeout: default::rpc_timeout(),
-        batch_delay: default::batch_delay(),
         trusted_node: default::trusted_node(),
         genesis_restart: None,
         account_prefix: chain_data.bech32_prefix,

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

@@ -14,7 +14,11 @@ use tendermint_rpc::{client::CompatMode, Client, HttpClient};
 use tokio::runtime::Runtime as TokioRuntime;
 use tracing::{error, info, instrument};
 
-use ibc_relayer::{chain::handle::Subscription, config::ChainConfig, event::monitor::EventMonitor};
+use ibc_relayer::{
+    chain::handle::Subscription,
+    config::{ChainConfig, EventSourceMode},
+    event::source::websocket::EventSource,
+};
 use ibc_relayer_types::{core::ics24_host::identifier::ChainId, events::IbcEvent};
 
 use crate::prelude::*;
@@ -140,23 +144,27 @@ fn subscribe(
     compat_mode: CompatMode,
     rt: Arc<TokioRuntime>,
 ) -> eyre::Result<Subscription> {
-    let (mut event_monitor, tx_cmd) = EventMonitor::new(
+    let EventSourceMode::Push { url, batch_delay } = &chain_config.event_source else {
+        return Err(eyre!("unsupported event source mode, only 'push' is supported for listening to events"));
+    };
+
+    let (mut event_source, tx_cmd) = EventSource::new(
         chain_config.id.clone(),
-        chain_config.websocket_addr.clone(),
+        url.clone(),
         compat_mode,
-        chain_config.batch_delay,
+        *batch_delay,
         rt,
     )
-    .map_err(|e| eyre!("could not initialize event monitor: {}", e))?;
+    .map_err(|e| eyre!("could not initialize event source: {}", e))?;
 
-    event_monitor
+    event_source
         .init_subscriptions()
         .map_err(|e| eyre!("could not initialize subscriptions: {}", e))?;
 
-    let queries = event_monitor.queries();
+    let queries = event_source.queries();
     info!("listening for queries: {}", queries.iter().format(", "),);
 
-    thread::spawn(|| event_monitor.run());
+    thread::spawn(|| event_source.run());
 
     let subscription = tx_cmd.subscribe()?;
     Ok(subscription)