Explicit Peering Agreement implementation (#8098)

networking/p2p/src/main/java/tech/pegasys/teku/networking/p2p/gossip/config/GossipConfig.java

@@ -18,6 +18,7 @@
 import java.time.Duration;
 import java.util.function.Consumer;
 import tech.pegasys.teku.infrastructure.exceptions.InvalidConfigurationException;
+import tech.pegasys.teku.networking.p2p.gossip.config.GossipPeerScoringConfig.DirectPeerManager;
 
 /**
  * Gossip options
@@ -32,8 +33,8 @@ public class GossipConfig {
   private static final int DEFAULT_ADVERTISE = 3;
   private static final int DEFAULT_HISTORY = 6;
   static final Duration DEFAULT_HEARTBEAT_INTERVAL = Duration.ofMillis(700);
-  // After EIP-7045, attestations are valid for up to 2 full epochs, so TTL is 65 slots
-  // 1115 * HEARTBEAT = 1115 * 0.7 / 12 = 65.125
+  // After EIP-7045, attestations are valid for up to 2 full epochs, so TTL is 65
+  // slots 1115 * HEARTBEAT = 1115 * 0.7 / 12 = 65.125
   static final Duration DEFAULT_SEEN_TTL = DEFAULT_HEARTBEAT_INTERVAL.multipliedBy(1115);
 
   private final int d;
@@ -215,5 +216,11 @@ public Builder seenTTL(final Duration seenTTL) {
       this.seenTTL = seenTTL;
       return this;
     }
+
+    public Builder directPeerManager(final DirectPeerManager directPeerManager) {
+      checkNotNull(directPeerManager);
+      this.scoringConfigBuilder.directPeerManager(directPeerManager);
+      return this;
+    }
   }
 }

networking/p2p/src/main/java/tech/pegasys/teku/networking/p2p/gossip/config/GossipScoringConfig.java

@@ -16,7 +16,9 @@
 import static com.google.common.base.Preconditions.checkNotNull;
 
 import java.util.Map;
+import java.util.Optional;
 import java.util.function.Consumer;
+import tech.pegasys.teku.networking.p2p.gossip.config.GossipPeerScoringConfig.DirectPeerManager;
 
 /** Gossip scoring config. Contains peer scoring and topic scoring configs. */
 public class GossipScoringConfig {
@@ -163,5 +165,12 @@ public Builder opportunisticGraftThreshold(final Double opportunisticGraftThresh
       this.opportunisticGraftThreshold = opportunisticGraftThreshold;
       return this;
     }
+
+    public Builder directPeerManager(final DirectPeerManager directPeerManager) {
+      checkNotNull(directPeerManager);
+
+      this.peerScoringConfigBuilder.directPeerManager(Optional.of(directPeerManager));
+      return this;
+    }
   }
 }

networking/p2p/src/main/java/tech/pegasys/teku/networking/p2p/network/config/NetworkConfig.java

@@ -21,6 +21,7 @@
 import java.net.SocketException;
 import java.net.UnknownHostException;
 import java.util.Enumeration;
+import java.util.List;
 import java.util.Optional;
 import java.util.OptionalInt;
 import java.util.function.Consumer;
@@ -29,6 +30,8 @@
 import tech.pegasys.teku.infrastructure.exceptions.InvalidConfigurationException;
 import tech.pegasys.teku.infrastructure.io.PortAvailability;
 import tech.pegasys.teku.networking.p2p.gossip.config.GossipConfig;
+import tech.pegasys.teku.networking.p2p.gossip.config.GossipPeerScoringConfig.DirectPeerManager;
+import tech.pegasys.teku.networking.p2p.peer.NodeId;
 
 public class NetworkConfig {
 
@@ -261,5 +264,12 @@ public Builder yamuxEnabled(final boolean yamuxEnabled) {
       this.yamuxEnabled = yamuxEnabled;
       return this;
     }
+
+    public Builder directPeers(final List<NodeId> directPeers) {
+      checkNotNull(directPeers);
+      final DirectPeerManager directPeerManager = directPeers::contains;
+      this.gossipConfigBuilder.directPeerManager(directPeerManager);
+      return this;
+    }
   }
 }

networking/p2p/src/test/java/tech/pegasys/teku/networking/p2p/network/config/NetworkConfigTest.java

@@ -15,8 +15,13 @@
 
 import static org.assertj.core.api.Assertions.assertThat;
 
+import io.libp2p.core.multiformats.Multiaddr;
+import java.util.List;
 import java.util.Optional;
 import org.junit.jupiter.api.Test;
