Rework duplicate connections check and fix #36

eth/p2p.nim

@@ -70,12 +70,10 @@ proc processIncoming(server: StreamServer,
   yield peerfut
   if not peerfut.failed:
     let peer = peerfut.read()
+    trace "Connection established (incoming)", peer
     if node.peerPool != nil:
-      if not node.peerPool.addPeer(peer):
-        # In case an outgoing connection was added in the meanwhile or a
-        # malicious peer opens multiple connections
-        debug "Disconnecting peer (incoming)", reason = AlreadyConnected
-        await peer.disconnect(AlreadyConnected)
+      node.peerPool.connectingNodes.excl(peer.remote)
+      node.peerPool.addPeer(peer)
 
 proc listeningAddress*(node: EthereumNode): ENode =
   return initENode(node.keys.pubKey, node.address)

eth/p2p/peer_pool.nim

@@ -107,24 +107,19 @@ proc getRandomBootnode(p: PeerPool): Option[Node] =
   if p.discovery.bootstrapNodes.len != 0:
     result = option(p.discovery.bootstrapNodes.rand())
 
-proc addPeer*(pool: PeerPool, peer: Peer): bool =
-  if peer.remote notin pool.connectedNodes:
-    pool.connectedNodes[peer.remote] = peer
-    for o in pool.observers.values:
-      if not o.onPeerConnected.isNil:
-        if o.protocol.isNil or peer.supports(o.protocol):
-          o.onPeerConnected(peer)
-    return true
-  else: return false
+proc addPeer*(pool: PeerPool, peer: Peer) =
+  doAssert(peer.remote notin pool.connectedNodes)
+  pool.connectedNodes[peer.remote] = peer
+  for o in pool.observers.values:
+    if not o.onPeerConnected.isNil:
+      if o.protocol.isNil or peer.supports(o.protocol):
+        o.onPeerConnected(peer)
 
 proc connectToNode*(p: PeerPool, n: Node) {.async.} =
   let peer = await p.connect(n)
   if not peer.isNil:
-    trace "Connection established", peer
-    if not p.addPeer(peer):
-      # In case an incoming connection was added in the meanwhile
-      trace "Disconnecting peer (outgoing)", reason = AlreadyConnected
-      await peer.disconnect(AlreadyConnected)
+    trace "Connection established (outgoing)", peer
+    p.addPeer(peer)
 
 proc connectToNodes(p: PeerPool, nodes: seq[Node]) {.async.} =
   for node in nodes:

eth/p2p/rlpx.nim

@@ -1480,6 +1480,16 @@ proc rlpxAccept*(node: EthereumNode,
                           udpPort: remote.port)
     result.remote = newNode(initEnode(handshake.remoteHPubkey, address))
 
+    # In case there is an outgoing connection started with this peer we give
+    # precedence to that one and we disconnect here with `AlreadyConnected`
+    if result.remote in node.peerPool.connectedNodes or
+        result.remote in node.peerPool.connectingNodes:
+      trace "Duplicate connection in rlpxAccept"
+      raisePeerDisconnected("Peer already connecting or connected",
+                            AlreadyConnected)
+
+    node.peerPool.connectingNodes.incl(result.remote)
+
     await postHelloSteps(result, response)
     ok = true
   except PeerDisconnected as e: