fix: reduce new peer notifications and subscribe to topics when skipping sync (#1075)

Author: MegaRedHand

lib/lambda_ethereum_consensus/beacon/sync_blocks.ex

@@ -29,6 +29,7 @@ defmodule LambdaEthereumConsensus.Beacon.SyncBlocks do
     initial_slot = Misc.compute_start_slot_at_epoch(checkpoint.epoch) + 1
     last_slot = BeaconChain.get_current_slot()
 
+    # If we're around genesis, we consider ourselves synced
     if last_slot > 0 do
       Enum.chunk_every(initial_slot..last_slot, @blocks_per_chunk)
       |> Enum.map(fn chunk ->
@@ -38,6 +39,8 @@ defmodule LambdaEthereumConsensus.Beacon.SyncBlocks do
         %{from: first_slot, count: count}
       end)
       |> perform_sync()
+    else
+      start_subscriptions()
     end
   end
 

lib/lambda_ethereum_consensus/p2p/peerbook.ex

@@ -54,9 +54,13 @@ defmodule LambdaEthereumConsensus.P2P.Peerbook do
 
   @impl true
   def handle_info({:new_peer, peer_id}, peerbook) do
-    :telemetry.execute([:peers, :connection], %{id: peer_id}, %{result: "success"})
-    updated_peerbook = Map.put(peerbook, peer_id, @initial_score)
-    {:noreply, updated_peerbook}
+    if Map.has_key?(peerbook, peer_id) do
+      {:noreply, peerbook}
+    else
+      :telemetry.execute([:peers, :connection], %{id: peer_id}, %{result: "success"})
+      updated_peerbook = Map.put(peerbook, peer_id, @initial_score)
+      {:noreply, updated_peerbook}
+    end
   end
 
   @impl true

native/libp2p_port/internal/discovery/discovery.go

@@ -9,10 +9,12 @@ import (
 	"libp2p_port/internal/reqresp"
 	"libp2p_port/internal/utils"
 	"net"
+	"time"
 
 	"github.com/ethereum/go-ethereum/p2p/discover"
 	"github.com/ethereum/go-ethereum/p2p/enode"
 	"github.com/ethereum/go-ethereum/p2p/enr"
+	"github.com/libp2p/go-libp2p/core/network"
 	"github.com/libp2p/go-libp2p/core/peer"
 	"github.com/libp2p/go-libp2p/core/peerstore"
 	ma "github.com/multiformats/go-multiaddr"
@@ -71,30 +73,18 @@ func lookForPeers(iter enode.Iterator, listener *reqresp.Listener, forkUpdates c
 	currentForkDigest := <-forkUpdates
 	for iter.Next() {
 		node := iter.Node()
+		time.Sleep(1 * time.Millisecond)
 		updateForkDigest(currentForkDigest[:], forkUpdates)
-		if !filterPeer(node, currentForkDigest[:]) {
-			continue
-		}
-		var addrArr []string
-		if node.TCP() != 0 {
-			str := fmt.Sprintf("/ip4/%s/tcp/%d", node.IP(), node.TCP())
-			addrArr = append(addrArr, str)
-		} else if node.UDP() != 0 {
-			str := fmt.Sprintf("/ip4/%s/udp/%d/quic", node.IP(), node.UDP())
-			addrArr = append(addrArr, str)
-		} else {
-			continue
-		}
-		key, err := utils.ConvertToInterfacePubkey(node.Pubkey())
-		if err != nil {
+
+		if !filterPeer(node, currentForkDigest[:], listener) {
 			continue
 		}
-		nodeID, err := peer.IDFromPublicKey(key)
+		addrInfo, err := convertToAddrInfo(node)
 		if err != nil {
 			continue
 		}
 		go func() {
-			listener.AddPeer([]byte(nodeID), addrArr, peerstore.PermanentAddrTTL)
+			listener.AddPeerWithAddrInfo(*addrInfo, peerstore.PermanentAddrTTL)
 		}()
 	}
 }
@@ -117,11 +107,11 @@ func updateForkDigest(currentForkDigest []byte, forkUpdates chan [4]byte) {
 //     connect to.
 //  2. Peer has a valid IP and TCP port set in their enr.
 //  3. ~Peer hasn't been marked as 'bad'~
-//  4. ~Peer is not currently active or connected.~
+//  4. Peer is not currently active or connected.
 //  5. ~Peer is ready to receive incoming connections.~
 //  6. Peer's fork digest in their ENR matches that of
 //     our localnodes.
-func filterPeer(node *enode.Node, currentForkDigest []byte) bool {
+func filterPeer(node *enode.Node, currentForkDigest []byte, listener *reqresp.Listener) bool {
 	// Ignore nil node entries passed in.
 	if node == nil {
 		return false
@@ -135,6 +125,10 @@ func filterPeer(node *enode.Node, currentForkDigest []byte) bool {
 	if err := nodeENR.Load(enr.WithEntry("tcp", new(enr.TCP))); err != nil {
 		return false
 	}
+	peerData, err := convertToAddrInfo(node)
+	if err != nil || listener.Host().Network().Connectedness(peerData.ID) == network.Connected {
+		return false
+	}
 	// Decide whether or not to connect to peer that does not
 	// match the proper fork ENR data with our local node.
 	sszEncodedForkEntry := make([]byte, 16)
@@ -195,16 +189,16 @@ func updateEnr(localNode *enode.LocalNode, e proto_helpers.Enr) {
 	localNode.Set(enr.WithEntry("syncnets", e.Syncnets))
 }
 
-func convertToAddrInfo(node *enode.Node) (*peer.AddrInfo, ma.Multiaddr, error) {
+func convertToAddrInfo(node *enode.Node) (*peer.AddrInfo, error) {
 	multiAddr, err := convertToSingleMultiAddr(node)
 	if err != nil {
-		return nil, nil, err
+		return nil, err
 	}
 	info, err := peer.AddrInfoFromP2pAddr(multiAddr)
 	if err != nil {
-		return nil, nil, err
+		return nil, err
 	}
-	return info, multiAddr, nil
+	return info, nil
 }
 
 func convertToSingleMultiAddr(node *enode.Node) (ma.Multiaddr, error) {