fix(event-source): handle websocket stream termination and add watchdog

crates/relayer/src/event/error.rs

@@ -48,6 +48,16 @@ define_error! {
         Rpc
             [ TraceError<RpcError> ]
             |_| { "RPC error" },
+
+        StreamTerminated
+            |_| { "event stream terminated unexpectedly" },
+
+        WatchdogTimeout
+            { timeout_secs: u64 }
+            |e| { format!("no events received for {} seconds", e.timeout_secs) },
+
+        DriverChannelClosed
+            |_| { "websocket driver channel closed" },
     }
 }
 

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

@@ -11,8 +11,9 @@ use futures::{
     Stream, TryStreamExt,
 };
 use tokio::task::JoinHandle;
+use tokio::time::sleep;
 use tokio::{runtime::Runtime as TokioRuntime, sync::mpsc};
-use tracing::{debug, error, info, instrument, trace};
+use tracing::{debug, error, info, instrument, trace, warn};
 
 use tendermint_rpc::{
     client::CompatMode, event::Event as RpcEvent, query::Query, SubscriptionClient,
@@ -24,6 +25,7 @@ use ibc_relayer_types::{core::ics24_host::identifier::ChainId, events::IbcEvent}
 use crate::{
     chain::tracking::TrackingId,
     event::{bus::EventBus, error::*, IbcEventWithHeight},
+    link::TIMEOUT as PENDING_TIMEOUT,
     telemetry,
     util::{
         retry::{retry_with_index, RetryResult},
@@ -312,8 +314,43 @@ impl EventSource {
             }
 
             let result = tokio::select! {
-                Some(batch) = batches.next() => batch,
-                Some(e) = self.rx_err.recv() => Err(Error::web_socket_driver(e)),
+                batch_opt = batches.next() => {
+                    match batch_opt {
+                        Some(batch) => batch,
+                        None => {
+                            // Stream terminated without error - this indicates a silent
+                            // WebSocket failure where the subscription ended unexpectedly.
+                            // Propagate error so supervisor clears pending packets.
+                            warn!("event batch stream terminated unexpectedly, triggering reconnect");
+                            self.propagate_error(Error::stream_terminated());
+                            return Next::Reconnect;
+                        }
+                    }
+                }
+                err_opt = self.rx_err.recv() => {
+                    match err_opt {
+                        Some(e) => Err(Error::web_socket_driver(e)),
+                        None => {
+                            // Error channel closed - driver has terminated.
+                            // Propagate error so supervisor clears pending packets.
+                            warn!("websocket driver channel closed, triggering reconnect");
+                            self.propagate_error(Error::driver_channel_closed());
+                            return Next::Reconnect;
+                        }
+                    }
+                }
+                _ = sleep(PENDING_TIMEOUT) => {
+                    // No events received within the watchdog timeout period.
+                    // This likely indicates a stale connection that is not receiving
+                    // events despite appearing connected. Propagate error so supervisor
+                    // clears pending packets, then trigger reconnection.
+                    warn!(
+                        "no events received for {} seconds, assuming stale connection",
+                        PENDING_TIMEOUT.as_secs()
+                    );
+                    self.propagate_error(Error::watchdog_timeout(PENDING_TIMEOUT.as_secs()));
+                    return Next::Reconnect;
+                }
             };
 
             // Before handling the batch, check if there are any pending shutdown or subscribe commands.

crates/relayer/src/link.rs

@@ -32,6 +32,9 @@ use tx_hashes::TxHashes;
 // Re-export the telemetries summary
 pub use relay_summary::RelaySummary;
 
+/// Timeout duration used for pending operations (5 minutes)
+pub use pending::TIMEOUT;
+
 pub use relay_path::{RelayPath, Resubmit};
 
 #[derive(Clone, Debug)]

crates/relayer/src/supervisor.rs

@@ -969,6 +969,24 @@ fn handle_batch<Chain: ChainHandle>(
             let _ = clear_pending_packets(workers, &chain_id)
                 .map_err(|e| error!("error during clearing pending packets: {}", e));
         }
+        Err(EventError(EventErrorDetail::StreamTerminated(_), _)) => {
+            warn!("event stream terminated unexpectedly, clearing pending packets");
+
+            let _ = clear_pending_packets(workers, &chain_id)
+                .map_err(|e| error!("error during clearing pending packets: {}", e));
+        }
+        Err(EventError(EventErrorDetail::WatchdogTimeout(_), _)) => {
+            warn!("websocket watchdog timeout, clearing pending packets");
+
+            let _ = clear_pending_packets(workers, &chain_id)
+                .map_err(|e| error!("error during clearing pending packets: {}", e));
+        }
+        Err(EventError(EventErrorDetail::DriverChannelClosed(_), _)) => {
+            warn!("websocket driver channel closed, clearing pending packets");
+
+            let _ = clear_pending_packets(workers, &chain_id)
+                .map_err(|e| error!("error during clearing pending packets: {}", e));
+        }
         Err(e) => {
             error!("error when receiving event batch: {}", e)
         }