P2P network stuck handler
Watch for situations in which no gossip has been received recently,
and actively try to unstuck those situations by checking if some
nodes can be lifted from quarantine and try to connect to them if
that's the case.
@@ -2,9 +2,12 @@ use super::{Gossip, Gossips, P2pTopology, Peer};
use crate::intercom::{NetworkMsg, PropagateMsg, TopologyMsg};
use crate::settings::start::network::Configuration;
use crate::utils::async_msg::{MessageBox, MessageQueue};
-use tokio::time::Interval;
+use std::time::Duration;
+use tokio::time::{Instant, Interval};
use tokio_stream::StreamExt;
+const STUCK_NETWORK_WARNING: Duration = Duration::from_secs(60 * 5); // 5 min
+
struct Process {
input: MessageQueue<TopologyMsg>,
network_msgbox: MessageBox<NetworkMsg>,
@@ -47,11 +50,17 @@ pub async fn start(task_data: TaskData) {
impl Process {
async fn handle_input(&mut self) {
+ let mut last_update = Instant::now();
+ let mut stuck_inverval = tokio::time::interval(Duration::from_secs(60));
+
loop {
tokio::select! {
Some(input) = self.input.next() => {
match input {
- TopologyMsg::AcceptGossip(gossip) => self.topology.accept_gossips(gossip),
+ TopologyMsg::AcceptGossip(gossip) => {
+ self.topology.accept_gossips(gossip);
+ last_update = Instant::now();
+ },
TopologyMsg::DemotePeer(id) => self.topology.report_node(&id),
TopologyMsg::PromotePeer(id) => self.topology.promote_node(&id),
TopologyMsg::View(selection, handle) => {
@@ -71,10 +80,30 @@ impl Process {
_ = self.gossip_interval.tick() => {
self.topology.update_gossip();
let view = self.topology.view(poldercast::layer::Selection::Any);
- for peer in view.peers {
- // Peers returned by the topology will always have a NodeId
- let gossip = self.topology.initiate_gossips(&peer.id());
+ if view.peers.is_empty() {
+ tracing::warn!("no peers to gossip with found, check your connection");
+ }
+ self.send_gossip_messages(view.peers)
+ }
+ // For some reason, resetting a tokio::time::Sleep instance when receiving a new gossip
+ // result in a stack overflow / allocation failure in some tests.
+ // This should achieve the same thing, without hopefully the blowing-up-the-stack part
+ _ = stuck_inverval.tick() => {
+ if last_update.elapsed() >= self.network_stuck_check {
+ tracing::warn!("p2p network have been too quiet for some time, will try to contact nodes for which quarantine have elapsed");
+ let quarantined_nodes = self.topology.lift_nodes_from_quarantine();
+ self.send_gossip_messages(quarantined_nodes);
+
+ last_update = Instant::now();
+ }
+ }
+ }
+ }
+ }
+ fn send_gossip_messages(&mut self, peers: Vec<Peer>) {
+ for peer in peers {
+ let gossip = self.topology.initiate_gossips(&peer.id());
self.network_msgbox
// do not block the current thread to avoid deadlocks
.try_send(NetworkMsg::Propagate(PropagateMsg::Gossip(peer, gossip)))
@@ -87,6 +116,3 @@ impl Process {
}
}
}
- }
- }
-}
