fix(network): broken users_cast (#261)

* fix(network): chain book is not updated for exits peer

If peer is already exists but without pubukey, it will not be inserted
to chain book on new session.

* fix(network): SharedSessions by_chain() leaks unconnected addresses

core/network/src/peer_manager/mod.rs

@@ -424,10 +424,18 @@ impl Inner {
         }
     }
 
+    pub fn add_chain_addr(&self, chain_addr: Address, peer: ArcPeer) {
+        self.chain.write().insert(chain_addr, peer);
+    }
+
     pub fn peer(&self, peer_id: &PeerId) -> Option<ArcPeer> {
         self.peers.read().get(peer_id).cloned()
     }
 
+    pub fn peer_by_chain(&self, chain_addr: &Address) -> Option<ArcPeer> {
+        self.chain.read().get(chain_addr).cloned()
+    }
+
     pub fn contains(&self, peer_id: &PeerId) -> bool {
         self.peers.read().contains(peer_id)
     }
@@ -734,6 +742,14 @@ impl PeerManager {
         if !remote_peer.has_pubkey() {
             if let Err(e) = remote_peer.set_pubkey(pubkey.clone()) {
                 error!("impossible, set public key failed {}", e);
+                error!("new session without peer pubkey, chain book will not be updated");
+            }
+        }
+
+        // Update chain book
+        if let Some(chain_addr) = remote_peer.owned_chain_addr() {
+            if self.inner.peer_by_chain(&chain_addr).is_none() {
+                self.inner.add_chain_addr(chain_addr, remote_peer.clone());
             }
         }
 

core/network/src/peer_manager/shared.rs

@@ -79,12 +79,11 @@ impl SessionBook for SharedSessions {
         let mut connected = Vec::new();
         let mut unconnected = Vec::new();
         for addr in addrs {
-            if let Some(peer) = chain.get(&addr) {
-                if peer.connectedness() == Connectedness::Connected {
+            match chain.get(&addr) {
+                Some(peer) if peer.connectedness() == Connectedness::Connected => {
                     connected.push(peer.session_id());
-                } else {
-                    unconnected.push(addr);
                 }
+                _ => unconnected.push(addr),
             }
         }
 
@@ -135,3 +134,39 @@ impl SessionBook for SharedSessions {
             .collect()
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::{SessionBook, SharedSessions, SharedSessionsConfig};
+    use crate::peer_manager::{Inner, Peer};
+
+    use tentacle::secio::SecioKeyPair;
+
+    use std::sync::Arc;
+
+    #[test]
+    fn should_push_not_found_chain_addr_to_unconneded_on_by_chain() {
+        let sess_conf = SharedSessionsConfig {
+            max_stream_window_size: 10,
+            write_timeout:          10,
+        };
+
+        let inner = Arc::new(Inner::new());
+        let sessions = SharedSessions::new(Arc::clone(&inner), sess_conf);
+
+        let keypair = SecioKeyPair::secp256k1_generated();
+        let pubkey = keypair.public_key();
+        let chain_addr = Peer::pubkey_to_chain_addr(&pubkey).expect("chain addr");
+
+        assert!(
+            inner.peer_by_chain(&chain_addr).is_none(),
+            "should not be registered"
+        );
+
+        let (_, unconnected) = sessions.by_chain(vec![chain_addr.clone()]);
+        assert!(
+            unconnected.contains(&chain_addr),
+            "should be inserted to unconnected"
+        );
+    }
+}

core/network/src/peer_manager/test_manager.rs

@@ -2952,3 +2952,36 @@ async fn should_setup_trust_metric_if_none_on_session_blocked() {
         "should have 1 bad event"
     );
 }
+
+#[tokio::test]
+async fn should_update_chain_book_on_new_session() {
+    let (mut mgr, _conn_rx) = make_manager(0, 20);
+
+    let remote_peer = make_peer(2077);
+    let remote_pubkey = remote_peer.owned_pubkey().expect("pubkey");
+    let remote_peer_id = remote_peer.owned_id();
+    let remote_addr = remote_peer.multiaddrs.all_raw().pop().expect("multiaddr");
+    let remote_chain_addr = remote_peer.owned_chain_addr().expect("chain addr");
+
+    let sess_ctx = SessionContext::make(
+        SessionId::new(1),
+        remote_addr.clone(),
+        SessionType::Inbound,
+        remote_pubkey.clone(),
+    );
+    let new_session = PeerManagerEvent::NewSession {
+        pid:    remote_peer_id.clone(),
+        pubkey: remote_pubkey.clone(),
+        ctx:    sess_ctx.arced(),
+    };
+    mgr.poll_event(new_session).await;
+
+    let inner = mgr.core_inner();
+    assert_eq!(inner.connected(), 1, "should have one without bootstrap");
+
+    let peer_by_chain = inner.peer_by_chain(&remote_chain_addr);
+    assert!(peer_by_chain.is_some(), "should insert peer to chain book");
+
+    let peer_id = peer_by_chain.map(|p| p.owned_id());
+    assert_eq!(peer_id, Some(remote_peer_id), "should be peer in session");
+}