Create/update/store ethereum node record(ENR) (#1680)

* #1561 - Create/store/update ENR when local node is created

Signed-off-by: David Mechler <david.mechler@consensys.net>

* #1561 - spotlessApply

Signed-off-by: David Mechler <david.mechler@consensys.net>

* Update discovery library to prod version 0.4.1

Signed-off-by: David Mechler <david.mechler@consensys.net>

acceptance-tests/dsl/src/main/java/org/hyperledger/besu/tests/acceptance/dsl/node/ThreadBesuNodeRunner.java

@@ -204,6 +204,7 @@ public class ThreadBesuNodeRunner implements BesuNodeRunner {
                     .collect(Collectors.toList()))
             .besuPluginContext(new BesuPluginContextImpl())
             .autoLogBloomCaching(false)
+            .storageProvider(storageProvider)
             .build();
 
     runner.start();

besu/src/main/java/org/hyperledger/besu/RunnerBuilder.java

@@ -89,6 +89,7 @@ import org.hyperledger.besu.ethereum.permissioning.node.InsufficientPeersPermiss
 import org.hyperledger.besu.ethereum.permissioning.node.NodePermissioningController;
 import org.hyperledger.besu.ethereum.permissioning.node.PeerPermissionsAdapter;
 import org.hyperledger.besu.ethereum.privacy.PrivateTransactionObserver;
+import org.hyperledger.besu.ethereum.storage.StorageProvider;
 import org.hyperledger.besu.ethereum.stratum.StratumServer;
 import org.hyperledger.besu.ethereum.transaction.TransactionSimulator;
 import org.hyperledger.besu.ethereum.worldstate.WorldStateArchive;
@@ -167,6 +168,7 @@ public class RunnerBuilder {
   private BesuPluginContextImpl besuPluginContext;
   private boolean autoLogBloomCaching = true;
   private boolean randomPeerPriority;
+  private StorageProvider storageProvider;
 
   public RunnerBuilder vertx(final Vertx vertx) {
     this.vertx = vertx;
@@ -333,6 +335,11 @@ public class RunnerBuilder {
     return this;
   }
 
+  public RunnerBuilder storageProvider(final StorageProvider storageProvider) {
+    this.storageProvider = storageProvider;
+    return this;
+  }
+
   public Runner build() {
 
     Preconditions.checkNotNull(besuController);
@@ -418,6 +425,7 @@ public class RunnerBuilder {
                 .supportedCapabilities(caps)
                 .natService(natService)
                 .randomPeerPriority(randomPeerPriority)
+                .storageProvider(storageProvider)
                 .build();
 
     final NetworkRunner networkRunner =

besu/src/main/java/org/hyperledger/besu/cli/BesuCommand.java

@@ -2139,6 +2139,7 @@ public class BesuCommand implements DefaultCommandValues, Runnable {
             .autoLogBloomCaching(autoLogBloomCachingEnabled)
             .ethstatsUrl(unstableEthstatsOptions.getEthstatsUrl())
             .ethstatsContact(unstableEthstatsOptions.getEthstatsContact())
+            .storageProvider(keyStorageProvider(keyValueStorageName))
             .build();
 
     addShutdownHook(runner);

besu/src/test/java/org/hyperledger/besu/RunnerBuilderTest.java

@@ -41,6 +41,7 @@ import org.hyperledger.besu.ethereum.eth.manager.EthScheduler;
 import org.hyperledger.besu.ethereum.eth.transactions.TransactionPool;
 import org.hyperledger.besu.ethereum.p2p.config.SubProtocolConfiguration;
 import org.hyperledger.besu.ethereum.p2p.peers.EnodeURL;
+import org.hyperledger.besu.ethereum.storage.keyvalue.KeyValueStorageProvider;
 import org.hyperledger.besu.metrics.ObservableMetricsSystem;
 import org.hyperledger.besu.metrics.prometheus.MetricsConfiguration;
 
@@ -119,6 +120,7 @@ public final class RunnerBuilderTest {
             .metricsConfiguration(mock(MetricsConfiguration.class))
             .vertx(vertx)
             .dataDir(dataDir.getRoot().toPath())
+            .storageProvider(mock(KeyValueStorageProvider.class))
             .build();
     runner.start();
 

besu/src/test/java/org/hyperledger/besu/RunnerTest.java

@@ -202,7 +202,8 @@ public final class RunnerTest {
             .p2pListenPort(0)
             .maxPeers(3)
             .metricsSystem(noOpMetricsSystem)
-            .staticNodes(emptySet());
+            .staticNodes(emptySet())
+            .storageProvider(new InMemoryStorageProvider());
 
     Runner runnerBehind = null;
     final Runner runnerAhead =

besu/src/test/java/org/hyperledger/besu/cli/CommandTestAbstract.java

@@ -236,6 +236,7 @@ public abstract class CommandTestAbstract {
     when(mockRunnerBuilder.pidPath(any())).thenReturn(mockRunnerBuilder);
     when(mockRunnerBuilder.ethstatsUrl(anyString())).thenReturn(mockRunnerBuilder);
     when(mockRunnerBuilder.ethstatsContact(anyString())).thenReturn(mockRunnerBuilder);
+    when(mockRunnerBuilder.storageProvider(any())).thenReturn(mockRunnerBuilder);
     when(mockRunnerBuilder.build()).thenReturn(mockRunner);
 
     final Bytes32 keyPairPrvKey =

ethereum/api/src/test/java/org/hyperledger/besu/ethereum/api/jsonrpc/JsonRpcHttpServiceRpcApisTest.java

@@ -29,6 +29,7 @@ import org.hyperledger.besu.ethereum.api.jsonrpc.methods.JsonRpcMethodsFactory;
 import org.hyperledger.besu.ethereum.api.jsonrpc.websocket.WebSocketConfiguration;
 import org.hyperledger.besu.ethereum.api.query.BlockchainQueries;
 import org.hyperledger.besu.ethereum.blockcreation.EthHashMiningCoordinator;
+import org.hyperledger.besu.ethereum.core.InMemoryStorageProvider;
 import org.hyperledger.besu.ethereum.core.PrivacyParameters;
 import org.hyperledger.besu.ethereum.core.Synchronizer;
 import org.hyperledger.besu.ethereum.eth.EthProtocol;
@@ -261,6 +262,7 @@ public class JsonRpcHttpServiceRpcApisTest {
             .vertx(vertx)
             .config(config)
             .metricsSystem(new NoOpMetricsSystem())
+            .storageProvider(new InMemoryStorageProvider())
             .build();
 
     p2pNetwork.start();

ethereum/eth/src/test/java/org/hyperledger/besu/ethereum/eth/transactions/TestNode.java

@@ -30,6 +30,7 @@ import org.hyperledger.besu.ethereum.ProtocolContext;
 import org.hyperledger.besu.ethereum.chain.GenesisState;
 import org.hyperledger.besu.ethereum.chain.MutableBlockchain;
 import org.hyperledger.besu.ethereum.core.BlockHeaderFunctions;
+import org.hyperledger.besu.ethereum.core.InMemoryStorageProvider;
 import org.hyperledger.besu.ethereum.core.Transaction;
 import org.hyperledger.besu.ethereum.core.Wei;
 import org.hyperledger.besu.ethereum.difficulty.fixed.FixedDifficultyProtocolSchedule;
@@ -166,6 +167,7 @@ public class TestNode implements Closeable {
                         .config(networkingConfiguration)
                         .metricsSystem(new NoOpMetricsSystem())
                         .supportedCapabilities(capabilities)
+                        .storageProvider(new InMemoryStorageProvider())
                         .build())
             .metricsSystem(new NoOpMetricsSystem())
             .build();

ethereum/p2p/build.gradle

@@ -51,6 +51,7 @@ dependencies {
 
   annotationProcessor "org.immutables:value"
   implementation "org.immutables:value-annotations"
+  implementation 'tech.pegasys.discovery:discovery'
 
   runtimeOnly 'org.apache.logging.log4j:log4j-core'
 

ethereum/p2p/src/main/java/org/hyperledger/besu/ethereum/p2p/discovery/PeerDiscoveryAgent.java

@@ -16,6 +16,7 @@ package org.hyperledger.besu.ethereum.p2p.discovery;
 
 import static com.google.common.base.Preconditions.checkArgument;
 import static com.google.common.base.Preconditions.checkNotNull;
+import static java.nio.charset.StandardCharsets.UTF_8;
 import static org.apache.tuweni.bytes.Bytes.wrapBuffer;
 
 import org.hyperledger.besu.crypto.NodeKey;
@@ -30,8 +31,12 @@ import org.hyperledger.besu.ethereum.p2p.peers.EnodeURL;
 import org.hyperledger.besu.ethereum.p2p.peers.Peer;
 import org.hyperledger.besu.ethereum.p2p.peers.PeerId;
 import org.hyperledger.besu.ethereum.p2p.permissions.PeerPermissions;
+import org.hyperledger.besu.ethereum.storage.StorageProvider;
+import org.hyperledger.besu.ethereum.storage.keyvalue.KeyValueSegmentIdentifier;
 import org.hyperledger.besu.nat.NatService;
 import org.hyperledger.besu.plugin.services.MetricsSystem;
+import org.hyperledger.besu.plugin.services.storage.KeyValueStorage;
+import org.hyperledger.besu.plugin.services.storage.KeyValueStorageTransaction;
 import org.hyperledger.besu.util.NetworkUtility;
 import org.hyperledger.besu.util.Subscribers;
 
@@ -49,6 +54,12 @@ import com.google.common.net.InetAddresses;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 import org.apache.tuweni.bytes.Bytes;
+import org.apache.tuweni.units.bigints.UInt64;
+import org.ethereum.beacon.discovery.schema.EnrField;
+import org.ethereum.beacon.discovery.schema.IdentitySchema;
+import org.ethereum.beacon.discovery.schema.IdentitySchemaInterpreter;
+import org.ethereum.beacon.discovery.schema.NodeRecord;
+import org.ethereum.beacon.discovery.schema.NodeRecordFactory;
 
 /**
  * The peer discovery agent is the network component that sends and receives peer discovery messages
@@ -56,6 +67,7 @@ import org.apache.tuweni.bytes.Bytes;
  */
 public abstract class PeerDiscoveryAgent {
   private static final Logger LOG = LogManager.getLogger();
+  private static final String SEQ_NO_STORE_KEY = "local-enr-seqno";
 
   // The devp2p specification says only accept packets up to 1280, but some
   // clients ignore that, so we add in a little extra padding.
@@ -83,12 +95,15 @@ public abstract class PeerDiscoveryAgent {
   private boolean isActive = false;
   protected final Subscribers<PeerBondedObserver> peerBondedObservers = Subscribers.create();
 
+  private final StorageProvider storageProvider;
+
   protected PeerDiscoveryAgent(
       final NodeKey nodeKey,
       final DiscoveryConfiguration config,
       final PeerPermissions peerPermissions,
       final NatService natService,
-      final MetricsSystem metricsSystem) {
+      final MetricsSystem metricsSystem,
+      final StorageProvider storageProvider) {
     this.metricsSystem = metricsSystem;
     checkArgument(nodeKey != null, "nodeKey cannot be null");
     checkArgument(config != null, "provided configuration cannot be null");
@@ -104,6 +119,8 @@ public abstract class PeerDiscoveryAgent {
     this.nodeKey = nodeKey;
 
     id = nodeKey.getPublicKey().getEncodedBytes();
+
+    this.storageProvider = storageProvider;
   }
 
   protected abstract TimerUtil createTimer();
@@ -144,6 +161,7 @@ public abstract class PeerDiscoveryAgent {
                 isActive = true;
                 LOG.info("P2P peer discovery agent started and listening on {}", localAddress);
                 startController(ourNode);
+                addLocalNodeRecord(id, advertisedAddress, tcpPort, discoveryPort);
                 return discoveryPort;
               });
     } else {
@@ -152,6 +170,45 @@ public abstract class PeerDiscoveryAgent {
     }
   }
 
+  private void addLocalNodeRecord(
+      final Bytes nodeId,
+      final String advertisedAddress,
+      final Integer tcpPort,
+      final Integer udpPort) {
+    final KeyValueStorage keyValueStorage =
+        storageProvider.getStorageBySegmentIdentifier(KeyValueSegmentIdentifier.BLOCKCHAIN);
+    final NodeRecordFactory nodeRecordFactory = new NodeRecordFactory(IdentitySchemaInterpreter.V4);
+    final Optional<NodeRecord> existingNodeRecord =
+        keyValueStorage
+            .get(Bytes.of(SEQ_NO_STORE_KEY.getBytes(UTF_8)).toArray())
+            .map(Bytes::of)
+            .flatMap(b -> Optional.of(nodeRecordFactory.fromBytes(b)));
+
+    final Bytes addressBytes = Bytes.of(advertisedAddress.getBytes(UTF_8));
+    if (existingNodeRecord.isEmpty()
+        || !existingNodeRecord.get().getNodeId().equals(nodeId)
+        || !addressBytes.equals(existingNodeRecord.get().get(EnrField.IP_V4))
+        || !tcpPort.equals(existingNodeRecord.get().get(EnrField.TCP))
+        || !udpPort.equals(existingNodeRecord.get().get(EnrField.UDP))) {
+      final UInt64 sequenceNumber =
+          existingNodeRecord.map(NodeRecord::getSeq).orElse(UInt64.ZERO).add(1);
+      final NodeRecord nodeRecord =
+          nodeRecordFactory.createFromValues(
+              sequenceNumber,
+              new EnrField(EnrField.ID, IdentitySchema.V4),
+              new EnrField(EnrField.PKEY_SECP256K1, nodeId),
+              new EnrField(EnrField.IP_V4, addressBytes),
+              new EnrField(EnrField.TCP, tcpPort),
+              new EnrField(EnrField.UDP, udpPort));
+
+      final KeyValueStorageTransaction keyValueStorageTransaction =
+          keyValueStorage.startTransaction();
+      keyValueStorageTransaction.put(
+          Bytes.wrap(SEQ_NO_STORE_KEY.getBytes(UTF_8)).toArray(), nodeRecord.serialize().toArray());
+      keyValueStorageTransaction.commit();
+    }
+  }
+
   public void addPeerRequirement(final PeerRequirement peerRequirement) {
     this.peerRequirements.add(peerRequirement);
   }

ethereum/p2p/src/main/java/org/hyperledger/besu/ethereum/p2p/discovery/VertxPeerDiscoveryAgent.java

@@ -24,6 +24,7 @@ import org.hyperledger.besu.ethereum.p2p.discovery.internal.PeerDiscoveryControl
 import org.hyperledger.besu.ethereum.p2p.discovery.internal.TimerUtil;
 import org.hyperledger.besu.ethereum.p2p.discovery.internal.VertxTimerUtil;
 import org.hyperledger.besu.ethereum.p2p.permissions.PeerPermissions;
+import org.hyperledger.besu.ethereum.storage.StorageProvider;
 import org.hyperledger.besu.metrics.BesuMetricCategory;
 import org.hyperledger.besu.nat.NatService;
 import org.hyperledger.besu.plugin.services.MetricsSystem;
@@ -62,8 +63,9 @@ public class VertxPeerDiscoveryAgent extends PeerDiscoveryAgent {
       final DiscoveryConfiguration config,
       final PeerPermissions peerPermissions,
       final NatService natService,
-      final MetricsSystem metricsSystem) {
+      final MetricsSystem metricsSystem,
-    super(nodeKey, config, peerPermissions, natService, metricsSystem);
+      final StorageProvider storageProvider) {
+    super(nodeKey, config, peerPermissions, natService, metricsSystem, storageProvider);
     checkArgument(vertx != null, "vertx instance cannot be null");
     this.vertx = vertx;
 

ethereum/p2p/src/main/java/org/hyperledger/besu/ethereum/p2p/network/DefaultP2PNetwork.java

@@ -41,6 +41,7 @@ import org.hyperledger.besu.ethereum.p2p.rlpx.RlpxAgent;
 import org.hyperledger.besu.ethereum.p2p.rlpx.connections.PeerConnection;
 import org.hyperledger.besu.ethereum.p2p.rlpx.wire.Capability;
 import org.hyperledger.besu.ethereum.p2p.rlpx.wire.messages.DisconnectMessage.DisconnectReason;
+import org.hyperledger.besu.ethereum.storage.StorageProvider;
 import org.hyperledger.besu.nat.NatMethod;
 import org.hyperledger.besu.nat.NatService;
 import org.hyperledger.besu.nat.core.domain.NatServiceType;
@@ -439,6 +440,7 @@ public class DefaultP2PNetwork implements P2PNetwork {
     private boolean randomPeerPriority;
 
     private MetricsSystem metricsSystem;
+    private StorageProvider storageProvider;
 
     public P2PNetwork build() {
       validate();
@@ -477,13 +479,20 @@ public class DefaultP2PNetwork implements P2PNetwork {
           supportedCapabilities != null && supportedCapabilities.size() > 0,
           "Supported capabilities must be set and non-empty.");
       checkState(metricsSystem != null, "MetricsSystem must be set.");
+      checkState(storageProvider != null, "StorageProvider must be set.");
       checkState(peerDiscoveryAgent != null || vertx != null, "Vertx must be set.");
     }
 
     private PeerDiscoveryAgent createDiscoveryAgent() {
 
       return new VertxPeerDiscoveryAgent(
-          vertx, nodeKey, config.getDiscovery(), peerPermissions, natService, metricsSystem);
+          vertx,
+          nodeKey,
+          config.getDiscovery(),
+          peerPermissions,
+          natService,
+          metricsSystem,
+          storageProvider);
     }
 
     private RlpxAgent createRlpxAgent(
@@ -568,5 +577,11 @@ public class DefaultP2PNetwork implements P2PNetwork {
       this.maintainedPeers = maintainedPeers;
       return this;
     }
+
+    public Builder storageProvider(final StorageProvider storageProvider) {
+      checkNotNull(storageProvider);
+      this.storageProvider = storageProvider;
+      return this;
+    }
   }
 }

ethereum/p2p/src/test/java/org/hyperledger/besu/ethereum/p2p/discovery/internal/MockPeerDiscoveryAgent.java

@@ -15,6 +15,7 @@
 package org.hyperledger.besu.ethereum.p2p.discovery.internal;
 
 import org.hyperledger.besu.crypto.NodeKey;
+import org.hyperledger.besu.ethereum.core.InMemoryStorageProvider;
 import org.hyperledger.besu.ethereum.p2p.config.DiscoveryConfiguration;
 import org.hyperledger.besu.ethereum.p2p.discovery.DiscoveryPeer;
 import org.hyperledger.besu.ethereum.p2p.discovery.PeerDiscoveryAgent;
@@ -49,7 +50,13 @@ public class MockPeerDiscoveryAgent extends PeerDiscoveryAgent {
       final PeerPermissions peerPermissions,
       final Map<Bytes, MockPeerDiscoveryAgent> agentNetwork,
       final NatService natService) {
-    super(nodeKey, config, peerPermissions, natService, new NoOpMetricsSystem());
+    super(
+        nodeKey,
+        config,
+        peerPermissions,
+        natService,
+        new NoOpMetricsSystem(),
+        new InMemoryStorageProvider());
     this.agentNetwork = agentNetwork;
   }
 

ethereum/p2p/src/test/java/org/hyperledger/besu/ethereum/p2p/network/DefaultP2PNetworkTest.java

@@ -26,6 +26,7 @@ import static org.mockito.Mockito.when;
 
 import org.hyperledger.besu.crypto.NodeKey;
 import org.hyperledger.besu.crypto.NodeKeyUtils;
+import org.hyperledger.besu.ethereum.core.InMemoryStorageProvider;
 import org.hyperledger.besu.ethereum.p2p.config.DiscoveryConfiguration;
 import org.hyperledger.besu.ethereum.p2p.config.NetworkingConfiguration;
 import org.hyperledger.besu.ethereum.p2p.config.RlpxConfiguration;
@@ -336,6 +337,7 @@ public final class DefaultP2PNetworkTest {
         .nodeKey(nodeKey)
         .maintainedPeers(maintainedPeers)
         .metricsSystem(new NoOpMetricsSystem())
-        .supportedCapabilities(Capability.create("eth", 63));
+        .supportedCapabilities(Capability.create("eth", 63))
+        .storageProvider(new InMemoryStorageProvider());
   }
 }

ethereum/p2p/src/test/java/org/hyperledger/besu/ethereum/p2p/network/NetworkingServiceLifecycleTest.java

@@ -20,6 +20,7 @@ import static org.hyperledger.besu.ethereum.p2p.NetworkingTestHelper.configWithR
 
 import org.hyperledger.besu.crypto.NodeKey;
 import org.hyperledger.besu.crypto.NodeKeyUtils;
+import org.hyperledger.besu.ethereum.core.InMemoryStorageProvider;
 import org.hyperledger.besu.ethereum.p2p.config.DiscoveryConfiguration;
 import org.hyperledger.besu.ethereum.p2p.config.NetworkingConfiguration;
 import org.hyperledger.besu.ethereum.p2p.discovery.PeerDiscoveryServiceException;
@@ -158,6 +159,7 @@ public class NetworkingServiceLifecycleTest {
         .nodeKey(nodeKey)
         .config(config)
         .metricsSystem(new NoOpMetricsSystem())
-        .supportedCapabilities(Arrays.asList(Capability.create("eth", 63)));
+        .supportedCapabilities(Arrays.asList(Capability.create("eth", 63)))
+        .storageProvider(new InMemoryStorageProvider());
   }
 }

ethereum/p2p/src/test/java/org/hyperledger/besu/ethereum/p2p/network/P2PNetworkTest.java

@@ -23,6 +23,7 @@ import static org.mockito.Mockito.when;
 
 import org.hyperledger.besu.crypto.NodeKey;
 import org.hyperledger.besu.crypto.NodeKeyUtils;
+import org.hyperledger.besu.ethereum.core.InMemoryStorageProvider;
 import org.hyperledger.besu.ethereum.p2p.config.DiscoveryConfiguration;
 import org.hyperledger.besu.ethereum.p2p.config.NetworkingConfiguration;
 import org.hyperledger.besu.ethereum.p2p.config.RlpxConfiguration;
@@ -333,6 +334,7 @@ public class P2PNetworkTest {
         .config(config)
         .nodeKey(NodeKeyUtils.generate())
         .metricsSystem(new NoOpMetricsSystem())
-        .supportedCapabilities(Arrays.asList(Capability.create("eth", 63)));
+        .supportedCapabilities(Arrays.asList(Capability.create("eth", 63)))
+        .storageProvider(new InMemoryStorageProvider());
   }
 }

gradle/versions.gradle

@@ -144,5 +144,7 @@ dependencyManagement {
     dependency 'tech.pegasys.ethsigner.internal:core:0.4.0'
     dependency 'tech.pegasys.ethsigner.internal:file-based:0.4.0'
     dependency 'tech.pegasys.ethsigner.internal:signing-api:0.4.0'
+
+    dependency 'tech.pegasys.discovery:discovery:0.4.1'
   }
 }