fix EvtPeerIdentificationCompleted handler

Author: algorandskiy

network/p2p/p2p.go

@@ -201,7 +201,7 @@ func MakeService(ctx context.Context, log logging.Logger, cfg config.Local, h ho
 	sm := makeStreamManager(ctx, log, h, wsStreamHandlers, cfg.EnableGossipService)
 	sub, err := h.EventBus().Subscribe(new(event.EvtPeerIdentificationCompleted))
 	if err != nil {
-		err0 := fmt.Errorf("Failed to subscribe to peer identification events: %v", err)
+		err0 := fmt.Errorf("failed to subscribe to peer identification events: %v", err)
 		return nil, err0
 	}
 	go sm.peerWatcher(ctx, sub)

network/p2p/streams.go

@@ -109,34 +109,32 @@ func (n *streamManager) dispatch(ctx context.Context, remotePeer peer.ID, stream
 	}
 }
 
-// Connected is called when a connection is opened
-// for both incoming (listener -> addConn) and outgoing (dialer -> addConn) connections.
-func (n *streamManager) Connected(net network.Network, conn network.Conn) {
-}
-
 func (n *streamManager) peerWatcher(ctx context.Context, sub event.Subscription) {
 	defer sub.Close()
 	for e := range sub.Out() {
 		evt := e.(event.EvtPeerIdentificationCompleted)
 		conn := evt.Conn
-		if conn.Stat().Direction == network.DirInbound && !n.allowIncomingGossip {
-			n.log.Debugf("ignoring incoming connection from %s", conn.RemotePeer().String())
-			continue
-		}
 
 		remotePeer := conn.RemotePeer()
 		localPeer := n.host.ID()
 
+		if conn.Stat().Direction == network.DirInbound && !n.allowIncomingGossip {
+			n.log.Debugf("%s: ignoring incoming connection from %s", localPeer.String(), remotePeer.String())
+			continue
+		}
+
 		// ensure that only one of the peers initiates the stream
 		if localPeer > remotePeer {
-			return
+			n.log.Debugf("%s: ignoring a lesser peer ID %s", localPeer.String(), remotePeer.String())
+			continue
 		}
 
 		n.streamsLock.Lock()
 		_, ok := n.streams[remotePeer]
 		if ok {
 			n.streamsLock.Unlock()
-			return // there's already an active stream with this peer for our protocol
+			n.log.Debugf("%s: already have a stream to/from %s", localPeer.String(), remotePeer.String())
+			continue // there's already an active stream with this peer for our protocol
 		}
 
 		protos := evt.Protocols
@@ -147,9 +145,9 @@ func (n *streamManager) peerWatcher(ctx context.Context, sub event.Subscription)
 
 		stream, err := n.host.NewStream(n.ctx, remotePeer, targetProto)
 		if err != nil {
-			n.log.Infof("Failed to open stream to %s (%s): %v", remotePeer, conn.RemoteMultiaddr().String(), err)
+			n.log.Infof("%s: failed to open stream to %s (%s): %v", localPeer.String(), remotePeer, conn.RemoteMultiaddr().String(), err)
 			n.streamsLock.Unlock()
-			return
+			continue
 		}
 		n.streams[remotePeer] = stream
 		n.streamsLock.Unlock()
@@ -158,7 +156,7 @@ func (n *streamManager) peerWatcher(ctx context.Context, sub event.Subscription)
 		if handler, ok := n.handlers[targetProto]; ok {
 			handler(n.ctx, remotePeer, stream, incoming)
 		} else {
-			n.log.Errorf("No handler for protocol %s, peer %s", targetProto, remotePeer)
+			n.log.Errorf("%s: no handler for protocol %s, peer %s", localPeer.String(), targetProto, remotePeer)
 			_ = stream.Reset()
 		}
 
@@ -170,6 +168,11 @@ func (n *streamManager) peerWatcher(ctx context.Context, sub event.Subscription)
 	}
 }
 
+// Connected is called when a connection is opened
+// for both incoming (listener -> addConn) and outgoing (dialer -> addConn) connections.
+func (n *streamManager) Connected(net network.Network, conn network.Conn) {
+}
+
 // Disconnected is called when a connection is closed
 func (n *streamManager) Disconnected(net network.Network, conn network.Conn) {
 	n.streamsLock.Lock()

node/node_test.go

@@ -1022,15 +1022,15 @@ func TestNodeP2PRelays(t *testing.T) {
 		switch i {
 		case 0:
 			// node R1 connects to R2
-			t.Logf("Node%d phonebook: %s", i, ni[1].p2pMultiAddr())
+			t.Logf("Node%d %s phonebook: %s", i, ni[0].p2pID, ni[1].p2pMultiAddr())
 			return []string{ni[1].p2pMultiAddr()}
 		case 1:
 			// node R2 connects to none one
-			t.Logf("Node%d phonebook: empty", i)
+			t.Logf("Node%d %s phonebook: empty", i, ni[1].p2pID)
 			return []string{}
 		case 2:
-			// node N only connects to R1
-			t.Logf("Node%d phonebook: %s", i, ni[1].p2pMultiAddr())
+			// node N only connects to R2
+			t.Logf("Node%d %s phonebook: %s", i, ni[2].p2pID, ni[1].p2pMultiAddr())
 			return []string{ni[1].p2pMultiAddr()}
 		default:
 			t.Errorf("not expected number of nodes: %d", i)