applies merge filtering to peers after 2 finalized (#4124)

* applies merge filtering to peers after 2 finalized
* fixes logical error in handling status messages

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

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

@@ -104,11 +104,13 @@ public class MergeBesuControllerBuilder extends BesuControllerBuilder {
     ethPeers.setBestChainComparator(mergeBestPeerComparator);
     mergeContext.observeNewIsPostMergeState(mergeBestPeerComparator);
 
-    if (mergePeerFilter.isPresent()) {
+    Optional<MergePeerFilter> filterToUse = Optional.of(new MergePeerFilter());
 
-      mergeContext.observeNewIsPostMergeState(mergePeerFilter.get());
-      mergeContext.addNewForkchoiceMessageListener(mergePeerFilter.get());
+    if (mergePeerFilter.isPresent()) {
+      filterToUse = mergePeerFilter;
     }
+    mergeContext.observeNewIsPostMergeState(filterToUse.get());
+    mergeContext.addNewForkchoiceMessageListener(filterToUse.get());
 
     EthProtocolManager ethProtocolManager =
         super.createEthProtocolManager(
@@ -121,7 +123,7 @@ public class MergeBesuControllerBuilder extends BesuControllerBuilder {
             ethMessages,
             scheduler,
             peerValidators,
-            mergePeerFilter);
+            filterToUse);
 
     return ethProtocolManager;
   }

ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/manager/EthProtocolManager.java

@@ -376,11 +376,9 @@ public class EthProtocolManager implements ProtocolManager, MinedBlockObserver {
             networkId,
             status.genesisHash());
         peer.disconnect(DisconnectReason.SUBPROTOCOL_TRIGGERED);
-      } else if (mergePeerFilter.isPresent()) {
-        final boolean disconnected = mergePeerFilter.get().disconnectIfPoW(status, peer);
-        if (disconnected) {
-          handleDisconnect(peer.getConnection(), DisconnectReason.SUBPROTOCOL_TRIGGERED, false);
-        }
+      } else if (mergePeerFilter.isPresent()
+          && mergePeerFilter.get().disconnectIfPoW(status, peer)) {
+        handleDisconnect(peer.getConnection(), DisconnectReason.SUBPROTOCOL_TRIGGERED, false);
       } else {
         LOG.debug("Received status message from {}: {}", peer, status);
         peer.registerStatusReceived(