Resolve peer topology inconsistencies (#1032)

pkg/p2p/libp2p/connections_test.go

@@ -194,13 +194,14 @@ func TestDoubleConnectOnAllAddresses(t *testing.T) {
 
 	s1, overlay1 := newService(t, 1, libp2pServiceOpts{})
 
-	s2, overlay2 := newService(t, 1, libp2pServiceOpts{})
-
 	addrs, err := s1.Addresses()
 	if err != nil {
 		t.Fatal(err)
 	}
 	for _, addr := range addrs {
+		// creating new remote host for each address
+		s2, overlay2 := newService(t, 1, libp2pServiceOpts{})
+
 		if _, err := s2.Connect(ctx, addr); err != nil {
 			t.Fatal(err)
 		}
@@ -221,6 +222,8 @@ func TestDoubleConnectOnAllAddresses(t *testing.T) {
 
 		expectPeers(t, s2)
 		expectPeersEventually(t, s1)
+
+		s2.Close()
 	}
 }
 

pkg/p2p/libp2p/libp2p.go

@@ -425,9 +425,13 @@ func (s *Service) Blocklist(overlay swarm.Address, duration time.Duration) error
 	return nil
 }
 
+func buildHostAddress(peerID libp2ppeer.ID) (ma.Multiaddr, error) {
+	return ma.NewMultiaddr(fmt.Sprintf("/p2p/%s", peerID.Pretty()))
+}
+
 func buildUnderlayAddress(addr ma.Multiaddr, peerID libp2ppeer.ID) (ma.Multiaddr, error) {
 	// Build host multiaddress
-	hostAddr, err := ma.NewMultiaddr(fmt.Sprintf("/p2p/%s", peerID.Pretty()))
+	hostAddr, err := buildHostAddress(peerID)
 	if err != nil {
 		return nil, err
 	}
@@ -442,7 +446,14 @@ func (s *Service) Connect(ctx context.Context, addr ma.Multiaddr) (address *bzz.
 		return nil, fmt.Errorf("addr from p2p: %w", err)
 	}
 
-	if _, found := s.peers.overlay(info.ID); found {
+	hostAddr, err := buildHostAddress(info.ID)
+	if err != nil {
+		return nil, fmt.Errorf("build host address: %w", err)
+	}
+
+	remoteAddr := addr.Decapsulate(hostAddr)
+
+	if _, found := s.peers.isConnected(info.ID, remoteAddr); found {
 		return nil, p2p.ErrAlreadyConnected
 	}
 

pkg/p2p/libp2p/peer.go

@@ -14,6 +14,7 @@ import (
 	"github.com/ethersphere/bee/pkg/swarm"
 	"github.com/libp2p/go-libp2p-core/network"
 	libp2ppeer "github.com/libp2p/go-libp2p-core/peer"
+	ma "github.com/multiformats/go-multiaddr"
 )
 
 type peerRegistry struct {
@@ -163,6 +164,35 @@ func (r *peerRegistry) overlay(peerID libp2ppeer.ID) (swarm.Address, bool) {
 	return overlay, found
 }
 
+func (r *peerRegistry) isConnected(peerID libp2ppeer.ID, remoteAddr ma.Multiaddr) (swarm.Address, bool) {
+	if remoteAddr == nil {
+		return swarm.ZeroAddress, false
+	}
+
+	r.mu.RLock()
+	defer r.mu.RUnlock()
+
+	overlay, found := r.overlays[peerID]
+	if !found {
+		return swarm.ZeroAddress, false
+	}
+
+	// check connection remote address
+	conns, ok := r.connections[peerID]
+	if !ok {
+		return swarm.ZeroAddress, false
+	}
+
+	for c := range conns {
+		if c.RemoteMultiaddr().Equal(remoteAddr) {
+			// we ARE connected to the peer on expected address
+			return overlay, true
+		}
+	}
+
+	return swarm.ZeroAddress, false
+}
+
 func (r *peerRegistry) remove(overlay swarm.Address) (bool, libp2ppeer.ID) {
 	r.mu.Lock()
 	peerID, found := r.underlays[overlay.ByteString()]