refactor: allow duplicate handshakes

pkg/p2p/libp2p/connections_test.go

@@ -489,7 +489,6 @@ func TestConnectWithEnabledWSTransports(t *testing.T) {
 
 // TestConnectRepeatHandshake tests if handshake was attempted more then once by the same peer
 func TestConnectRepeatHandshake(t *testing.T) {
-	t.Skip("test flaking")
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 
@@ -518,12 +517,12 @@ func TestConnectRepeatHandshake(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	if _, err := s2.HandshakeService().Handshake(ctx, libp2p.NewStream(stream), info.Addrs[0], info.ID); err == nil {
-		t.Fatalf("expected stream error")
+	if _, err := s2.HandshakeService().Handshake(ctx, libp2p.NewStream(stream), info.Addrs[0], info.ID); err != nil {
+		t.Fatal(err)
 	}
 
-	expectPeersEventually(t, s2)
-	expectPeersEventually(t, s1)
+	expectPeersEventually(t, s2, overlay1)
+	expectPeersEventually(t, s1, overlay2)
 }
 
 func TestBlocklisting(t *testing.T) {

pkg/p2p/libp2p/internal/handshake/handshake.go

@@ -8,7 +8,6 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"sync"
 	"sync/atomic"
 	"time"
 
@@ -20,7 +19,6 @@ import (
 	"github.com/ethersphere/bee/pkg/p2p/protobuf"
 	"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"
 )
@@ -41,9 +39,6 @@ var (
 	// ErrNetworkIDIncompatible is returned if response from the other peer does not have valid networkID.
 	ErrNetworkIDIncompatible = errors.New("incompatible network ID")
 
-	// ErrHandshakeDuplicate is returned  if the handshake response has been received by an already processed peer.
-	ErrHandshakeDuplicate = errors.New("duplicate handshake")
-
 	// ErrInvalidAck is returned if data in received in ack is not valid (invalid signature for example).
 	ErrInvalidAck = errors.New("invalid ack")
 
@@ -76,12 +71,9 @@ type Service struct {
 	transaction           []byte
 	networkID             uint64
 	welcomeMessage        atomic.Value
-	receivedHandshakes    map[libp2ppeer.ID]struct{}
-	receivedHandshakesMu  sync.Mutex
 	logger                logging.Logger
 	libp2pID              libp2ppeer.ID
 	metrics               metrics
-	network.Notifiee      // handshake service can be the receiver for network.Notify
 	picker                p2p.Picker
 }
 
@@ -113,11 +105,9 @@ func New(signer crypto.Signer, advertisableAddresser AdvertisableAddressResolver
 		fullNode:              fullNode,
 		transaction:           transaction,
 		senderMatcher:         isSender,
-		receivedHandshakes:    make(map[libp2ppeer.ID]struct{}),
 		libp2pID:              ownPeerID,
 		logger:                logger,
 		metrics:               newMetrics(),
-		Notifiee:              new(network.NoopNotifiee),
 	}
 	svc.welcomeMessage.Store(welcomeMessage)
 
@@ -233,14 +223,6 @@ func (s *Service) Handle(ctx context.Context, stream p2p.Stream, remoteMultiaddr
 	ctx, cancel := context.WithTimeout(ctx, handshakeTimeout)
 	defer cancel()
 
-	s.receivedHandshakesMu.Lock()
-	if _, exists := s.receivedHandshakes[remotePeerID]; exists {
-		s.receivedHandshakesMu.Unlock()
-		return nil, ErrHandshakeDuplicate
-	}
-
-	s.receivedHandshakes[remotePeerID] = struct{}{}
-	s.receivedHandshakesMu.Unlock()
 	w, r := protobuf.NewWriterAndReader(stream)
 	fullRemoteMA, err := buildFullMA(remoteMultiaddr, remotePeerID)
 	if err != nil {
@@ -342,13 +324,6 @@ func (s *Service) Handle(ctx context.Context, stream p2p.Stream, remoteMultiaddr
 	}, nil
 }
 
-// Disconnected is called when the peer disconnects.
-func (s *Service) Disconnected(_ network.Network, c network.Conn) {
-	s.receivedHandshakesMu.Lock()
-	defer s.receivedHandshakesMu.Unlock()
-	delete(s.receivedHandshakes, c.RemotePeer())
-}
-
 // SetWelcomeMessage sets the new handshake welcome message.
 func (s *Service) SetWelcomeMessage(msg string) (err error) {
 	if len(msg) > MaxWelcomeMessageLength {

pkg/p2p/libp2p/internal/handshake/handshake_test.go

@@ -601,68 +601,6 @@ func TestHandshake(t *testing.T) {
 		}
 	})
 
-	t.Run("Handle - duplicate handshake", func(t *testing.T) {
-		handshakeService, err := handshake.New(signer1, aaddresser, senderMatcher, node1Info.BzzAddress.Overlay, networkID, true, trxHash, "", node1AddrInfo.ID, logger)
-		if err != nil {
-			t.Fatal(err)
-		}
-		var buffer1 bytes.Buffer
-		var buffer2 bytes.Buffer
-		stream1 := mock.NewStream(&buffer1, &buffer2)
-		stream2 := mock.NewStream(&buffer2, &buffer1)
-
-		w := protobuf.NewWriter(stream2)
-		if err := w.WriteMsg(&pb.Syn{
-			ObservedUnderlay: node1maBinary,
-		}); err != nil {
-			t.Fatal(err)
-		}
-
-		if err := w.WriteMsg(&pb.Ack{
-			Address: &pb.BzzAddress{
-				Underlay:  node2maBinary,
-				Overlay:   node2BzzAddress.Overlay.Bytes(),
-				Signature: node2BzzAddress.Signature,
-			},
-			NetworkID:   networkID,
-			Transaction: trxHash,
-			FullNode:    true,
-		}); err != nil {
-			t.Fatal(err)
-		}
-
-		res, err := handshakeService.Handle(context.Background(), stream1, node2AddrInfo.Addrs[0], node2AddrInfo.ID)
-		if err != nil {
-			t.Fatal(err)
-		}
-
-		testInfo(t, *res, node2Info)
-
-		_, r := protobuf.NewWriterAndReader(stream2)
-		var got pb.SynAck
-		if err := r.ReadMsg(&got); err != nil {
-			t.Fatal(err)
-		}
-
-		if !bytes.Equal(got.Syn.ObservedUnderlay, node2maBinary) {
-			t.Fatalf("got bad syn")
-		}
-		bzzAddress, err := bzz.ParseAddress(got.Ack.Address.Underlay, got.Ack.Address.Overlay, got.Ack.Address.Signature, got.Ack.Transaction, blockhash, got.Ack.NetworkID)
-		if err != nil {
-			t.Fatal(err)
-		}
-
-		testInfo(t, node1Info, handshake.Info{
-			BzzAddress: bzzAddress,
-			FullNode:   got.Ack.FullNode,
-		})
-
-		_, err = handshakeService.Handle(context.Background(), stream1, node2AddrInfo.Addrs[0], node2AddrInfo.ID)
-		if err != handshake.ErrHandshakeDuplicate {
-			t.Fatalf("expected %s, got %s", handshake.ErrHandshakeDuplicate, err)
-		}
-	})
-
 	t.Run("Handle - invalid ack", func(t *testing.T) {
 		handshakeService, err := handshake.New(signer1, aaddresser, senderMatcher, node1Info.BzzAddress.Overlay, networkID, true, nil, "", node1AddrInfo.ID, logger)
 		if err != nil {

pkg/p2p/libp2p/libp2p.go

@@ -293,8 +293,7 @@ func New(ctx context.Context, signer beecrypto.Signer, networkID uint64, overlay
 	s.host.SetStreamHandlerMatch(id, matcher, s.handleIncoming)
 
 	connMetricNotify := newConnMetricNotify(s.metrics)
-	h.Network().Notify(peerRegistry)       // update peer registry on network events
-	h.Network().Notify(s.handshakeService) // update handshake service on network events
+	h.Network().Notify(peerRegistry) // update peer registry on network events
 	h.Network().Notify(connMetricNotify)
 	return s, nil
 }