fix(vn): prevent possible inf loop on shutdown

applications/tari_indexer/src/network_state_sync/committee_client.rs

@@ -46,11 +46,20 @@ impl ValidatorCommitteeRpcPool {
                 .epoch_manager
                 .get_random_committee_member(epoch, Some(self.shard_group), self.past_failed_nodes.clone())
                 .await
-                .optional()?
-                .ok_or_else(|| ValidatorCommitteeClientError::AllValidatorsFailed {
-                    committee_size: self.past_failed_nodes.len(),
+                .optional()?;
+
+            let Some(member) = member else {
+                // All validators have been attempted and failed - no real choice but to clear the past failed nodes and
+                // try again if this is called again
+                let committee_size = self.past_failed_nodes.len();
+                self.past_failed_nodes.clear();
+                // Clamp max mem usage to 7300 bytes (Multihash size x 100) - this is likely to always be a no-op
+                self.past_failed_nodes.shrink_to(100);
+                return Err(ValidatorCommitteeClientError::AllValidatorsFailed {
+                    committee_size,
                     last_error: last_error.as_ref().map(|e| e.to_string()),
-                })?;
+                });
+            };
             let result = self.session_for_peer(member.address).await;
             match result {
                 Ok(session) => return Ok(session),

applications/tari_validator_node/src/bootstrap.rs

@@ -421,32 +421,15 @@ impl<TStore> Services<TStore> {
         let (res, _, _) = future::select_all(fused).await;
         res.unwrap_or_else(|e| Err(anyhow!("Task panicked: {}", e)))
     }
+
+    pub async fn join_all(self) -> Result<(), anyhow::Error> {
+        let results = future::try_join_all(self.handles).await?;
+        for res in results {
+            res?;
+        }
+        Ok(())
+    }
 }
-// pub struct Services {
-//     pub keypair: RistrettoKeypair,
-//     pub networking: NetworkingHandle<TariMessagingSpec>,
-//     pub mempool: MempoolHandle,
-//     pub epoch_manager: EpochManagerHandle<PeerAddress>,
-//     pub template_manager: TemplateManagerHandle,
-//     pub consensus_handle: ConsensusHandle,
-//     // pub global_db: GlobalDb<SqliteGlobalDbAdapter<PeerAddress>>,
-//     pub dry_run_transaction_processor: DryRunTransactionProcessor,
-//     // pub validator_node_client_factory: TariValidatorNodeRpcClientFactory,
-//     // pub consensus_gossip_service: ConsensusGossipHandle,
-//     pub state_store: SqliteStateStore<PeerAddress>,
-//     pub global_db: GlobalDb<SqliteGlobalDbAdapter<PeerAddress>>,
-//
-//     pub handles: Vec<JoinHandle<Result<(), anyhow::Error>>>,
-// }
-//
-// impl Services {
-//     pub async fn on_any_exit(&mut self) -> Result<(), anyhow::Error> {
-//         // JoinHandler panics if polled again after reading the Result, we fuse the future to prevent this.
-//         let fused = self.handles.iter_mut().map(|h| h.fuse());
-//         let (res, _, _) = future::select_all(fused).await;
-//         res.unwrap_or_else(|e| Err(anyhow!("Task panicked: {}", e)))
-//     }
-// }
 
 async fn spawn_p2p_rpc<TStateStore: StateStore + Clone + Send + Sync + 'static>(
     config: &ApplicationConfig,

applications/tari_validator_node/src/consensus/mod.rs

@@ -114,7 +114,7 @@ pub async fn spawn(
 
     let consensus_handle = ConsensusHandle::new(
         rx_current_state,
-        EventSubscription::new(tx_hotstuff_events),
+        EventSubscription::new(tx_hotstuff_events.downgrade()),
         current_view,
         tx_new_transaction,
     );

applications/tari_validator_node/src/event_subscription.rs

@@ -28,15 +28,16 @@ use tokio::sync::broadcast;
 /// We hold a sender because if we held a receiver then the broadcast buffer would always fill up because the receiver
 /// isn't reading off of it.
 #[derive(Debug)]
-pub struct EventSubscription<T>(broadcast::Sender<T>);
+pub struct EventSubscription<T>(broadcast::WeakSender<T>);
 
 impl<T> EventSubscription<T> {
-    pub fn new(sender: broadcast::Sender<T>) -> Self {
+    pub fn new(sender: broadcast::WeakSender<T>) -> Self {
         Self(sender)
     }
 
     pub fn subscribe(&self) -> broadcast::Receiver<T> {
-        self.0.subscribe()
+        let sender = self.0.upgrade().unwrap_or_else(|| broadcast::Sender::new(1));
+        sender.subscribe()
     }
 }
 

applications/tari_validator_node/src/node.rs

@@ -20,6 +20,8 @@
 //  WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE
 //  USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 
+use std::process;
+
 use log::*;
 use tari_consensus::hotstuff::HotstuffEvent;
 use tari_epoch_manager::{EpochManagerEvent, EpochManagerReader};
@@ -50,9 +52,6 @@ impl ValidatorNode {
         //     error!(target: LOG_TARGET, "Failed to dial local shard peers: {}", err);
         // }
 
-        // let sigint = tokio::signal::ctrl_c();
-        // let mut sigterm = signal(SignalKind::terminate())?;
-
         loop {
             let metrics = tokio::runtime::Handle::current().metrics();
             info!(
@@ -66,12 +65,8 @@ impl ValidatorNode {
             tokio::select! {
                 _ = tokio::signal::ctrl_c() => {
                     info!(target: LOG_TARGET, "💤 Received SIGINT");
-                    // Second SIGINT forces shutdown
-                    if shutdown.is_triggered() {
-                        warn!(target: LOG_TARGET, "💤 Shutdown NOW");
-                        break;
-                    }
                     shutdown.trigger();
+                    break;
                 },
 
                 Ok(event) = hotstuff_events.recv() => if let Err(err) = self.handle_hotstuff_event(event).await {
@@ -85,11 +80,10 @@ impl ValidatorNode {
                 result = self.services.on_any_exit() => {
                     match result {
                         Ok(_) => {
-                            if shutdown.is_triggered() {
-                                info!(target: LOG_TARGET, "🏁 All services have exited cleanly");
-                            } else {
+                            if !shutdown.is_triggered() {
                                 warn!(target: LOG_TARGET, "❓️ A service has exited unexpectedly. Shutting down...");
                             }
+                            shutdown.trigger();
                             break;
                         },
                         Err(err) => {
@@ -98,7 +92,20 @@ impl ValidatorNode {
                         }
                     }
                 }
+            }
+        }
 
+        info!(target: LOG_TARGET, "💤 Waiting for all services to shut down... ctrl+c to force shutdown");
+
+        tokio::select! {
+            _ = tokio::signal::ctrl_c() => {
+                // Second SIGINT forces shutdown
+                warn!(target: LOG_TARGET, "💤 Shutdown NOW");
+                process::exit(1);
+            },
+            res = self.services.join_all() => {
+                res?;
+                info!(target: LOG_TARGET, "🏁 All services have exited cleanly");
             }
         }
 

applications/tari_validator_node/src/p2p/services/mempool/service.rs

@@ -31,7 +31,7 @@ use tari_ootle_common_types::{optional::Optional, PeerAddress, ShardGroup};
 use tari_ootle_p2p::{NewTransactionMessage, TariMessage, TariMessagingSpec};
 use tari_ootle_storage::{consensus_models::TransactionRecord, StateStore, StateStoreReadTransaction};
 use tari_transaction::{Transaction, TransactionId};
-use tokio::sync::{mpsc, oneshot};
+use tokio::sync::{broadcast, mpsc, oneshot};
 
 #[cfg(feature = "metrics")]
 use super::metrics::PrometheusMempoolMetrics;
@@ -96,31 +96,59 @@ where
 
         loop {
             tokio::select! {
-                Some(req) = self.mempool_requests.recv() => self.handle_request(req).await,
-                Some(result) = self.gossip.next_message() => {
-                    if let Err(e) = self.handle_new_transaction_from_remote(result).await {
-                        warn!(target: LOG_TARGET, "Mempool rejected transaction: {}", e);
+                req = self.mempool_requests.recv() => {
+                    match req {
+                        Some(req) => self.handle_request(req).await,
+                        None => {
+                            info!(target: LOG_TARGET, "Mempool request channel closed, shutting down");
+                            break;
+                        }
                     }
+                },
+                result = self.gossip.next_message() => {
+                    match result {
+                        Some(msg) => {
+                            if let Err(e) = self.handle_new_transaction_from_remote(msg).await {
+                                warn!(target: LOG_TARGET, "Mempool rejected transaction: {}", e);
+                            }
+                        }
+                        None => {
+                            info!(target: LOG_TARGET, "Gossip channel closed, shutting down mempool service");
+                            break;
+                        }
+                    };
                 }
-                Ok(HotstuffEvent::EpochChanged { epoch, registered_shard_group}) = consensus_events.recv() => {
-                    if let Some(shard_group) = registered_shard_group {
-                        info!(target: LOG_TARGET, "Mempool service subscribing transaction messages for {shard_group} in {epoch}");
-                        self.gossip.subscribe(shard_group).await?;
-                    } else {
-                        info!(target: LOG_TARGET, "Not registered for epoch {epoch}, unsubscribing from gossip if necessary");
-                        self.gossip.unsubscribe().await?;
+                event = consensus_events.recv() => {
+                    match event {
+                        Ok(HotstuffEvent::EpochChanged { epoch, registered_shard_group})  => {
+                            if let Some(shard_group) = registered_shard_group {
+                                info!(target: LOG_TARGET, "Mempool service subscribing transaction messages for {shard_group} in {epoch}");
+                                self.gossip.subscribe(shard_group).await?;
+                            } else {
+                                info!(target: LOG_TARGET, "Not registered for epoch {epoch}, unsubscribing from gossip if necessary");
+                                self.gossip.unsubscribe().await?;
+                            }
+                        },
+                        Ok(_) => {},
+                        Err(broadcast::error::RecvError::Lagged(n)) => {
+                            warn!(target: LOG_TARGET, "Missed {} consensus events", n);
+                        }
+                        Err(broadcast::error::RecvError::Closed) => {
+                            info!(target: LOG_TARGET, "Consensus event channel closed, shutting down mempool service");
+                            break;
+                        }
                     }
                 },
 
                 else => {
-                    info!(target: LOG_TARGET, "Mempool service shutting down");
                     break;
                 }
             }
         }
 
         self.gossip.unsubscribe().await?;
 
+        info!(target: LOG_TARGET, "💤 Mempool service shutting down");
         Ok(())
     }
 

crates/consensus/src/hotstuff/on_message_validate.rs

@@ -45,7 +45,7 @@ pub struct OnMessageValidate<TConsensusSpec: ConsensusSpec> {
     leader_strategy: TConsensusSpec::LeaderStrategy,
     vote_signing_service: TConsensusSpec::SignerService,
     outbound_messaging: TConsensusSpec::OutboundMessaging,
-    tx_events: broadcast::Sender<HotstuffEvent>,
+    tx_events: broadcast::WeakSender<HotstuffEvent>,
     /// Keep track of max 32 in-flight requests
     active_missing_transaction_requests: SimpleFixedArray<u32, 32>,
     current_request_id: u32,
@@ -60,7 +60,7 @@ impl<TConsensusSpec: ConsensusSpec> OnMessageValidate<TConsensusSpec> {
         leader_strategy: TConsensusSpec::LeaderStrategy,
         vote_signing_service: TConsensusSpec::SignerService,
         outbound_messaging: TConsensusSpec::OutboundMessaging,
-        tx_events: broadcast::Sender<HotstuffEvent>,
+        tx_events: broadcast::WeakSender<HotstuffEvent>,
     ) -> Self {
         Self {
             config,
@@ -204,7 +204,7 @@ impl<TConsensusSpec: ConsensusSpec> OnMessageValidate<TConsensusSpec> {
                 {
                     info!(target: LOG_TARGET, "♻️ all transactions for local block {unparked_block} are ready for consensus");
 
-                    let _ignore = self.tx_events.send(HotstuffEvent::ParkedBlockReady {
+                    self.publish_event(HotstuffEvent::ParkedBlockReady {
                         block: unparked_block.as_leaf(),
                     });
 
@@ -270,7 +270,7 @@ impl<TConsensusSpec: ConsensusSpec> OnMessageValidate<TConsensusSpec> {
             });
         }
 
-        let _ignore = self.tx_events.send(HotstuffEvent::ProposedBlockParked {
+        self.publish_event(HotstuffEvent::ProposedBlockParked {
             block: proposal.block.as_leaf(),
             num_missing_txs: missing_tx_ids.len(),
             // TODO: remove
@@ -440,6 +440,12 @@ impl<TConsensusSpec: ConsensusSpec> OnMessageValidate<TConsensusSpec> {
 
         Ok(missing)
     }
+
+    fn publish_event(&self, event: HotstuffEvent) {
+        if let Some(sender) = self.tx_events.upgrade() {
+            let _ignore = sender.send(event);
+        }
+    }
 }
 
 #[derive(Debug)]

crates/consensus/src/hotstuff/on_ready_to_vote_on_local_block.rs

@@ -71,7 +71,7 @@ pub struct OnReadyToVoteOnLocalBlock<TConsensusSpec: ConsensusSpec> {
     local_validator_pk: RistrettoPublicKey,
     config: HotstuffConfig,
     transaction_pool: TransactionPool<TConsensusSpec::StateStore>,
-    tx_events: broadcast::Sender<HotstuffEvent>,
+    tx_events: broadcast::WeakSender<HotstuffEvent>,
     transaction_manager: ConsensusTransactionManager<TConsensusSpec::TransactionExecutor, TConsensusSpec::StateStore>,
 }
 
@@ -82,7 +82,7 @@ where TConsensusSpec: ConsensusSpec
         local_validator_pk: RistrettoPublicKey,
         config: HotstuffConfig,
         transaction_pool: TransactionPool<TConsensusSpec::StateStore>,
-        tx_events: broadcast::Sender<HotstuffEvent>,
+        tx_events: broadcast::WeakSender<HotstuffEvent>,
         transaction_manager: ConsensusTransactionManager<
             TConsensusSpec::TransactionExecutor,
             TConsensusSpec::StateStore,
@@ -1533,7 +1533,9 @@ where TConsensusSpec: ConsensusSpec
     }
 
     fn publish_event(&self, event: HotstuffEvent) {
-        let _ignore = self.tx_events.send(event);
+        if let Some(sender) = self.tx_events.upgrade() {
+            let _ignore = sender.send(event);
+        }
     }
 
     fn finalize_block(

crates/consensus/src/hotstuff/on_receive_local_proposal.rs

@@ -75,7 +75,7 @@ pub struct OnReceiveLocalProposalHandler<TConsensusSpec: ConsensusSpec> {
     outbound_messaging: TConsensusSpec::OutboundMessaging,
     signing_service: TConsensusSpec::SignerService,
     on_receive_foreign_proposal: OnReceiveForeignProposalHandler<TConsensusSpec>,
-    tx_events: broadcast::Sender<HotstuffEvent>,
+    tx_events: broadcast::WeakSender<HotstuffEvent>,
     hooks: TConsensusSpec::Hooks,
 }
 
@@ -89,7 +89,7 @@ impl<TConsensusSpec: ConsensusSpec> OnReceiveLocalProposalHandler<TConsensusSpec
         outbound_messaging: TConsensusSpec::OutboundMessaging,
         signing_service: TConsensusSpec::SignerService,
         transaction_pool: TransactionPool<TConsensusSpec::StateStore>,
-        tx_events: broadcast::Sender<HotstuffEvent>,
+        tx_events: broadcast::WeakSender<HotstuffEvent>,
         transaction_manager: ConsensusTransactionManager<
             TConsensusSpec::TransactionExecutor,
             TConsensusSpec::StateStore,
@@ -540,7 +540,9 @@ impl<TConsensusSpec: ConsensusSpec> OnReceiveLocalProposalHandler<TConsensusSpec
     }
 
     fn publish_event(&self, event: HotstuffEvent) {
-        let _ignore = self.tx_events.send(event);
+        if let Some(sender) = self.tx_events.upgrade() {
+            let _ignore = sender.send(event);
+        }
     }
 
     async fn send_vote_to_leader(

crates/consensus/src/hotstuff/state_machine/idle.rs

@@ -58,6 +58,8 @@ where TSpec: ConsensusSpec
 
         loop {
             tokio::select! {
+                biased;
+
                 event = epoch_events.recv() => {
                     match event {
                         Ok(event) => {
@@ -69,12 +71,13 @@ where TSpec: ConsensusSpec
                             debug!(target: LOG_TARGET, "Idle state lagged behind by {n} epoch manager events");
                         },
                         Err(broadcast::error::RecvError::Closed) => {
+                            debug!(target: LOG_TARGET, "Epoch manager event stream closed");
                             break;
                         },
                     }
                 },
                 // Ignore hotstuff messages while idle
-                _ = context.hotstuff.discard_messages() => { }
+                _ = context.hotstuff.discard_messages() => { },
             }
         }
 

crates/consensus/src/hotstuff/state_machine/worker.rs

@@ -134,6 +134,7 @@ where
             state = self.transition(state, next_event);
             let _ignore = context.tx_current_state.send((&state).into());
             if state.is_shutdown() {
+                info!(target: LOG_TARGET, "💤 Consensus state machine shutting down");
                 break;
             }
         }
@@ -143,6 +144,7 @@ where
     where Fut: Future<Output = Result<ConsensusStateEvent, HotStuffError>> {
         let mut shutdown_signal = self.shutdown_signal.clone();
         let result = tokio::select! {
+            biased;
             _ = shutdown_signal.wait() => Ok(ConsensusStateEvent::Shutdown),
             ret = fut => ret,
         };