Automatically restrict trailing peers while syncing (#1167)

The 25% of maximum peers is fairly arbitrary but is deliberately above zero to ensure we contribute something to the overall network even while catching up.

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

@@ -95,7 +95,7 @@ public DefaultSynchronizer(
   private TrailingPeerRequirements calculateTrailingPeerRequirements() {
     return fastSynchronizer
         .flatMap(FastSynchronizer::calculateTrailingPeerRequirements)
-        .orElse(TrailingPeerRequirements.UNRESTRICTED);
+        .orElseGet(fullSyncDownloader::calculateTrailingPeerRequirements);
   }
 
   @Override

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

@@ -56,6 +56,7 @@ public class SynchronizerConfiguration {
   private final int downloaderParallelism;
   private final int transactionsParallelism;
   private final int computationParallelism;
+  private final int maxTrailingPeers;
 
   private SynchronizerConfiguration(
       final int fastSyncPivotDistance,
@@ -75,7 +76,8 @@ private SynchronizerConfiguration(
       final int downloaderChainSegmentSize,
       final int downloaderParallelism,
       final int transactionsParallelism,
-      final int computationParallelism) {
+      final int computationParallelism,
+      final int maxTrailingPeers) {
     this.fastSyncPivotDistance = fastSyncPivotDistance;
     this.fastSyncFullValidationRate = fastSyncFullValidationRate;
     this.fastSyncMinimumPeerCount = fastSyncMinimumPeerCount;
@@ -94,6 +96,7 @@ private SynchronizerConfiguration(
     this.downloaderParallelism = downloaderParallelism;
     this.transactionsParallelism = transactionsParallelism;
     this.computationParallelism = computationParallelism;
+    this.maxTrailingPeers = maxTrailingPeers;
   }
 
   public static Builder builder() {
@@ -196,6 +199,10 @@ public int getWorldStateMaxRequestsWithoutProgress() {
     return worldStateMaxRequestsWithoutProgress;
   }
 
+  public int getMaxTrailingPeers() {
+    return maxTrailingPeers;
+  }
+
   public static class Builder {
     private SyncMode syncMode = SyncMode.FULL;
     private Range<Long> blockPropagationRange = Range.closed(-10L, 30L);
@@ -216,6 +223,7 @@ public static class Builder {
     private int worldStateMaxRequestsWithoutProgress =
         DEFAULT_WORLD_STATE_MAX_REQUESTS_WITHOUT_PROGRESS;
     private Duration fastSyncMaximumPeerWaitTime = DEFAULT_FAST_SYNC_MAXIMUM_PEER_WAIT_TIME;
+    private int maxTrailingPeers = Integer.MAX_VALUE;
 
     public Builder fastSyncPivotDistance(final int distance) {
       fastSyncPivotDistance = distance;
@@ -313,6 +321,11 @@ public Builder fastSyncMaximumPeerWaitTime(final Duration fastSyncMaximumPeerWai
       return this;
     }
 
+    public Builder maxTrailingPeers(final int maxTailingPeers) {
+      this.maxTrailingPeers = maxTailingPeers;
+      return this;
+    }
+
     public SynchronizerConfiguration build() {
       return new SynchronizerConfiguration(
           fastSyncPivotDistance,
@@ -332,7 +345,8 @@ public SynchronizerConfiguration build() {
           downloaderChainSegmentSize,
           downloaderParallelism,
           transactionsParallelism,
-          computationParallelism);
+          computationParallelism,
+          maxTrailingPeers);
     }
   }
 }

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

@@ -63,4 +63,19 @@ public String toString() {
         .add("maxTrailingPeers", maxTrailingPeers)
         .toString();
   }
+
+  /**
+   * Calculate a reasonable value for the maximum number of trailing peers to allow while behind the
+   * chain.
+   *
+   * <p>The number of peers is restricted to ensure we have room for peers ahead of us on the chain
+   * to connect and to limit the amount of work required to support requests from trailing peers
+   * while we're working to catch up.
+   *
+   * @param maxPeers the overall peer limit.
+   * @return the number of trailing peers allowed while catching up to the best peer.
+   */
+  public static int calculateMaxTrailingPeers(final int maxPeers) {
+    return Math.max((int) Math.ceil(maxPeers * 0.25), 1);
+  }
 }

ethereum/eth/src/main/java/tech/pegasys/pantheon/ethereum/eth/sync/fullsync/FullSyncDownloader.java

@@ -20,6 +20,7 @@
 import tech.pegasys.pantheon.ethereum.eth.sync.ChainDownloader;
 import tech.pegasys.pantheon.ethereum.eth.sync.CheckpointHeaderManager;
 import tech.pegasys.pantheon.ethereum.eth.sync.SynchronizerConfiguration;
+import tech.pegasys.pantheon.ethereum.eth.sync.TrailingPeerRequirements;
 import tech.pegasys.pantheon.ethereum.eth.sync.state.SyncState;
 import tech.pegasys.pantheon.ethereum.eth.sync.tasks.ImportBlocksTask;
 import tech.pegasys.pantheon.ethereum.eth.sync.tasks.ParallelImportChainSegmentTask;
@@ -38,6 +39,7 @@ public class FullSyncDownloader<C> {
   private final ProtocolSchedule<C> protocolSchedule;
   private final ProtocolContext<C> protocolContext;
   private final EthContext ethContext;
+  private final SyncState syncState;
   private final MetricsSystem metricsSystem;
 
   public FullSyncDownloader(
@@ -51,6 +53,7 @@ public FullSyncDownloader(
     this.protocolSchedule = protocolSchedule;
     this.protocolContext = protocolContext;
     this.ethContext = ethContext;
+    this.syncState = syncState;
     this.metricsSystem = metricsSystem;
     chainDownloader =
         new ChainDownloader<>(
@@ -104,4 +107,10 @@ private CompletableFuture<List<Block>> importBlocksForCheckpoints(
     }
     return importedBlocks;
   }
+
+  public TrailingPeerRequirements calculateTrailingPeerRequirements() {
+    return syncState.isInSync()
+        ? TrailingPeerRequirements.UNRESTRICTED
+        : new TrailingPeerRequirements(syncState.chainHeadNumber(), config.getMaxTrailingPeers());
+  }
 }

ethereum/eth/src/test/java/tech/pegasys/pantheon/ethereum/eth/sync/fullsync/FullSyncDownloaderTest.java

@@ -38,6 +38,7 @@
 import tech.pegasys.pantheon.ethereum.eth.messages.EthPV62;
 import tech.pegasys.pantheon.ethereum.eth.messages.GetBlockHeadersMessage;
 import tech.pegasys.pantheon.ethereum.eth.sync.SynchronizerConfiguration;
+import tech.pegasys.pantheon.ethereum.eth.sync.TrailingPeerRequirements;
 import tech.pegasys.pantheon.ethereum.eth.sync.state.SyncState;
 import tech.pegasys.pantheon.ethereum.mainnet.ProtocolSchedule;
 import tech.pegasys.pantheon.ethereum.p2p.api.MessageData;
@@ -577,6 +578,36 @@ public void requestsCheckpointsFromSyncTarget() {
     }
   }
 
+  @Test
+  public void shouldLimitTrailingPeersWhenBehindChain() {
+    localBlockchainSetup.importFirstBlocks(2);
+    final int maxTailingPeers = 5;
+    final FullSyncDownloader<?> downloader =
+        downloader(SynchronizerConfiguration.builder().maxTrailingPeers(maxTailingPeers).build());
+
+    final RespondingEthPeer bestPeer =
+        EthProtocolManagerTestUtil.createPeer(ethProtocolManager, 100);
+    syncState.setSyncTarget(bestPeer.getEthPeer(), localBlockchain.getChainHeadHeader());
+
+    final TrailingPeerRequirements expected =
+        new TrailingPeerRequirements(localBlockchain.getChainHeadBlockNumber(), maxTailingPeers);
+    assertThat(downloader.calculateTrailingPeerRequirements()).isEqualTo(expected);
+  }
+
+  @Test
+  public void shouldNotLimitTrailingPeersWhenInSync() {
+    localBlockchainSetup.importFirstBlocks(2);
+    final int maxTailingPeers = 5;
+    final FullSyncDownloader<?> downloader =
+        downloader(SynchronizerConfiguration.builder().maxTrailingPeers(maxTailingPeers).build());
+
+    final RespondingEthPeer bestPeer = EthProtocolManagerTestUtil.createPeer(ethProtocolManager, 2);
+    syncState.setSyncTarget(bestPeer.getEthPeer(), localBlockchain.getChainHeadHeader());
+
+    assertThat(downloader.calculateTrailingPeerRequirements())
+        .isEqualTo(TrailingPeerRequirements.UNRESTRICTED);
+  }
+
   private MutableBlockchain createShortChain(
       final Blockchain blockchain, final long truncateAtBlockNumber) {
     final BlockHeader genesisHeader =

pantheon/src/main/java/tech/pegasys/pantheon/cli/PantheonCommand.java

@@ -44,6 +44,7 @@
 import tech.pegasys.pantheon.ethereum.core.Wei;
 import tech.pegasys.pantheon.ethereum.eth.sync.SyncMode;
 import tech.pegasys.pantheon.ethereum.eth.sync.SynchronizerConfiguration;
+import tech.pegasys.pantheon.ethereum.eth.sync.TrailingPeerRequirements;
 import tech.pegasys.pantheon.ethereum.jsonrpc.JsonRpcConfiguration;
 import tech.pegasys.pantheon.ethereum.jsonrpc.RpcApi;
 import tech.pegasys.pantheon.ethereum.jsonrpc.RpcApis;
@@ -906,8 +907,10 @@ private PrivacyParameters privacyParameters() throws IOException {
   }
 
   private SynchronizerConfiguration buildSyncConfig() {
-    synchronizerConfigurationBuilder.syncMode(syncMode);
-    return synchronizerConfigurationBuilder.build();
+    return synchronizerConfigurationBuilder
+        .syncMode(syncMode)
+        .maxTrailingPeers(TrailingPeerRequirements.calculateMaxTrailingPeers(maxPeers))
+        .build();
   }
 
   // Blockchain synchronisation from peers.

pantheon/src/test/java/tech/pegasys/pantheon/cli/CommandTestAbstract.java

@@ -108,6 +108,8 @@ public void initMocks() throws Exception {
     when(mockControllerBuilder.metricsSystem(any())).thenReturn(mockControllerBuilder);
     when(mockControllerBuilder.privacyParameters(any())).thenReturn(mockControllerBuilder);
 
+    when(mockSyncConfBuilder.syncMode(any())).thenReturn(mockSyncConfBuilder);
+    when(mockSyncConfBuilder.maxTrailingPeers(anyInt())).thenReturn(mockSyncConfBuilder);
     when(mockSyncConfBuilder.build()).thenReturn(mockSyncConf);
 
     when(mockRunnerBuilder.vertx(any())).thenReturn(mockRunnerBuilder);