fix: handle and propagate errors in event channel monitors (#573)

xdustinface · web-flow · commit 86d9a9529ae1 · 2026-04-01T13:23:33.000+11:00
* fix: handle and propagate errors in event channel monitors - Add `on_failure` cancellation token to `spawn_broadcast_monitor` and `spawn_progress_monitor` to signal the run loop on fatal monitor errors - Treat Lagged broadcast errors as fatal, lost events are unrecoverable at the moment - Distinguish expected vs unexpected channel close by checking the shutdown token - Run loop watches `monitor_failure.cancelled()` and exits with `ChannelFailure` error * fix: guard `Lagged` error during shutdown in event monitor Addresses CodeRabbit review comment on PR #573 #573 (comment) * fix: replace `CancellationToken` with `mpsc` channel for monitor error propagation Use an `mpsc::channel` instead of a `CancellationToken` for `on_failure` so monitors propagate the actual error message to the coordinator. The coordinator is now the single source of `handler.on_error()` calls, eliminating double-reporting. Also cancel `monitor_shutdown` immediately after the loop exits to close the shutdown race window. Addresses CodeRabbit review comment on PR #573 #573 (comment) #573 (comment)
diff --git a/dash-spv/src/client/event_handler.rs b/dash-spv/src/client/event_handler.rs
@@ -6,7 +6,7 @@
 
 use std::sync::Arc;
 
-use tokio::sync::{broadcast, watch};
+use tokio::sync::{broadcast, mpsc, watch};
 use tokio::task::JoinHandle;
 use tokio_util::sync::CancellationToken;
 
@@ -35,11 +35,15 @@ pub trait EventHandler: Send + Sync + 'static {
 impl EventHandler for () {}
 
 /// Spawns a task that monitors a broadcast channel and dispatches events to the handler.
+///
+/// On failure, the error message is sent via `on_failure` so the coordinator can report
+/// it as the single source of error handling.
 pub(crate) fn spawn_broadcast_monitor<E, H, F>(
     name: &'static str,
     mut receiver: broadcast::Receiver<E>,
     handler: Arc<H>,
     shutdown: CancellationToken,
+    on_failure: mpsc::Sender<String>,
     dispatch_fn: F,
 ) -> JoinHandle<()>
 where
@@ -54,8 +58,20 @@ where
                 result = receiver.recv() => {
                     match result {
                         Ok(event) => dispatch_fn(&handler, &event),
-                        Err(broadcast::error::RecvError::Closed) => break,
-                        Err(broadcast::error::RecvError::Lagged(_)) => continue,
+                        Err(broadcast::error::RecvError::Closed) if shutdown.is_cancelled() => break,
+                        Err(broadcast::error::RecvError::Closed) => {
+                            let msg = format!("{} monitor channel closed unexpectedly", name);
+                            tracing::error!("{}", msg);
+                            let _ = on_failure.try_send(msg);
+                            break;
+                        }
+                        Err(broadcast::error::RecvError::Lagged(_)) if shutdown.is_cancelled() => break,
+                        Err(broadcast::error::RecvError::Lagged(n)) => {
+                            let msg = format!("{} monitor lagged, missed {} events", name, n);
+                            tracing::error!("{}", msg);
+                            let _ = on_failure.try_send(msg);
+                            break;
+                        }
                     }
                 }
                 _ = shutdown.cancelled() => break,
@@ -72,6 +88,7 @@ pub(crate) fn spawn_progress_monitor<H: EventHandler>(
     mut receiver: watch::Receiver<SyncProgress>,
     handler: Arc<H>,
     shutdown: CancellationToken,
+    on_failure: mpsc::Sender<String>,
 ) -> JoinHandle<()> {
     tokio::spawn(async move {
         tracing::debug!("Progress monitoring task started");
@@ -83,7 +100,13 @@ pub(crate) fn spawn_progress_monitor<H: EventHandler>(
                 result = receiver.changed() => {
                     match result {
                         Ok(()) => handler.on_progress(&receiver.borrow_and_update()),
-                        Err(_) => break,
+                        Err(_) if shutdown.is_cancelled() => break,
+                        Err(_) => {
+                            let msg = "Progress monitor channel closed unexpectedly".to_string();
+                            tracing::error!("{}", msg);
+                            let _ = on_failure.try_send(msg);
+                            break;
+                        }
                     }
                 }
                 _ = shutdown.cancelled() => break,
@@ -99,7 +122,7 @@ mod tests {
     use std::sync::atomic::{AtomicUsize, Ordering};
     use std::sync::Arc;
 
-    use tokio::sync::{broadcast, watch};
+    use tokio::sync::{broadcast, mpsc, watch};
     use tokio_util::sync::CancellationToken;
 
     use super::{spawn_broadcast_monitor, spawn_progress_monitor, EventHandler};
@@ -166,12 +189,14 @@ mod tests {
         let (tx, rx) = broadcast::channel(16);
         let handler = Arc::new(RecordingHandler::new());
         let shutdown = CancellationToken::new();
+        let (failure_tx, _failure_rx) = mpsc::channel(1);
 
         let task = spawn_broadcast_monitor(
             "test",
             rx,
             handler.clone(),
             shutdown.clone(),
+            failure_tx,
             |h: &RecordingHandler, event: &SyncEvent| h.on_sync_event(event),
         );
 
@@ -202,12 +227,14 @@ mod tests {
         let (_tx, rx) = broadcast::channel::<SyncEvent>(16);
         let handler = Arc::new(RecordingHandler::new());
         let shutdown = CancellationToken::new();
+        let (failure_tx, _failure_rx) = mpsc::channel(1);
 
         let task = spawn_broadcast_monitor(
             "test",
             rx,
             handler.clone(),
             shutdown.clone(),
+            failure_tx,
             |h: &RecordingHandler, event: &SyncEvent| h.on_sync_event(event),
         );
 
@@ -218,28 +245,36 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn broadcast_monitor_exits_on_channel_close() {
+    async fn broadcast_monitor_fails_on_unexpected_channel_close() {
         let (tx, rx) = broadcast::channel::<SyncEvent>(16);
         let handler = Arc::new(RecordingHandler::new());
         let shutdown = CancellationToken::new();
+        let (failure_tx, mut failure_rx) = mpsc::channel(1);
 
         let task = spawn_broadcast_monitor(
             "test",
             rx,
             handler.clone(),
             shutdown.clone(),
+            failure_tx,
             |h: &RecordingHandler, event: &SyncEvent| h.on_sync_event(event),
         );
 
+        // Drop sender without cancelling shutdown — this is unexpected
         drop(tx);
         task.await.unwrap();
+
+        let msg = failure_rx.try_recv().expect("should have received failure message");
+        assert!(msg.contains("closed unexpectedly"));
+        assert_eq!(handler.error_count.load(Ordering::SeqCst), 0);
     }
 
     #[tokio::test]
-    async fn broadcast_monitor_handles_lagged_receiver() {
+    async fn broadcast_monitor_exits_on_lagged_receiver() {
         let (tx, rx) = broadcast::channel(2);
         let handler = Arc::new(RecordingHandler::new());
         let shutdown = CancellationToken::new();
+        let (failure_tx, mut failure_rx) = mpsc::channel(1);
 
         // Send more messages than the buffer can hold before spawning the monitor
         tx.send(SyncEvent::BlockHeadersStored {
@@ -260,31 +295,28 @@ mod tests {
             rx,
             handler.clone(),
             shutdown.clone(),
+            failure_tx,
             |h: &RecordingHandler, event: &SyncEvent| h.on_sync_event(event),
         );
 
-        // Send one more after the monitor starts
-        tx.send(SyncEvent::BlockHeadersStored {
-            tip_height: 4,
-        })
-        .unwrap();
-        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
-
-        shutdown.cancel();
+        // The monitor should exit on its own due to the lagged error
         task.await.unwrap();
 
-        // The monitor should have received at least the last message (and possibly
-        // one from the lagged recovery). The key thing is it doesn't crash.
-        assert!(handler.sync_count.load(Ordering::SeqCst) >= 1);
+        // No sync events should have been dispatched, failure sent via channel
+        assert_eq!(handler.sync_count.load(Ordering::SeqCst), 0);
+        assert_eq!(handler.error_count.load(Ordering::SeqCst), 0);
+        let msg = failure_rx.try_recv().expect("should have received failure message");
+        assert!(msg.contains("lagged"));
     }
 
     #[tokio::test]
     async fn progress_monitor_sends_initial_and_updates() {
         let (tx, rx) = watch::channel(SyncProgress::default());
         let handler = Arc::new(RecordingHandler::new());
         let shutdown = CancellationToken::new();
+        let (failure_tx, _failure_rx) = mpsc::channel(1);
 
-        let task = spawn_progress_monitor(rx, handler.clone(), shutdown.clone());
+        let task = spawn_progress_monitor(rx, handler.clone(), shutdown.clone(), failure_tx);
 
         // Give the task time to send initial progress
         tokio::time::sleep(std::time::Duration::from_millis(50)).await;
@@ -302,34 +334,41 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn progress_monitor_exits_on_sender_drop() {
+    async fn progress_monitor_fails_on_unexpected_sender_drop() {
         let (tx, rx) = watch::channel(SyncProgress::default());
         let handler = Arc::new(RecordingHandler::new());
         let shutdown = CancellationToken::new();
+        let (failure_tx, mut failure_rx) = mpsc::channel(1);
 
-        let task = spawn_progress_monitor(rx, handler.clone(), shutdown.clone());
+        let task = spawn_progress_monitor(rx, handler.clone(), shutdown.clone(), failure_tx);
 
         // Give it time to send initial
         tokio::time::sleep(std::time::Duration::from_millis(50)).await;
 
+        // Drop sender without cancelling shutdown — this is unexpected
         drop(tx);
         task.await.unwrap();
 
-        // At least the initial progress was sent
+        // At least the initial progress was sent, failure sent via channel
         assert!(handler.progress_count.load(Ordering::SeqCst) >= 1);
+        assert_eq!(handler.error_count.load(Ordering::SeqCst), 0);
+        let msg = failure_rx.try_recv().expect("should have received failure message");
+        assert!(msg.contains("closed unexpectedly"));
     }
 
     #[tokio::test]
     async fn network_event_dispatch() {
         let (tx, rx) = broadcast::channel(16);
         let handler = Arc::new(RecordingHandler::new());
         let shutdown = CancellationToken::new();
+        let (failure_tx, _failure_rx) = mpsc::channel(1);
 
         let task = spawn_broadcast_monitor(
             "network",
             rx,
             handler.clone(),
             shutdown.clone(),
+            failure_tx,
             |h: &RecordingHandler, event: &NetworkEvent| h.on_network_event(event),
         );
 
diff --git a/dash-spv/src/client/sync_coordinator.rs b/dash-spv/src/client/sync_coordinator.rs
@@ -2,6 +2,7 @@
 
 use std::time::Duration;
 
+use tokio::sync::mpsc;
 use tokio_util::sync::CancellationToken;
 
 use super::event_handler::{spawn_broadcast_monitor, spawn_progress_monitor};
@@ -39,6 +40,7 @@ impl<W: WalletInterface, N: NetworkManager, S: StorageManager, H: EventHandler>
         tracing::info!("Starting continuous network monitoring...");
 
         let monitor_shutdown = CancellationToken::new();
+        let (monitor_failure_tx, mut monitor_failure_rx) = mpsc::channel::<String>(1);
 
         // Subscribe to channels
         let sync_event_rx = self.subscribe_sync_events().await;
@@ -52,6 +54,7 @@ impl<W: WalletInterface, N: NetworkManager, S: StorageManager, H: EventHandler>
             sync_event_rx,
             handler.clone(),
             monitor_shutdown.clone(),
+            monitor_failure_tx.clone(),
             |h, event| h.on_sync_event(event),
         );
 
@@ -60,6 +63,7 @@ impl<W: WalletInterface, N: NetworkManager, S: StorageManager, H: EventHandler>
             network_event_rx,
             handler.clone(),
             monitor_shutdown.clone(),
+            monitor_failure_tx.clone(),
             |h, event| h.on_network_event(event),
         );
 
@@ -68,11 +72,16 @@ impl<W: WalletInterface, N: NetworkManager, S: StorageManager, H: EventHandler>
             wallet_event_rx,
             handler.clone(),
             monitor_shutdown.clone(),
+            monitor_failure_tx.clone(),
             |h, event| h.on_wallet_event(event),
         );
 
-        let progress_task =
-            spawn_progress_monitor(progress_rx, handler.clone(), monitor_shutdown.clone());
+        let progress_task = spawn_progress_monitor(
+            progress_rx,
+            handler.clone(),
+            monitor_shutdown.clone(),
+            monitor_failure_tx,
+        );
 
         // Run the sync loop
         let mut sync_coordinator_tick_interval = tokio::time::interval(SYNC_COORDINATOR_TICK_MS);
@@ -93,18 +102,27 @@ impl<W: WalletInterface, N: NetworkManager, S: StorageManager, H: EventHandler>
                     tracing::debug!("DashSpvClient run loop cancelled");
                     break None
                 }
+                Some(msg) = monitor_failure_rx.recv() => {
+                    break Some(crate::SpvError::ChannelFailure(
+                        "event monitor".into(),
+                        msg,
+                    ))
+                }
             };
 
-            if let Some(ref e) = error {
-                handler.on_error(&e.to_string());
+            if error.is_some() {
                 break error;
             }
         };
 
-        // Cancel monitoring tasks and wait for them
+        // Signal monitors to shut down before channels close
         monitor_shutdown.cancel();
         let _ = tokio::join!(sync_task, network_task, wallet_task, progress_task);
 
+        if let Some(ref e) = error {
+            handler.on_error(&e.to_string());
+        }
+
         let stop_result = self.stop().await;
 
         match error {
