Break task loops when channel closes

crates/hotshot/src/tasks/mod.rs

@@ -10,7 +10,7 @@
 pub mod task_state;
 use std::{fmt::Debug, sync::Arc};
 
-use async_broadcast::broadcast;
+use async_broadcast::{broadcast, RecvError};
 use async_compatibility_layer::art::async_spawn;
 use async_lock::RwLock;
 use async_trait::async_trait;
@@ -261,6 +261,9 @@ pub fn create_shutdown_event_monitor<TYPES: NodeType, I: NodeImplementation<TYPE
                         return;
                     }
                 }
+                Err(RecvError::Closed) => {
+                    return;
+                }
                 Err(e) => {
                     tracing::error!("Shutdown event monitor channel recv error: {}", e);
                 }

crates/task-impls/src/consensus2/mod.rs

@@ -6,11 +6,6 @@
 
 use std::sync::Arc;
 
-use self::handlers::{
-    handle_quorum_vote_recv, handle_timeout, handle_timeout_vote_recv, handle_view_change,
-};
-use crate::helpers::broadcast_event;
-use crate::{events::HotShotEvent, vote_collection::VoteCollectorsMap};
 use anyhow::Result;
 use async_broadcast::{Receiver, Sender};
 use async_lock::RwLock;
@@ -34,6 +29,11 @@ use hotshot_types::{
 use tokio::task::JoinHandle;
 use tracing::instrument;
 
+use self::handlers::{
+    handle_quorum_vote_recv, handle_timeout, handle_timeout_vote_recv, handle_view_change,
+};
+use crate::{events::HotShotEvent, helpers::broadcast_event, vote_collection::VoteCollectorsMap};
+
 /// Event handlers for use in the `handle` method.
 mod handlers;
 

crates/task/src/task.rs

@@ -7,7 +7,7 @@
 use std::sync::Arc;
 
 use anyhow::Result;
-use async_broadcast::{Receiver, Sender};
+use async_broadcast::{Receiver, RecvError, Sender};
 #[cfg(async_executor_impl = "async-std")]
 use async_std::task::{spawn, JoinHandle};
 use async_trait::async_trait;
@@ -93,6 +93,9 @@ impl<S: TaskState + Send + 'static> Task<S> {
                                 .await
                                 .inspect_err(|e| tracing::info!("{e}"));
                     }
+                    Err(RecvError::Closed) => {
+                        break self.boxed_state();
+                    }
                     Err(e) => {
                         tracing::error!("Failed to receive from event stream Error: {}", e);
                     }

crates/types/src/vid.rs

@@ -35,7 +35,8 @@ use serde::{Deserialize, Serialize};
 use sha2::Sha256;
 
 use crate::{
-    constants::SRS_DEGREE, data::VidDisperse as HotShotVidDisperse, data::VidDisperseShare,
+    constants::SRS_DEGREE,
+    data::{VidDisperse as HotShotVidDisperse, VidDisperseShare},
     message::Proposal,
 };