fix(chain manager): drop beacon-coincident outbound peers if differen…

…t superblock consensus
witnet · Nov 23, 2020 · c5aa51d · c5aa51d
1 parent 8605817
commit c5aa51d
Show file tree

Hide file tree

Showing 5 changed files with 42 additions and 9 deletions.
diff --git a/node/.Cargo.toml.swp b/node/.Cargo.toml.swp
diff --git a/node/src/actors/chain_manager/mod.rs b/node/src/actors/chain_manager/mod.rs
@@ -70,7 +70,7 @@ use crate::{
         inventory_manager::InventoryManager,
         json_rpc::JsonRpcServer,
         messages::{
-            AddItem, AddItems, AddTransaction, Anycast, BlockNotify, Broadcast,
+            AddItem, AddItems, AddTransaction, Anycast, BlockNotify, Broadcast, DropOutboundPeers,
             GetBlocksEpochRange, GetItemBlock, NodeStatusNotify, SendInventoryItem,
             SendInventoryRequest, SendLastBeacon, SendSuperBlockVote, StoreInventoryItem,
             SuperBlockNotify,
@@ -1368,6 +1368,9 @@ impl ChainManager {
                         "Superblock consensus {} different from current superblock",
                         target_superblock_hash
                     );
+                    let sessions_manager_addr = SessionsManager::from_registry();
+
+                    sessions_manager_addr.do_send(DropOutboundPeers {});
                     act.initialize_from_storage(ctx);
                     act.update_state_machine(StateMachine::WaitingConsensus);
 

diff --git a/node/src/actors/messages.rs b/node/src/actors/messages.rs
@@ -941,6 +941,13 @@ impl Message for LogMessage {
     type Result = SessionsUnitResult;
 }
 
+/// Drop all outbound peers
+#[derive(Clone, Debug)]
+pub struct DropOutboundPeers {}
+impl Message for DropOutboundPeers {
+    type Result = ();
+}
+
 /// Set the LastBeacon
 #[derive(Clone, Debug)]
 pub struct SetLastBeacon {

diff --git a/node/src/actors/sessions_manager/handlers.rs b/node/src/actors/sessions_manager/handlers.rs
@@ -1,6 +1,7 @@
 use std::{
     fmt::{Debug, Display},
     marker::Send,
+    net::SocketAddr,
 };
 
 use actix::{
@@ -16,9 +17,10 @@ use crate::actors::{
     chain_manager::ChainManager,
     codec::P2PCodec,
     messages::{
-        AddConsolidatedPeer, AddPeers, Anycast, Broadcast, Consolidate, Create, EpochNotification,
-        GetConsolidatedPeers, LogMessage, NumSessions, NumSessionsResult, PeerBeacon, Register,
-        RemoveAddressesFromTried, SessionsUnitResult, SetLastBeacon, TryMineBlock, Unregister,
+        AddConsolidatedPeer, AddPeers, Anycast, Broadcast, Consolidate, Create, DropOutboundPeers,
+        EpochNotification, GetConsolidatedPeers, LogMessage, NumSessions, NumSessionsResult,
+        PeerBeacon, Register, RemoveAddressesFromTried, SessionsUnitResult, SetLastBeacon,
+        TryMineBlock, Unregister,
     },
     peers_manager::PeersManager,
     session::Session,
@@ -468,3 +470,18 @@ impl Handler<SetLastBeacon> for SessionsManager {
         self.last_beacon = Some(msg.beacon);
     }
 }
+
+impl Handler<DropOutboundPeers> for SessionsManager {
+    type Result = <DropOutboundPeers as Message>::Result;
+
+    fn handle(&mut self, _msg: DropOutboundPeers, _ctx: &mut Context<Self>) -> Self::Result {
+        let outbound_peers: Vec<SocketAddr> = self
+            .sessions
+            .outbound_consolidated
+            .collection
+            .keys()
+            .cloned()
+            .collect();
+        self.drop_outbound_peers(outbound_peers.as_ref());
+    }
+}
diff --git a/node/src/actors/sessions_manager/mod.rs b/node/src/actors/sessions_manager/mod.rs
@@ -260,12 +260,8 @@ impl SessionsManager {
                             );
                         }
                         // Unregister peers out of consensus
+                        act.drop_outbound_peers(peers_to_unregister.as_ref());
                         for peer in peers_to_unregister {
-                            if let Some(a) =
-                                act.sessions.outbound_consolidated.collection.get(&peer)
-                            {
-                                a.reference.do_send(CloseSession);
-                            }
                             peers_to_keep.remove(&peer);
                         }
                         // Mark remaining peers as safu
@@ -291,6 +287,16 @@ impl SessionsManager {
                 .cloned(),
         );
     }
+
+    /// Drop outbound peers
+    fn drop_outbound_peers(&mut self, peers_to_unregister: &[SocketAddr]) {
+        // Unregister peers out of consensus
+        for peer in peers_to_unregister {
+            if let Some(a) = self.sessions.outbound_consolidated.collection.get(&peer) {
+                a.reference.do_send(CloseSession);
+            }
+        }
+    }
 }
 
 /// Required traits for being able to retrieve SessionsManager address from registry