[NC-1273] Consider peer count insufficient until minimum peers for fast sync are connected (#629)

Makes Synchronizer responsible for deciding if it has enough peers or not rather than ProtocolManager.

Signed-off-by: Adrian Sutton <adrian.sutton@consensys.net>

consensus/ibft/src/main/java/tech/pegasys/pantheon/consensus/ibft/protocol/IbftProtocolManager.java

@@ -95,9 +95,4 @@ public class IbftProtocolManager implements ProtocolManager {
       final boolean initiatedByPeer) {
     peers.remove(peerConnection);
   }
-
-  @Override
-  public boolean hasSufficientPeers() {
-    return true;
-  }
 }

ethereum/core/src/main/java/tech/pegasys/pantheon/ethereum/core/Synchronizer.java

@@ -24,4 +24,6 @@ public interface Synchronizer {
    *     empty
    */
   Optional<SyncStatus> getSyncStatus();
+
+  boolean hasSufficientPeers();
 }

ethereum/eth/src/main/java/tech/pegasys/pantheon/ethereum/eth/manager/EthProtocolManager.java

@@ -237,11 +237,6 @@ public class EthProtocolManager implements ProtocolManager, MinedBlockObserver {
     }
   }
 
-  @Override
-  public boolean hasSufficientPeers() {
-    return ethPeers.availablePeerCount() > 0;
-  }
-
   private void handleStatusMessage(final EthPeer peer, final MessageData data) {
     final StatusMessage status = StatusMessage.readFrom(data);
     try {

ethereum/eth/src/main/java/tech/pegasys/pantheon/ethereum/eth/sync/DefaultSynchronizer.java

@@ -37,6 +37,8 @@ public class DefaultSynchronizer<C> implements Synchronizer {
 
   private static final Logger LOG = LogManager.getLogger();
 
+  private final SynchronizerConfiguration syncConfig;
+  private final EthContext ethContext;
   private final SyncState syncState;
   private final AtomicBoolean started = new AtomicBoolean(false);
   private final BlockPropagationManager<C> blockPropagationManager;
@@ -50,6 +52,8 @@ public class DefaultSynchronizer<C> implements Synchronizer {
       final EthContext ethContext,
       final SyncState syncState,
       final LabelledMetric<OperationTimer> ethTasksTimer) {
+    this.syncConfig = syncConfig;
+    this.ethContext = ethContext;
     this.syncState = syncState;
     this.blockPropagationManager =
         new BlockPropagationManager<>(
@@ -121,4 +125,11 @@ public class DefaultSynchronizer<C> implements Synchronizer {
     }
     return Optional.of(syncState.syncStatus());
   }
+
+  @Override
+  public boolean hasSufficientPeers() {
+    final int requiredPeerCount =
+        fastSyncDownloader.isPresent() ? syncConfig.getFastSyncMinimumPeerCount() : 1;
+    return ethContext.getEthPeers().availablePeerCount() >= requiredPeerCount;
+  }
 }

ethereum/eth/src/main/java/tech/pegasys/pantheon/ethereum/eth/sync/SynchronizerConfiguration.java

@@ -32,7 +32,7 @@ public class SynchronizerConfiguration {
   public static int DEFAULT_PIVOT_DISTANCE_FROM_HEAD = 500;
   public static float DEFAULT_FULL_VALIDATION_RATE = .1f;
   public static int DEFAULT_FAST_SYNC_MINIMUM_PEERS = 5;
-  private static final Duration DEFAULT_FAST_SYNC_MAXIMUM_PEER_WAIT_TIME = Duration.ofSeconds(3);
+  private static final Duration DEFAULT_FAST_SYNC_MAXIMUM_PEER_WAIT_TIME = Duration.ofMinutes(3);
 
   // Fast sync config
   private final int fastSyncPivotDistance;

ethereum/eth/src/test/java/tech/pegasys/pantheon/ethereum/eth/transactions/TestNode.java

@@ -115,7 +115,7 @@ public class TestNode implements Closeable {
                         this.kp,
                         networkingConfiguration,
                         capabilities,
-                        ethProtocolManager,
+                        () -> true,
                         new PeerBlacklist(),
                         new NoOpMetricsSystem(),
                         new NodeWhitelistController(PermissioningConfiguration.createDefault())))

ethereum/p2p/src/main/java/tech/pegasys/pantheon/ethereum/p2p/api/ProtocolManager.java

@@ -12,14 +12,13 @@
  */
 package tech.pegasys.pantheon.ethereum.p2p.api;
 
-import tech.pegasys.pantheon.ethereum.p2p.discovery.internal.PeerRequirement;
 import tech.pegasys.pantheon.ethereum.p2p.wire.Capability;
 import tech.pegasys.pantheon.ethereum.p2p.wire.messages.DisconnectMessage.DisconnectReason;
 
 import java.util.List;
 
 /** Represents an object responsible for managing a wire subprotocol. */
-public interface ProtocolManager extends AutoCloseable, PeerRequirement {
+public interface ProtocolManager extends AutoCloseable {
 
   String getSupportedProtocol();
 

ethereum/p2p/src/main/java/tech/pegasys/pantheon/ethereum/p2p/discovery/internal/PeerRequirement.java

@@ -12,14 +12,8 @@
  */
 package tech.pegasys.pantheon.ethereum.p2p.discovery.internal;
 
-import java.util.Collection;
-
 @FunctionalInterface
 public interface PeerRequirement {
 
   boolean hasSufficientPeers();
-
-  static PeerRequirement aggregateOf(final Collection<? extends PeerRequirement> peers) {
-    return () -> peers.stream().allMatch(PeerRequirement::hasSufficientPeers);
-  }
 }

pantheon/src/main/java/tech/pegasys/pantheon/RunnerBuilder.java

@@ -46,7 +46,6 @@ import tech.pegasys.pantheon.ethereum.p2p.config.DiscoveryConfiguration;
 import tech.pegasys.pantheon.ethereum.p2p.config.NetworkingConfiguration;
 import tech.pegasys.pantheon.ethereum.p2p.config.RlpxConfiguration;
 import tech.pegasys.pantheon.ethereum.p2p.config.SubProtocolConfiguration;
-import tech.pegasys.pantheon.ethereum.p2p.discovery.internal.PeerRequirement;
 import tech.pegasys.pantheon.ethereum.p2p.netty.NettyP2PNetwork;
 import tech.pegasys.pantheon.ethereum.p2p.peers.PeerBlacklist;
 import tech.pegasys.pantheon.ethereum.p2p.permissioning.NodeWhitelistController;
@@ -213,9 +212,11 @@ public class RunnerBuilder {
         new PeerBlacklist(
             bannedNodeIds.stream().map(BytesValue::fromHexString).collect(Collectors.toSet()));
 
-    NodeWhitelistController nodeWhitelistController =
+    final NodeWhitelistController nodeWhitelistController =
         new NodeWhitelistController(permissioningConfiguration);
 
+    final Synchronizer synchronizer = pantheonController.getSynchronizer();
+
     final NetworkRunner networkRunner =
         NetworkRunner.builder()
             .protocolManagers(protocolManagers)
@@ -228,7 +229,7 @@ public class RunnerBuilder {
                             keyPair,
                             networkConfig,
                             caps,
-                            PeerRequirement.aggregateOf(protocolManagers),
+                            synchronizer::hasSufficientPeers,
                             peerBlacklist,
                             metricsSystem,
                             nodeWhitelistController)
@@ -236,7 +237,6 @@ public class RunnerBuilder {
             .metricsSystem(metricsSystem)
             .build();
 
-    final Synchronizer synchronizer = pantheonController.getSynchronizer();
     final TransactionPool transactionPool = pantheonController.getTransactionPool();
     final MiningCoordinator miningCoordinator = pantheonController.getMiningCoordinator();