Ensure all events in a batch are processed, even when it contains events for unknown chains

[ljoss17]

Closes: #4021
Closes: #4034

Description

Note: This PR also bumps all the MSRVs to v1.76.0

---

PR author checklist:

• Added changelog entry, using unclog.
• Added tests: integration (for Hermes) or unit/mock tests (for modules).
• Linked to GitHub issue.
• Updated code comments and documentation (e.g., docs/).
• Tagged one reviewer who will be the one responsible for shepherding this PR.

Reviewer checklist:

• Reviewed Files changed in the GitHub PR explorer.
• Manually tested (in case integration/unit/mock tests are absent).

[ancazamfir]

Curious why we didn't implement something like client_state_filter.control_query_object() here in the past?

[romac]

An oversight. We could also implement that check there but right it's so simple that it does not require the whole filtering machinery.

[ancazamfir]

sure but the added check is not really a "client_filter_outcome". Not sure if we tested the scenario with this PR branch in the issue but it looks like if the chain is not configured we will return an error from relay_on_object() and print a misleading trace.

[ancazamfir]

Related to my other comment, why can't we do the config check for object.dst_chain_id() before relay_on_object call ?
Then do proper client filtering in relay_on_object and also fix that trace message.

[ancazamfir]

Another thing, without the fix I think we hit this line and we exit without processing the other events in the batch, maybe we need to log an error and continue.

[romac]

Related to my other comment, why can't we do the config check for object.dst_chain_id() before relay_on_object call ?
Then do proper client filtering in relay_on_object and also fix that trace message

Agreed, seems like the best way to go

[romac]

@ancazamfir What do you think of this patch? It should address all the problems you've raised.

Show diff

diff --git a/crates/relayer/src/object.rs b/crates/relayer/src/object.rs
index 4716a5d5c..aac3efb25 100644
--- a/crates/relayer/src/object.rs
+++ b/crates/relayer/src/object.rs
@@ -21,9 +21,7 @@ use crate::chain::{
     handle::ChainHandle,
     requests::{IncludeProof, QueryClientStateRequest, QueryHeight},
 };
-use crate::config::ChainConfig;
 use crate::error::Error as RelayerError;
-use crate::supervisor::client_state_filter::Permission;
 use crate::supervisor::Error as SupervisorError;
 
 /// Client
@@ -147,17 +145,6 @@ impl CrossChainQuery {
     pub fn short_name(&self) -> String {
         format!("cross_chain_query::{}/{}", self.dst_chain_id, self.query_id)
     }
-
-    pub fn intended_for_known_dst_chain(&self, chain_configs: &[ChainConfig]) -> Permission {
-        if chain_configs
-            .iter()
-            .any(|config| *config.id() == self.dst_chain_id)
-        {
-            Permission::Allow
-        } else {
-            Permission::Deny
-        }
-    }
 }
 
 /// An object determines the amount of parallelism that can
diff --git a/crates/relayer/src/supervisor.rs b/crates/relayer/src/supervisor.rs
index 9a5b0f17a..478aecb84 100644
--- a/crates/relayer/src/supervisor.rs
+++ b/crates/relayer/src/supervisor.rs
@@ -381,7 +381,7 @@ fn relay_on_object<Chain: ChainHandle>(
         Object::Connection(conn) => client_state_filter.control_conn_object(registry, conn),
         Object::Channel(chan) => client_state_filter.control_chan_object(registry, chan),
         Object::Packet(packet) => client_state_filter.control_packet_object(registry, packet),
-        Object::CrossChainQuery(ccq) => Ok(ccq.intended_for_known_dst_chain(&config.chains)),
+        Object::CrossChainQuery(_ccq) => Ok(Permission::Allow),
         Object::Wallet(_wallet) => Ok(Permission::Allow),
     };
 
@@ -816,8 +816,33 @@ fn process_batch<Chain: ChainHandle>(
         workers.notify_new_block(&src_chain.id(), batch.height, new_block);
     }
 
-    // Forward the IBC events.
+    // Forward the IBC events to the appropriate workers
     for (object, events_with_heights) in collected.per_object.into_iter() {
+        if events_with_heights.is_empty() {
+            // Event batch is empty, nothing to do
+            continue;
+        }
+
+        let Ok(src_chain) = registry.get_or_spawn(object.src_chain_id()) else {
+            trace!(
+                "skipping events for '{}': source chain '{}' is not registered",
+                object.short_name(),
+                object.src_chain_id()
+            );
+
+            continue;
+        };
+
+        let Ok(dst_chain) = registry.get_or_spawn(object.dst_chain_id()) else {
+            trace!(
+                "skipping events for '{}': destination chain '{}' is not registered",
+                object.short_name(),
+                object.src_chain_id()
+            );
+
+            continue;
+        };
+
         if !relay_on_object(
             config,
             registry,
@@ -826,32 +851,23 @@ fn process_batch<Chain: ChainHandle>(
             &object,
         ) {
             trace!(
-                "skipping events for '{}'. \
-                reason: filtering is enabled and channel does not match any allowed channels",
+                "skipping events for '{}': filtering is enabled and channel does not match any allowed channels",
                 object.short_name()
             );
 
             continue;
         }
 
-        if events_with_heights.is_empty() {
-            continue;
-        }
-
-        let src = registry
-            .get_or_spawn(object.src_chain_id())
-            .map_err(Error::spawn)?;
-
-        let dst = registry
-            .get_or_spawn(object.dst_chain_id())
-            .map_err(Error::spawn)?;
-
         if let Object::Packet(ref _path) = object {
-            // Update telemetry info
-            telemetry!(send_telemetry(&src, &dst, &events_with_heights, _path));
+            telemetry!(send_telemetry(
+                &src_chain,
+                &dst_chain,
+                &events_with_heights,
+                _path
+            ));
         }
 
-        let worker = workers.get_or_spawn(object, src, dst, config);
+        let worker = workers.get_or_spawn(object, src_chain, dst_chain, config);
 
         worker.send_events(
             batch.height,

[ancazamfir]

Looks great! The skipping events trace message should be more general imo, it's not only channel filters that can dictate an event should be ignored, it's filtering in general.

[ancazamfir]

Perhaps also change the issue/PR to make it more general as it applies to all events

[romac]

Created a new issue and added a new changelog entry

[ancazamfir]

great work @ljoss17 and @romac 🚀