diff --git a/node/src/endpoint_gossip/handler.rs b/node/src/endpoint_gossip/handler.rs index 2491298ca85b374fd04247add61db155bf1dfe9e..ee4e3fe82c569aeeb9664b501c008308bf59d74c 100644 --- a/node/src/endpoint_gossip/handler.rs +++ b/node/src/endpoint_gossip/handler.rs @@ -10,7 +10,7 @@ use sc_network::{ utils::interval, NetworkEventStream, NetworkPeers, NetworkStateInfo, NotificationService, ObservedRole, PeerId, }; -use sc_utils::mpsc::{tracing_unbounded, TracingUnboundedReceiver, TracingUnboundedSender}; +use sc_utils::mpsc::{TracingUnboundedReceiver, TracingUnboundedSender}; use sp_api::__private::BlockT; use std::{collections::HashMap, marker::PhantomData, pin::Pin}; @@ -34,10 +34,7 @@ pub fn build< .fuse(), network, peers: HashMap::new(), - command_rx: command_rx.unwrap_or_else(|| { - let (_tx, rx) = tracing_unbounded("mpsc_duniter_peering_rpc_command", 1_000); - rx - }), + command_rx, self_peering: Peering { endpoints }, events_reporter: DuniterEventsReporter { sink: rpc_sink, @@ -98,7 +95,7 @@ pub struct GossipsHandler< /// Internal sink to report events. events_reporter: DuniterEventsReporter, /// Receiver for external commands (tests/RPC methods). - command_rx: TracingUnboundedReceiver<DuniterPeeringCommand>, + command_rx: Option<TracingUnboundedReceiver<DuniterPeeringCommand>>, /// Handle that is used to communicate with `sc_network::Notifications`. notification_service: Box<dyn NotificationService>, } @@ -119,27 +116,32 @@ where )); // Then start the network loop loop { + // First: handle any command. This cannot be placed in the `select!` macro because it would consume 100% of one CPU core (the command_rx is always ready). + if let Some(command_rx) = self.command_rx.as_mut() { + if let Some(command) = command_rx.next().fuse().await { + self.handle_command(command).await + } + } + futures::select! { _ = self.propagate_timeout.next() => { for (peer, peer_data) in self.peers.iter_mut() { if !peer_data.sent_peering { + debug!(target: "duniter-libp2p", "[{}] sending self peering to {}", self.network.local_peer_id(), peer); match self.notification_service.send_async_notification(peer, self.self_peering.encode()).await { Ok(_) => { peer_data.sent_peering = true; + debug!(target: "duniter-libp2p", "[{}] self peering sent to {}", self.network.local_peer_id(), peer); self.events_reporter.report_event(DuniterPeeringEvent::SelfPeeringPropagationSuccess(*peer, self.self_peering.clone())); } Err(e) => { + debug!(target: "duniter-libp2p", "[{}] failed to send self peering to {}: {}", self.network.local_peer_id(), peer, e); self.events_reporter.report_event(DuniterPeeringEvent::SelfPeeringPropagationFailed(*peer, self.self_peering.clone(), e.to_string())); } } } } }, - command = self.command_rx.next().fuse() => { - if let Some(command) = command { - self.handle_command(command).await - } - }, event = self.notification_service.next_event().fuse() => { if let Some(event) = event { self.handle_notification_event(event) @@ -160,12 +162,14 @@ where result_tx, .. } => { + debug!(target: "duniter-libp2p", "[{}] validating stream from {}", self.network.local_peer_id(), peer); // only accept peers whose role can be determined let result = self .network .peer_role(peer, handshake) .map_or(ValidationResult::Reject, |_| ValidationResult::Accept); let duniter_validation = DuniterStreamValidationResult::from(result); + debug!(target: "duniter-libp2p", "[{}] stream validation result for {}: {:?}", self.network.local_peer_id(), peer, duniter_validation); self.events_reporter .report_event(DuniterPeeringEvent::StreamValidation( peer, @@ -189,23 +193,28 @@ where }, ); debug_assert!(_was_in.is_none()); + debug!(target: "duniter-libp2p", "[{}] stream opened with {peer}", self.network.local_peer_id()); self.events_reporter .report_event(DuniterPeeringEvent::StreamOpened(peer, role)); } NotificationEvent::NotificationStreamClosed { peer } => { let _peer = self.peers.remove(&peer); debug_assert!(_peer.is_some()); + debug!(target: "duniter-libp2p", "[{}] stream closed with {peer}", self.network.local_peer_id()); self.events_reporter .report_event(DuniterPeeringEvent::StreamClosed(peer)); } NotificationEvent::NotificationReceived { peer, notification } => { + debug!(target: "duniter-libp2p", "[{}] received gossip from {}", self.network.local_peer_id(), peer); if let Ok(peering) = <Peering as Decode>::decode(&mut notification.as_ref()) { self.events_reporter .report_event(DuniterPeeringEvent::GossipReceived(peer, true)); + debug!(target: "duniter-libp2p", "[{}] received gossip from {}: {:?}", self.network.local_peer_id(), peer, peering); self.on_peering(peer, peering); } else { self.events_reporter .report_event(DuniterPeeringEvent::GossipReceived(peer, false)); + debug!(target: "duniter-libp2p", "[{}] received gossip from {} but couldn't decode it", self.network.local_peer_id(), peer); self.network.report_peer(peer, rep::BAD_PEERING); } }