3890 disconnect pow net (hyperledger#3944)

* rearranged some things so EthProtocolManager can get notified of ttd
* adds some needed state and locking
* listens to MergeContext
* tests for preventing pow peers from joining
* disconnecting block senders after ttd
* wait till finalizing 2 blocks before locking out pow nodes
* hacking around clique networks, will need to be fixed still for Goerli

Signed-off-by: Justin Florentine <justin+github@florentine.us>

besu/src/main/java/org/hyperledger/besu/controller/BesuControllerBuilder.java

@@ -98,6 +98,7 @@
 import java.util.OptionalLong;
 import java.util.function.Supplier;
 
+import org.jetbrains.annotations.NotNull;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -359,11 +360,15 @@ public BesuController build() {
           Optional.of(
               ImmutableCheckpoint.builder()
                   .blockHash(
-                      Hash.fromHexString(configOptions.getCheckpointOptions().getHash().get()))
+                      Hash.fromHexString(
+                          configOptions.getCheckpointOptions().getHash().get())) // NOSONAR
                   .blockNumber(configOptions.getCheckpointOptions().getNumber().getAsLong())
                   .totalDifficulty(
                       Difficulty.fromHexString(
-                          configOptions.getCheckpointOptions().getTotalDifficulty().get()))
+                          configOptions
+                              .getCheckpointOptions()
+                              .getTotalDifficulty()
+                              .get())) // NOSONAR
                   .build());
     }
 
