feat: close outdated connections to non-RT peers

sn_networking/src/driver.rs

@@ -41,7 +41,7 @@ use libp2p::{
     swarm::{
         behaviour::toggle::Toggle,
         dial_opts::{DialOpts, PeerCondition},
-        DialError, NetworkBehaviour, StreamProtocol, Swarm,
+        ConnectionId, DialError, NetworkBehaviour, StreamProtocol, Swarm,
     },
     Multiaddr, PeerId, Transport,
 };
@@ -58,7 +58,7 @@ use std::{
     net::SocketAddr,
     num::NonZeroUsize,
     path::PathBuf,
-    time::Duration,
+    time::{Duration, Instant},
 };
 use tiny_keccak::{Hasher, Sha3};
 use tokio::sync::{mpsc, oneshot};
@@ -453,8 +453,6 @@ impl NetworkBuilder {
         let gossipsub = if self.enable_gossip {
             // Gossipsub behaviour
             let gossipsub_config = libp2p::gossipsub::ConfigBuilder::default()
-                // disable sending to ALL_PEERS subscribed to a topic, which is the default behaviour
-                .flood_publish(false)
                 // we don't currently require source peer id and/or signing
                 .validation_mode(libp2p::gossipsub::ValidationMode::Permissive)
                 // we use the hash of the msg content as the msg id to deduplicate them
@@ -467,6 +465,10 @@ impl NetworkBuilder {
                 })
                 // set the heartbeat interval to be higher than default 1sec
                 .heartbeat_interval(Duration::from_secs(5))
+                // default is 3sec, increase to 10sec to avoid false alert
+                .iwant_followup_time(Duration::from_secs(10))
+                // default is 10sec, increase to 60sec to reduce the risk of looping
+                .published_message_ids_cache_time(Duration::from_secs(60))
                 .build()
                 .map_err(|err| Error::GossipsubConfigError(err.to_string()))?;
 
@@ -547,6 +549,7 @@ impl NetworkBuilder {
             is_gossip_handler: false,
             network_discovery: NetworkDiscovery::new(&peer_id),
             bootstrap_peers: Default::default(),
+            live_connected_peers: Default::default(),
         };
 
         Ok((
@@ -593,6 +596,9 @@ pub struct SwarmDriver {
     // This is to ensure a more accurate network discovery.
     pub(crate) network_discovery: NetworkDiscovery,
     pub(crate) bootstrap_peers: BTreeMap<Option<u32>, HashSet<PeerId>>,
+    // Peers that having live connection to. Any peer got contacted during kad network query
+    // will have live connection established. And they may not appear in the RT.
+    pub(crate) live_connected_peers: BTreeMap<ConnectionId, (PeerId, Instant)>,
 }
 
 impl SwarmDriver {

sn_networking/src/event.rs

@@ -38,6 +38,7 @@ use sn_protocol::{
 use std::{
     collections::HashSet,
     fmt::{Debug, Formatter},
+    time::{Duration, Instant},
 };
 use tokio::sync::oneshot;
 use tracing::{info, warn};
@@ -459,7 +460,12 @@ impl SwarmDriver {
             } => {
                 event_string = "ConnectionEstablished";
                 trace!(%peer_id, num_established, "ConnectionEstablished ({connection_id:?}): {}", endpoint_str(&endpoint));
-                info!("{:?}", self.swarm.network_info());
+                info!(%peer_id, ?connection_id, "ConnectionEstablished {:?}", self.swarm.network_info());
+
+                let _ = self.live_connected_peers.insert(
+                    connection_id,
+                    (peer_id, Instant::now() + Duration::from_secs(60)),
+                );
 
                 if endpoint.is_dialer() {
                     self.dialed_peers
@@ -475,8 +481,9 @@ impl SwarmDriver {
                 connection_id,
             } => {
                 event_string = "ConnectionClosed";
-                info!("{:?}", self.swarm.network_info());
+                info!(%peer_id, ?connection_id, "ConnectionClosed: {:?}", self.swarm.network_info());
                 trace!(%peer_id, ?connection_id, ?cause, num_established, "ConnectionClosed: {}", endpoint_str(&endpoint));
+                let _ = self.live_connected_peers.remove(&connection_id);
             }
             SwarmEvent::OutgoingConnectionError {
                 peer_id: Some(failed_peer_id),
@@ -582,6 +589,9 @@ impl SwarmDriver {
                 trace!("SwarmEvent has been ignored: {other:?}")
             }
         }
+
+        self.remove_outdated_connections();
+
         trace!(
             "SwarmEvent handled in {:?}: {event_string:?}",
             start.elapsed()
@@ -1025,6 +1035,36 @@ impl SwarmDriver {
                 .remove_peer(&to_be_removed_bootstrap);
         }
     }
+
+    // Remove outdated connection to a peer if it is not in the RT.
+    fn remove_outdated_connections(&mut self) {
+        let mut shall_removed = vec![];
+
+        self.live_connected_peers
+            .retain(|connection_id, (peer_id, timeout)| {
+                let shall_retained = *timeout > Instant::now();
+                if !shall_retained {
+                    shall_removed.push((*connection_id, *peer_id))
+                }
+                shall_retained
+            });
+
+        // Only remove outdated peer not in the RT
+        for (connection_id, peer_id) in shall_removed {
+            if let Some(kbucket) = self.swarm.behaviour_mut().kademlia.kbucket(peer_id) {
+                if kbucket
+                    .iter()
+                    .any(|peer_entry| peer_id == *peer_entry.node.key.preimage())
+                {
+                    // Skip the connection as peer presents in the RT.
+                    continue;
+                }
+            }
+
+            info!("Removing outdated connection {connection_id:?} to {peer_id:?}");
+            let _result = self.swarm.close_connection(connection_id);
+        }
+    }
 }
 
 /// Helper function to print formatted connection role info.

sn_node/src/node.rs

@@ -42,8 +42,9 @@ use tokio::{
 /// serialised transfer info encrypted against the referenced public key.
 pub const ROYALTY_TRANSFER_NOTIF_TOPIC: &str = "ROYALTY_TRANSFER_NOTIFICATION";
 
-/// Defines the percentage (ie 1/FORWARDER_CHOOSING_FACTOR th of all nodes) of nodes which will act as royalty_transfer_notify forwarder.
-const FORWARDER_CHOOSING_FACTOR: usize = 50;
+/// Defines the percentage (ie 1/FORWARDER_CHOOSING_FACTOR th of all nodes) of nodes
+/// which will act as royalty_transfer_notify forwarder.
+const FORWARDER_CHOOSING_FACTOR: usize = 10;
 
 /// Interval to trigger replication of all records to all peers.
 /// This is the max time it should take. Minimum interval at any ndoe will be half this