diff --git a/applications/tari_indexer/src/network_state_sync/committee_client.rs b/applications/tari_indexer/src/network_state_sync/committee_client.rs index 85a8162498..a71a2ab793 100644 --- a/applications/tari_indexer/src/network_state_sync/committee_client.rs +++ b/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), diff --git a/applications/tari_validator_node/src/bootstrap.rs b/applications/tari_validator_node/src/bootstrap.rs index cf14fead58..fb9fbab3dc 100644 --- a/applications/tari_validator_node/src/bootstrap.rs +++ b/applications/tari_validator_node/src/bootstrap.rs @@ -421,32 +421,15 @@ impl Services { 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, -// pub mempool: MempoolHandle, -// pub epoch_manager: EpochManagerHandle, -// pub template_manager: TemplateManagerHandle, -// pub consensus_handle: ConsensusHandle, -// // pub global_db: GlobalDb>, -// pub dry_run_transaction_processor: DryRunTransactionProcessor, -// // pub validator_node_client_factory: TariValidatorNodeRpcClientFactory, -// // pub consensus_gossip_service: ConsensusGossipHandle, -// pub state_store: SqliteStateStore, -// pub global_db: GlobalDb>, -// -// pub handles: Vec>>, -// } -// -// 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( config: &ApplicationConfig, diff --git a/applications/tari_validator_node/src/consensus/mod.rs b/applications/tari_validator_node/src/consensus/mod.rs index bd0337e8ff..da890e6ce7 100644 --- a/applications/tari_validator_node/src/consensus/mod.rs +++ b/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, ); diff --git a/applications/tari_validator_node/src/event_subscription.rs b/applications/tari_validator_node/src/event_subscription.rs index 3aa61c508d..4663d61663 100644 --- a/applications/tari_validator_node/src/event_subscription.rs +++ b/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(broadcast::Sender); +pub struct EventSubscription(broadcast::WeakSender); impl EventSubscription { - pub fn new(sender: broadcast::Sender) -> Self { + pub fn new(sender: broadcast::WeakSender) -> Self { Self(sender) } pub fn subscribe(&self) -> broadcast::Receiver { - self.0.subscribe() + let sender = self.0.upgrade().unwrap_or_else(|| broadcast::Sender::new(1)); + sender.subscribe() } } diff --git a/applications/tari_validator_node/src/node.rs b/applications/tari_validator_node/src/node.rs index bc19bc63e1..55644fd00f 100644 --- a/applications/tari_validator_node/src/node.rs +++ b/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"); } } diff --git a/applications/tari_validator_node/src/p2p/services/mempool/service.rs b/applications/tari_validator_node/src/p2p/services/mempool/service.rs index f573a014ae..2d60616742 100644 --- a/applications/tari_validator_node/src/p2p/services/mempool/service.rs +++ b/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,24 +96,51 @@ 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; } } @@ -121,6 +148,7 @@ where self.gossip.unsubscribe().await?; + info!(target: LOG_TARGET, "💤 Mempool service shutting down"); Ok(()) } diff --git a/crates/consensus/src/hotstuff/on_message_validate.rs b/crates/consensus/src/hotstuff/on_message_validate.rs index 737a89647f..6c04bd190d 100644 --- a/crates/consensus/src/hotstuff/on_message_validate.rs +++ b/crates/consensus/src/hotstuff/on_message_validate.rs @@ -45,7 +45,7 @@ pub struct OnMessageValidate { leader_strategy: TConsensusSpec::LeaderStrategy, vote_signing_service: TConsensusSpec::SignerService, outbound_messaging: TConsensusSpec::OutboundMessaging, - tx_events: broadcast::Sender, + tx_events: broadcast::WeakSender, /// Keep track of max 32 in-flight requests active_missing_transaction_requests: SimpleFixedArray, current_request_id: u32, @@ -60,7 +60,7 @@ impl OnMessageValidate { leader_strategy: TConsensusSpec::LeaderStrategy, vote_signing_service: TConsensusSpec::SignerService, outbound_messaging: TConsensusSpec::OutboundMessaging, - tx_events: broadcast::Sender, + tx_events: broadcast::WeakSender, ) -> Self { Self { config, @@ -204,7 +204,7 @@ impl OnMessageValidate { { 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 OnMessageValidate { }); } - 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 OnMessageValidate { Ok(missing) } + + fn publish_event(&self, event: HotstuffEvent) { + if let Some(sender) = self.tx_events.upgrade() { + let _ignore = sender.send(event); + } + } } #[derive(Debug)] diff --git a/crates/consensus/src/hotstuff/on_ready_to_vote_on_local_block.rs b/crates/consensus/src/hotstuff/on_ready_to_vote_on_local_block.rs index 4413b0f67d..93fe4d3c71 100644 --- a/crates/consensus/src/hotstuff/on_ready_to_vote_on_local_block.rs +++ b/crates/consensus/src/hotstuff/on_ready_to_vote_on_local_block.rs @@ -71,7 +71,7 @@ pub struct OnReadyToVoteOnLocalBlock { local_validator_pk: RistrettoPublicKey, config: HotstuffConfig, transaction_pool: TransactionPool, - tx_events: broadcast::Sender, + tx_events: broadcast::WeakSender, transaction_manager: ConsensusTransactionManager, } @@ -82,7 +82,7 @@ where TConsensusSpec: ConsensusSpec local_validator_pk: RistrettoPublicKey, config: HotstuffConfig, transaction_pool: TransactionPool, - tx_events: broadcast::Sender, + tx_events: broadcast::WeakSender, 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( diff --git a/crates/consensus/src/hotstuff/on_receive_local_proposal.rs b/crates/consensus/src/hotstuff/on_receive_local_proposal.rs index 078dcf0802..7f35bf3c79 100644 --- a/crates/consensus/src/hotstuff/on_receive_local_proposal.rs +++ b/crates/consensus/src/hotstuff/on_receive_local_proposal.rs @@ -75,7 +75,7 @@ pub struct OnReceiveLocalProposalHandler { outbound_messaging: TConsensusSpec::OutboundMessaging, signing_service: TConsensusSpec::SignerService, on_receive_foreign_proposal: OnReceiveForeignProposalHandler, - tx_events: broadcast::Sender, + tx_events: broadcast::WeakSender, hooks: TConsensusSpec::Hooks, } @@ -89,7 +89,7 @@ impl OnReceiveLocalProposalHandler, - tx_events: broadcast::Sender, + tx_events: broadcast::WeakSender, transaction_manager: ConsensusTransactionManager< TConsensusSpec::TransactionExecutor, TConsensusSpec::StateStore, @@ -540,7 +540,9 @@ impl OnReceiveLocalProposalHandler { 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() => { }, } } diff --git a/crates/consensus/src/hotstuff/state_machine/worker.rs b/crates/consensus/src/hotstuff/state_machine/worker.rs index a33d1c2c20..f25856ffba 100644 --- a/crates/consensus/src/hotstuff/state_machine/worker.rs +++ b/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> { let mut shutdown_signal = self.shutdown_signal.clone(); let result = tokio::select! { + biased; _ = shutdown_signal.wait() => Ok(ConsensusStateEvent::Shutdown), ret = fut => ret, }; diff --git a/crates/consensus/src/hotstuff/worker.rs b/crates/consensus/src/hotstuff/worker.rs index 37993ec591..0f7afb31d1 100644 --- a/crates/consensus/src/hotstuff/worker.rs +++ b/crates/consensus/src/hotstuff/worker.rs @@ -144,7 +144,6 @@ impl HotstuffWorker { local_validator_addr: local_validator_addr.clone(), config: config.clone(), - tx_events: tx_events.clone(), rx_new_transactions, rx_missing_transactions, @@ -157,7 +156,7 @@ impl HotstuffWorker { leader_strategy.clone(), signing_service.clone(), outbound_messaging.clone(), - tx_events.clone(), + tx_events.downgrade(), ), on_next_sync_view: OnNextSyncViewHandler::new( @@ -174,7 +173,7 @@ impl HotstuffWorker { outbound_messaging.clone(), signing_service.clone(), transaction_pool.clone(), - tx_events, + tx_events.downgrade(), transaction_manager.clone(), config.clone(), hooks.clone(), @@ -230,6 +229,7 @@ impl HotstuffWorker { pacemaker: pacemaker.clone_handle(), pacemaker_worker: Some(pacemaker), hooks, + tx_events, shutdown, } } @@ -623,6 +623,10 @@ impl HotstuffWorker { self.on_inbound_message.clear_buffer(); } + pub fn shutdown_signal(&self) -> &ShutdownSignal { + &self.shutdown + } + /// Read and discard messages. This should be used only when consensus is inactive. pub async fn discard_messages(&mut self) { loop { diff --git a/crates/epoch_manager/src/service/epoch_manager_service.rs b/crates/epoch_manager/src/service/epoch_manager_service.rs index 6d1dfbd74a..f831b730fb 100644 --- a/crates/epoch_manager/src/service/epoch_manager_service.rs +++ b/crates/epoch_manager/src/service/epoch_manager_service.rs @@ -84,7 +84,7 @@ impl EpochManagerService { let (tx_request, rx_request) = mpsc::channel(10); let (events, _) = broadcast::channel(100); let current_epoch = Arc::new(AtomicU64::new(0)); - let epoch_manager_handle = EpochManagerHandle::new(tx_request, events.clone(), current_epoch.clone()); + let epoch_manager_handle = EpochManagerHandle::new(tx_request, events.downgrade(), current_epoch.clone()); let task_handle = tokio::spawn(async move { Self { @@ -149,6 +149,7 @@ impl EpochManagerService { } } } + Ok(()) } diff --git a/crates/epoch_manager/src/service/handle.rs b/crates/epoch_manager/src/service/handle.rs index 9be5262f47..f8797a7daa 100644 --- a/crates/epoch_manager/src/service/handle.rs +++ b/crates/epoch_manager/src/service/handle.rs @@ -31,13 +31,13 @@ use crate::{ pub struct EpochManagerHandle { tx_request: mpsc::Sender>, current_epoch: Arc, - events: broadcast::Sender, + events: broadcast::WeakSender, } impl EpochManagerHandle { pub fn new( tx_request: mpsc::Sender>, - events: broadcast::Sender, + events: broadcast::WeakSender, current_epoch: Arc, ) -> Self { Self { @@ -129,7 +129,12 @@ impl EpochManagerReader for EpochManagerHandle { type Addr = TAddr; fn subscribe(&self) -> broadcast::Receiver { - self.events.subscribe() + let sender = self.events.upgrade().unwrap_or_else(|| { + // Should the sender be closed (upgrade() returns None), create a "dummy" channel that will + // immediately close. This is more in-line with what you would expect from the api. + broadcast::Sender::new(1) + }); + sender.subscribe() } async fn wait_for_initial_scanning_to_complete(&self) -> Result<(), EpochManagerError> { diff --git a/crates/epoch_oracles/src/configured/real_time_ticker.rs b/crates/epoch_oracles/src/configured/real_time_ticker.rs index dc875b0c62..bdf526d683 100644 --- a/crates/epoch_oracles/src/configured/real_time_ticker.rs +++ b/crates/epoch_oracles/src/configured/real_time_ticker.rs @@ -97,7 +97,7 @@ impl EpochTicker for RealTimeEpochTicker { return Poll::Ready(Some(EpochTickerData { epoch, epoch_hash, - // Catching up + // Catching up if calculated_epoch > self.epoch done_for_now: calculated_epoch == epoch, })); } @@ -171,8 +171,10 @@ mod tests { async fn check_config_file() { let mut file = std::fs::File::open("../../data/ec.json").unwrap(); let config = serde_json::from_reader::<_, Config>(&mut file).unwrap(); - let mut ticker = RealTimeEpochTicker::new(config.initial_epoch, config.base_time, Epoch(1965)) - .with_epoch_time_secs((config.epoch_time.unwrap().as_secs() / 4).try_into().unwrap()); + let mut ticker = RealTimeEpochTicker::new(config.initial_epoch, config.base_time, Epoch(128)) + .with_epoch_time_secs(config.epoch_time.unwrap().as_secs().try_into().unwrap()); + let current = ticker.calc_current_epoch(); + eprintln!("current: {:?}", current); loop { let res = poll_fn(|cx| ticker.poll_tick(cx)).await; diff --git a/crates/template_manager/src/implementation/downloader.rs b/crates/template_manager/src/implementation/downloader.rs index ea6f0cefef..464914f6bd 100644 --- a/crates/template_manager/src/implementation/downloader.rs +++ b/crates/template_manager/src/implementation/downloader.rs @@ -25,12 +25,15 @@ use bytes::Bytes; use futures::{future::BoxFuture, stream::FuturesUnordered}; +use log::info; use tari_common_types::types::FixedHash; use tari_ootle_storage::global::DbTemplateType; use tari_template_lib::types::TemplateAddress; use tokio::{sync::mpsc, task}; use tokio_stream::StreamExt; +const LOG_TARGET: &str = "tari::ootle::template_manager::downloader"; + pub struct DownloadRequest { pub address: TemplateAddress, pub template_type: DbTemplateType, @@ -68,7 +71,10 @@ impl TemplateDownloadWorker { Some(req) => { self.pending_downloads.push(Box::pin(download(req))); }, - None => break, + None => { + info!(target: LOG_TARGET, "💤 Download queue closed, shutting down downloader"); + break + }, } }, Some(result) = self.pending_downloads.next() => { diff --git a/crates/template_manager/src/implementation/template_sync_task.rs b/crates/template_manager/src/implementation/template_sync_task.rs index e44d23e737..28d9974d66 100644 --- a/crates/template_manager/src/implementation/template_sync_task.rs +++ b/crates/template_manager/src/implementation/template_sync_task.rs @@ -192,6 +192,10 @@ where .await .optional()? else { + // No more committee members to try - no real choice but to clear the failed list and try again if this + // is re-attempted + self.recently_failed_clients.clear(); + self.recently_failed_clients.shrink_to(100); return Err(TemplateSyncError::NoMoreSyncValidators); }; diff --git a/networking/core/src/builder.rs b/networking/core/src/builder.rs index a5db6b99a7..9995605afb 100644 --- a/networking/core/src/builder.rs +++ b/networking/core/src/builder.rs @@ -144,6 +144,6 @@ where worker = worker.with_metrics(registry_mut); } let handle = tokio::spawn(worker.run()); - Ok((NetworkingHandle::new(local_peer_id, tx, tx_events), handle)) + Ok((NetworkingHandle::new(local_peer_id, tx, tx_events.downgrade()), handle)) } } diff --git a/networking/core/src/handle.rs b/networking/core/src/handle.rs index c196ed96fb..d5376139a2 100644 --- a/networking/core/src/handle.rs +++ b/networking/core/src/handle.rs @@ -168,14 +168,14 @@ impl FromIterator for MulticastDestination { pub struct NetworkingHandle { tx_request: mpsc::Sender>, local_peer_id: PeerId, - tx_events: broadcast::Sender, + tx_events: broadcast::WeakSender, } impl NetworkingHandle { pub(super) fn new( local_peer_id: PeerId, tx_request: mpsc::Sender>, - tx_events: broadcast::Sender, + tx_events: broadcast::WeakSender, ) -> Self { Self { tx_request, @@ -185,7 +185,10 @@ impl NetworkingHandle { } pub fn subscribe_events(&self) -> broadcast::Receiver { - self.tx_events.subscribe() + self.tx_events + .upgrade() + .unwrap_or_else(|| broadcast::Sender::new(1)) + .subscribe() } pub async fn is_subscribed_to_topic>(&self, topic: T) -> Result { diff --git a/networking/core/src/worker.rs b/networking/core/src/worker.rs index 638601c499..2d1cc475d3 100644 --- a/networking/core/src/worker.rs +++ b/networking/core/src/worker.rs @@ -197,6 +197,7 @@ where }, _ = self.shutdown_signal.wait() => { + info!(target: LOG_TARGET, "💤 Networking service shutting down"); break; } } @@ -1033,6 +1034,9 @@ where error, } => { debug!(target: LOG_TARGET, "Inbound substream failed from peer {peer_id} with stream id {stream_id}: {error}"); + if let Some(waiting_reply) = self.pending_substream_requests.remove(&stream_id) { + let _ignore = waiting_reply.send(Err(NetworkingError::FailedToOpenSubstream(error))); + } }, OutboundFailure { error, @@ -1045,7 +1049,6 @@ where let _ignore = waiting_reply.send(Err(NetworkingError::FailedToOpenSubstream(error))); } }, - Error(_) => {}, } } diff --git a/networking/libp2p-substream/src/behaviour.rs b/networking/libp2p-substream/src/behaviour.rs index 4a52586b01..da527f26b6 100644 --- a/networking/libp2p-substream/src/behaviour.rs +++ b/networking/libp2p-substream/src/behaviour.rs @@ -55,7 +55,7 @@ pub struct Behaviour { /// reachable addresses, if any. connected: HashMap>, pending_outbound_streams: HashMap>, - next_outbound_stream_id: StreamId, + next_stream_id: StreamId, } impl Behaviour { @@ -65,7 +65,7 @@ impl Behaviour { pending_events: VecDeque::new(), pending_outbound_streams: HashMap::new(), connected: HashMap::new(), - next_outbound_stream_id: StreamId::default(), + next_stream_id: StreamId::default(), } } @@ -88,7 +88,7 @@ impl Behaviour { } pub fn open_substream(&mut self, peer_id: PeerId, protocol: StreamProtocol) -> StreamId { - let stream_id = self.next_outbound_stream_id(); + let stream_id = self.next_stream_id(); let request = OpenStreamRequest::new(stream_id, peer_id, protocol); match self.get_connections(&peer_id) { @@ -113,9 +113,9 @@ impl Behaviour { stream_id } - fn next_outbound_stream_id(&mut self) -> StreamId { - let request_id = self.next_outbound_stream_id; - self.next_outbound_stream_id = self.next_outbound_stream_id.wrapping_add(1); + fn next_stream_id(&mut self) -> StreamId { + let request_id = self.next_stream_id; + self.next_stream_id = self.next_stream_id.wrapping_add(1); request_id } @@ -285,7 +285,6 @@ impl NetworkBehaviour for Behaviour { Event::InboundSubstreamOpen { .. } => {}, Event::InboundFailure { .. } => {}, Event::OutboundFailure { .. } => {}, - Event::Error(_) => {}, } self.pending_events.push_back(ToSwarm::GenerateEvent(event)); diff --git a/networking/libp2p-substream/src/event.rs b/networking/libp2p-substream/src/event.rs index 4e7b999372..24ab56aa2f 100644 --- a/networking/libp2p-substream/src/event.rs +++ b/networking/libp2p-substream/src/event.rs @@ -27,5 +27,4 @@ pub enum Event { stream_id: StreamId, error: Error, }, - Error(Error), } diff --git a/networking/libp2p-substream/src/metrics.rs b/networking/libp2p-substream/src/metrics.rs index 62f6d84cc2..23c97e05d1 100644 --- a/networking/libp2p-substream/src/metrics.rs +++ b/networking/libp2p-substream/src/metrics.rs @@ -97,6 +97,7 @@ impl libp2p::metrics::Recorder for Metrics { error, } => { self.inbound_failure_count.inc(); + self.error_count.inc(); tracing::error!(error = ?error, "Inbound substream failure for peer {peer_id} (ID = {stream_id})"); }, Event::OutboundFailure { @@ -106,11 +107,8 @@ impl libp2p::metrics::Recorder for Metrics { error, } => { self.outbound_failure_count.inc(); - tracing::error!(error = ?error, "Outbound substream failure ({stream_id}) peer {peer_id} on protocol {protocol}"); - }, - Event::Error(error) => { self.error_count.inc(); - tracing::error!(error = ?error, "Error in substream event"); + tracing::error!(error = ?error, "Outbound substream failure ({stream_id}) peer {peer_id} on protocol {protocol}"); }, } }