Fix bug where Subscribe command would sometimes get dropped in the event monitor

.changelog/unreleased/bug-fixes/ibc-relayer/3070-listen-fix.md

@@ -0,0 +1,4 @@
+- Fix bug where one could sometimes not subscribe to events.
+  This mostly affected the `listen` command but also external
+  consumers of events via the `EventMonitor` interface
+  ([\#3070](https://github.com/informalsystems/hermes/issues/3070))

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

@@ -97,18 +97,10 @@ impl Runnable for ListenCmd {
 }
 
 /// Listen to events
-
 #[instrument(skip_all, level = "error", fields(chain = %config.id))]
 pub fn listen(config: &ChainConfig, filters: &[EventFilter]) -> eyre::Result<()> {
     let rt = Arc::new(TokioRuntime::new()?);
-    let (event_monitor, rx) = subscribe(config, rt)?;
-
-    info!(
-        "listening for queries: {}",
-        event_monitor.queries().iter().format(", "),
-    );
-
-    thread::spawn(|| event_monitor.run());
+    let rx = subscribe(config, rt)?;
 
     while let Ok(event_batch) = rx.recv() {
         match event_batch.as_ref() {
@@ -141,10 +133,7 @@ fn event_match(event: &IbcEvent, filters: &[EventFilter]) -> bool {
     filters.iter().any(|f| f.matches(event))
 }
 
-fn subscribe(
-    chain_config: &ChainConfig,
-    rt: Arc<TokioRuntime>,
-) -> eyre::Result<(EventMonitor, Subscription)> {
+fn subscribe(chain_config: &ChainConfig, rt: Arc<TokioRuntime>) -> eyre::Result<Subscription> {
     let (mut event_monitor, tx_cmd) = EventMonitor::new(
         chain_config.id.clone(),
         chain_config.websocket_addr.clone(),
@@ -156,9 +145,13 @@ fn subscribe(
         .init_subscriptions()
         .map_err(|e| eyre!("could not initialize subscriptions: {}", e))?;
 
-    let subscription = tx_cmd.subscribe()?;
+    let queries = event_monitor.queries();
+    info!("listening for queries: {}", queries.iter().format(", "),);
+
+    thread::spawn(|| event_monitor.run());
 
-    Ok((event_monitor, subscription))
+    let subscription = tx_cmd.subscribe()?;
+    Ok(subscription)
 }
 
 #[cfg(test)]

crates/relayer/src/event/monitor.rs

@@ -78,12 +78,12 @@ impl TxMonitorCmd {
 
     pub fn subscribe(&self) -> Result<Subscription> {
         let (tx, rx) = crossbeam_channel::bounded(1);
+
         self.0
             .send(MonitorCmd::Subscribe(tx))
             .map_err(|_| Error::channel_send_failed())?;
 
         let subscription = rx.recv().map_err(|_| Error::channel_recv_failed())?;
-
         Ok(subscription)
     }
 }
@@ -378,7 +378,11 @@ impl EventMonitor {
             if let Ok(cmd) = self.rx_cmd.try_recv() {
                 match cmd {
                     MonitorCmd::Shutdown => return Next::Abort,
-                    MonitorCmd::Subscribe(tx) => tx.send(self.event_bus.subscribe()).unwrap(),
+                    MonitorCmd::Subscribe(tx) => {
+                        if let Err(e) = tx.send(self.event_bus.subscribe()) {
+                            error!("failed to send back subscription: {e}");
+                        }
+                    }
                 }
             }
 
@@ -389,12 +393,6 @@ impl EventMonitor {
                 }
             });
 
-            // Repeat check of shutdown command here, as previous recv()
-            // may block for a long time.
-            if let Ok(MonitorCmd::Shutdown) = self.rx_cmd.try_recv() {
-                return Next::Abort;
-            }
-
             match result {
                 Ok(batch) => self.process_batch(batch),
                 Err(e) => {