+import tech.pegasys.teku.networking.p2p.gossip.config.GossipPeerScoringConfig.DirectPeerManager;
+import tech.pegasys.teku.networking.p2p.libp2p.LibP2PNodeId;
+import tech.pegasys.teku.networking.p2p.peer.NodeId;
 
 @SuppressWarnings("AddressSelection")
 class NetworkConfigTest {
@@ -75,4 +80,32 @@ void checkPrivateKeySourceCreatedCorrectly() {
   private NetworkConfig createConfig() {
     return NetworkConfig.builder().advertisedIp(advertisedIp).networkInterface(listenIp).build();
   }
+
+  @Test
+  void checkDirectPeersConfigCreatedCorrectly() {
+    final String peerAddress1 =
+        "/ip4/198.51.100.0/tcp/4242/p2p/QmYyQSo1c1Ym7orWxLYvCrM2EmxFTANf8wXmmE7DWjhx5N";
+    final String peerAddress2 =
+        "/ip4/198.51.100.0/tcp/4242/p2p/QmTESTo1c1Ym7orWxLYvCrM2EmxFTANf8wXmmE7DWjTEST";
+    final LibP2PNodeId peerId1 = new LibP2PNodeId(new Multiaddr(peerAddress1).getPeerId());
+    final LibP2PNodeId peerId2 = new LibP2PNodeId(new Multiaddr(peerAddress2).getPeerId());
+
+    final List<NodeId> directPeers = List.of(peerId1);
+
+    final NetworkConfig config =
+        NetworkConfig.builder()
+            .advertisedIp(advertisedIp)
+            .networkInterface(listenIp)
+            .directPeers(directPeers)
+            .build();
+
+    final Optional<DirectPeerManager> optionalDirectPeerManager =
+        config.getGossipConfig().getScoringConfig().getPeerScoringConfig().getDirectPeerManager();
+    assertThat(optionalDirectPeerManager).isPresent();
+
+    final DirectPeerManager manager = optionalDirectPeerManager.get();
+
+    assert manager.isDirectPeer(peerId1);
+    assert !manager.isDirectPeer(peerId2);
+  }
 }

teku/src/main/java/tech/pegasys/teku/cli/options/P2POptions.java

@@ -26,6 +26,7 @@
 import tech.pegasys.teku.config.TekuConfiguration;
 import tech.pegasys.teku.networking.eth2.P2PConfig;
 import tech.pegasys.teku.networking.p2p.discovery.DiscoveryConfig;
+import tech.pegasys.teku.networking.p2p.libp2p.MultiaddrPeerAddress;
 import tech.pegasys.teku.networking.p2p.network.config.NetworkConfig;
 
 public class P2POptions {
@@ -143,11 +144,23 @@ public class P2POptions {
   @Option(
       names = {"--p2p-static-peers"},
       paramLabel = "<PEER_ADDRESSES>",
-      description = "Static peers",
+      description =
+          "Specifies a list of 'static' peers with which to establish and maintain connections",
       split = ",",
       arity = "0..*")
   private List<String> p2pStaticPeers = new ArrayList<>();
 
+  @Option(
+      names = {"--p2p-direct-peers"},
+      paramLabel = "<PEER_ADDRESSES>",
+      description =
+          "Specifies a list of 'direct' peers with which to establish and maintain connections.\n"
+              + "Direct peers are static peers with which this node will always exchange full messages, regardless of peer scoring mechanisms.\n"
+              + "Such peers will also need to enable you as direct in order to work.",
+      split = ",",
+      arity = "0..*")
+  private List<String> p2pDirectPeers = new ArrayList<>();
+
   @Option(
       names = {"--Xp2p-multipeer-sync-enabled"},
       paramLabel = "<BOOLEAN>",
@@ -321,6 +334,9 @@ private int getP2pUpperBound() {
   }
 
   public void configure(final TekuConfiguration.Builder builder) {
+    // From a discovery configuration perspective, direct peers are static peers
+    p2pStaticPeers.addAll(p2pDirectPeers);
+
     builder
         .p2p(
             b ->
@@ -362,6 +378,13 @@ public void configure(final TekuConfiguration.Builder builder) {
               if (p2pAdvertisedPort != null) {
                 n.advertisedPort(OptionalInt.of(p2pAdvertisedPort));
               }
+              if (!p2pDirectPeers.isEmpty()) {
+                n.directPeers(
+                    p2pDirectPeers.stream()
+                        .map(MultiaddrPeerAddress::fromAddress)
+                        .map(MultiaddrPeerAddress::getId)
+                        .toList());
+              }
               n.networkInterface(p2pInterface)
                   .isEnabled(p2pEnabled)
                   .listenPort(p2pPort)