@@ -403,19 +408,14 @@ public BesuController build() {
     final PivotBlockSelector pivotBlockSelector = createPivotSelector(protocolContext);
 
     final Synchronizer synchronizer =
-        new DefaultSynchronizer(
-            syncConfig,
+        createSynchronizer(
             protocolSchedule,
-            protocolContext,
             worldStateStorage,
-            ethProtocolManager.getBlockBroadcaster(),
+            protocolContext,
             maybePruner,
             ethContext,
             syncState,
-            dataDirectory,
-            clock,
-            metricsSystem,
-            getFullSyncTerminationCondition(protocolContext.getBlockchain()),
+            ethProtocolManager,
             pivotBlockSelector);
 
     final MiningCoordinator miningCoordinator =
@@ -432,7 +432,6 @@ public BesuController build() {
 
     final SubProtocolConfiguration subProtocolConfiguration =
         createSubProtocolConfiguration(ethProtocolManager, maybeSnapProtocolManager);
-    ;
 
     final JsonRpcMethods additionalJsonRpcMethodFactory =
         createAdditionalJsonRpcMethodFactory(protocolContext);
@@ -461,6 +460,32 @@ public BesuController build() {
         additionalPluginServices);
   }
 
+  @NotNull
+  protected DefaultSynchronizer createSynchronizer(
+      final ProtocolSchedule protocolSchedule,
+      final WorldStateStorage worldStateStorage,
+      final ProtocolContext protocolContext,
+      final Optional<Pruner> maybePruner,
+      final EthContext ethContext,
+      final SyncState syncState,
+      final EthProtocolManager ethProtocolManager,
+      final PivotBlockSelector pivotBlockSelector) {
+    return new DefaultSynchronizer(
+        syncConfig,
+        protocolSchedule,
+        protocolContext,
+        worldStateStorage,
+        ethProtocolManager.getBlockBroadcaster(),
+        maybePruner,
+        ethContext,
+        syncState,
+        dataDirectory,
+        clock,
+        metricsSystem,
+        getFullSyncTerminationCondition(protocolContext.getBlockchain()),
+        pivotBlockSelector);
+  }
+
   private PivotBlockSelector createPivotSelector(final ProtocolContext protocolContext) {
 
     final PivotSelectorFromPeers pivotSelectorFromPeers = new PivotSelectorFromPeers(syncConfig);
@@ -507,7 +532,7 @@ protected SyncTerminationCondition getFullSyncTerminationCondition(final Blockch
   protected void prepForBuild() {}
 
   protected JsonRpcMethods createAdditionalJsonRpcMethodFactory(
-      final ProtocolContext protocolContext) {
+      final ProtocolContext protocolContext) { // NOSONAR
     return apis -> Collections.emptyMap();
   }
 
@@ -517,9 +542,8 @@ protected SubProtocolConfiguration createSubProtocolConfiguration(
     final SubProtocolConfiguration subProtocolConfiguration =
         new SubProtocolConfiguration().withSubProtocol(EthProtocol.get(), ethProtocolManager);
     maybeSnapProtocolManager.ifPresent(
-        snapProtocolManager -> {
-          subProtocolConfiguration.withSubProtocol(SnapProtocol.get(), snapProtocolManager);
-        });
+        snapProtocolManager ->
+            subProtocolConfiguration.withSubProtocol(SnapProtocol.get(), snapProtocolManager));
     return subProtocolConfiguration;
   }
 

besu/src/main/java/org/hyperledger/besu/controller/TransitionBesuControllerBuilder.java

@@ -15,6 +15,7 @@
 package org.hyperledger.besu.controller;
 
 import org.hyperledger.besu.config.GenesisConfigFile;
+import org.hyperledger.besu.consensus.merge.MergeContext;
 import org.hyperledger.besu.consensus.merge.PostMergeContext;
 import org.hyperledger.besu.consensus.merge.TransitionBackwardSyncContext;
 import org.hyperledger.besu.consensus.merge.TransitionContext;
@@ -31,7 +32,14 @@
 import org.hyperledger.besu.ethereum.core.MiningParameters;
 import org.hyperledger.besu.ethereum.core.PrivacyParameters;
 import org.hyperledger.besu.ethereum.eth.EthProtocolConfiguration;
+import org.hyperledger.besu.ethereum.eth.manager.EthContext;
+import org.hyperledger.besu.ethereum.eth.manager.EthMessages;
+import org.hyperledger.besu.ethereum.eth.manager.EthPeers;
 import org.hyperledger.besu.ethereum.eth.manager.EthProtocolManager;
+import org.hyperledger.besu.ethereum.eth.manager.EthScheduler;
+import org.hyperledger.besu.ethereum.eth.peervalidation.PeerValidator;
+import org.hyperledger.besu.ethereum.eth.sync.DefaultSynchronizer;
+import org.hyperledger.besu.ethereum.eth.sync.PivotBlockSelector;
 import org.hyperledger.besu.ethereum.eth.sync.SynchronizerConfiguration;
 import org.hyperledger.besu.ethereum.eth.sync.backwardsync.BackwardSyncContext;
 import org.hyperledger.besu.ethereum.eth.sync.state.SyncState;
@@ -40,8 +48,10 @@
 import org.hyperledger.besu.ethereum.mainnet.ProtocolSchedule;
 import org.hyperledger.besu.ethereum.storage.StorageProvider;
 import org.hyperledger.besu.ethereum.worldstate.DataStorageConfiguration;
+import org.hyperledger.besu.ethereum.worldstate.Pruner;
 import org.hyperledger.besu.ethereum.worldstate.PrunerConfiguration;
 import org.hyperledger.besu.ethereum.worldstate.WorldStateArchive;
+import org.hyperledger.besu.ethereum.worldstate.WorldStateStorage;
 import org.hyperledger.besu.evm.internal.EvmConfiguration;
 import org.hyperledger.besu.metrics.ObservableMetricsSystem;
 import org.hyperledger.besu.plugin.services.permissioning.NodeMessagePermissioningProvider;
@@ -145,13 +155,79 @@ protected PluginServiceFactory createAdditionalPluginServices(
     return new NoopPluginServiceFactory();
   }
 
+  @Override
+  protected DefaultSynchronizer createSynchronizer(
+      final ProtocolSchedule protocolSchedule,
+      final WorldStateStorage worldStateStorage,
+      final ProtocolContext protocolContext,
+      final Optional<Pruner> maybePruner,
+      final EthContext ethContext,
+      final SyncState syncState,
+      final EthProtocolManager ethProtocolManager,
+      final PivotBlockSelector pivotBlockSelector) {
+
+    DefaultSynchronizer sync =
+        super.createSynchronizer(
+            protocolSchedule,
+            worldStateStorage,
+            protocolContext,
+            maybePruner,
+            ethContext,
+            syncState,
+            ethProtocolManager,
+            pivotBlockSelector);
+
+    ConsensusContext cc = protocolContext.getConsensusContext(ConsensusContext.class);
+    if (cc instanceof MergeContext) {
+      protocolContext.getConsensusContext(MergeContext.class).addNewForkchoiceMessageListener(sync);
+    }
+    return sync;
+  }
+
+  @Override
+  protected EthProtocolManager createEthProtocolManager(
+      final ProtocolContext protocolContext,
+      final boolean fastSyncEnabled,
+      final TransactionPool transactionPool,
+      final EthProtocolConfiguration ethereumWireProtocolConfiguration,
+      final EthPeers ethPeers,
+      final EthContext ethContext,
+      final EthMessages ethMessages,
+      final EthScheduler scheduler,
+      final List<PeerValidator> peerValidators) {
+
+    EthProtocolManager ethProtocolManager =
+        super.createEthProtocolManager(
+            protocolContext,
+            fastSyncEnabled,
+            transactionPool,
+            ethereumWireProtocolConfiguration,
+            ethPeers,
+            ethContext,
+            ethMessages,
+            scheduler,
+            peerValidators);
+
+    ConsensusContext cc = protocolContext.getConsensusContext(ConsensusContext.class);
+    if (cc instanceof MergeContext) {
+      protocolContext
+          .getConsensusContext(MergeContext.class)
+          .observeNewIsPostMergeState(ethProtocolManager);
+      protocolContext
+          .getConsensusContext(MergeContext.class)
+          .addNewForkchoiceMessageListener(ethProtocolManager);
+    }
+
+    return ethProtocolManager;
+  }
+
   private void initTransitionWatcher(
       final ProtocolContext protocolContext, final TransitionCoordinator composedCoordinator) {
 
     PostMergeContext postMergeContext = protocolContext.getConsensusContext(PostMergeContext.class);
     postMergeContext.observeNewIsPostMergeState(
-        newIsPostMergeState -> {
-          if (newIsPostMergeState) {
+        (isPoS, difficultyStoppedAt) -> {
+          if (isPoS) {
             // if we transitioned to post-merge, stop and disable any mining
             composedCoordinator.getPreMergeObject().disable();
             composedCoordinator.getPreMergeObject().stop();

consensus/merge/src/main/java/org/hyperledger/besu/consensus/merge/FinalizedBlockHashSupplier.java

@@ -14,7 +14,6 @@
  */
 package org.hyperledger.besu.consensus.merge;
 
-import org.hyperledger.besu.consensus.merge.MergeContext.NewForkchoiceMessageListener;
 import org.hyperledger.besu.datatypes.Hash;
 
 import java.util.Optional;

consensus/merge/src/main/java/org/hyperledger/besu/consensus/merge/MergeContext.java

@@ -67,15 +67,4 @@ void fireNewUnverifiedForkchoiceMessageEvent(
       final Hash headBlockHash,
       final Optional<Hash> maybeFinalizedBlockHash,
       final Hash safeBlockHash);
-
-  interface NewMergeStateCallback {
-    void onNewIsPostMergeState(final boolean newIsPostMergeState);
-  }
-
-  interface NewForkchoiceMessageListener {
-    void onNewForkchoiceMessage(
-        final Hash headBlockHash,
-        final Optional<Hash> maybeFinalizedBlockHash,
-        final Hash safeBlockHash);
-  }
 }

consensus/merge/src/main/java/org/hyperledger/besu/consensus/merge/PostMergeContext.java

@@ -83,13 +83,13 @@ public PostMergeContext setTerminalTotalDifficulty(final Difficulty newTerminalT
   }
 
   @Override
-  public void setIsPostMerge(final Difficulty totalDifficulty) {
+  public void setIsPostMerge(final Difficulty difficultyStoppedAt) {
     if (isPostMerge.get().orElse(Boolean.FALSE) && lastFinalized.get() != null) {
       // if we have finalized, we never switch back to a pre-merge once we have transitioned
       // post-TTD.
       return;
     }
-    final boolean newState = terminalTotalDifficulty.get().lessOrEqualThan(totalDifficulty);
+    final boolean newState = terminalTotalDifficulty.get().lessOrEqualThan(difficultyStoppedAt);
     final Optional<Boolean> oldState = isPostMerge.getAndSet(Optional.of(newState));
 
     // if we are past TTD, set it:
@@ -99,7 +99,9 @@ public void setIsPostMerge(final Difficulty totalDifficulty) {
 
     if (oldState.isEmpty() || oldState.get() != newState) {
       newMergeStateCallbackSubscribers.forEach(
-          newMergeStateCallback -> newMergeStateCallback.onNewIsPostMergeState(newState));
+          newMergeStateCallback ->
+              newMergeStateCallback.onCrossingMergeBoundary(
+                  newState, Optional.of(difficultyStoppedAt)));
     }
   }
 

consensus/merge/src/test/java/org/hyperledger/besu/consensus/merge/PostMergeContextTest.java

@@ -22,7 +22,6 @@
 import static org.mockito.Mockito.verifyNoInteractions;
 import static org.mockito.Mockito.when;
 
-import org.hyperledger.besu.consensus.merge.MergeContext.NewMergeStateCallback;
 import org.hyperledger.besu.consensus.merge.blockcreation.PayloadIdentifier;
 import org.hyperledger.besu.ethereum.core.Block;
 import org.hyperledger.besu.ethereum.core.BlockHeader;
@@ -173,8 +172,9 @@ private static class MergeStateChangeCollector implements NewMergeStateCallback
     final List<Boolean> stateChanges = new ArrayList<>();
 
     @Override
-    public void onNewIsPostMergeState(final boolean newIsPostMergeState) {
-      stateChanges.add(newIsPostMergeState);
+    public void onCrossingMergeBoundary(
+        final boolean isPoS, final Optional<Difficulty> difficultyStoppedAt) {
+      stateChanges.add(isPoS);
     }
 
     public void reset() {

ethereum/eth/src/main/java/org/hyperledger/besu/consensus/merge/NewForkchoiceMessageListener.java

@@ -0,0 +1,28 @@
+/*
+ * Copyright Hyperledger Besu Contributors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
+ *  the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
+ * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations under the License.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ */
+
+package org.hyperledger.besu.consensus.merge;
+
+import org.hyperledger.besu.datatypes.Hash;
+
+import java.util.Optional;
+
+public interface NewForkchoiceMessageListener {
+  void onNewForkchoiceMessage(
+      final Hash headBlockHash,
+      final Optional<Hash> maybeFinalizedBlockHash,
+      final Hash safeBlockHash);
+}

ethereum/eth/src/main/java/org/hyperledger/besu/consensus/merge/NewMergeStateCallback.java

@@ -0,0 +1,26 @@
+/*
+ * Copyright Hyperledger Besu Contributors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
+ *  the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
+ * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations under the License.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ */
+
+package org.hyperledger.besu.consensus.merge;
+
+import org.hyperledger.besu.ethereum.core.Difficulty;
+
+import java.util.Optional;
+
+public interface NewMergeStateCallback {
+
+  void onCrossingMergeBoundary(final boolean isPoS, final Optional<Difficulty> difficultyStoppedAt);
+}