Request open ended headers from sync target (#1355)

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

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

@@ -15,6 +15,7 @@ package tech.pegasys.pantheon.ethereum.eth.sync;
 import static java.lang.Math.toIntExact;
 
 import tech.pegasys.pantheon.ethereum.core.BlockHeader;
+import tech.pegasys.pantheon.ethereum.eth.manager.EthPeer;
 
 import java.util.Objects;
 import java.util.Optional;
@@ -22,19 +23,27 @@ import java.util.Optional;
 import com.google.common.base.MoreObjects;
 
 public class CheckpointRange {
+
+  private final EthPeer syncTarget;
   private final BlockHeader start;
   private final Optional<BlockHeader> end;
 
-  public CheckpointRange(final BlockHeader start) {
+  public CheckpointRange(final EthPeer syncTarget, final BlockHeader start) {
+    this.syncTarget = syncTarget;
     this.start = start;
     this.end = Optional.empty();
   }
 
-  public CheckpointRange(final BlockHeader start, final BlockHeader end) {
+  public CheckpointRange(final EthPeer syncTarget, final BlockHeader start, final BlockHeader end) {
+    this.syncTarget = syncTarget;
     this.start = start;
     this.end = Optional.of(end);
   }
 
+  public EthPeer getSyncTarget() {
+    return syncTarget;
+  }
+
   public BlockHeader getStart() {
     return start;
   }
@@ -60,16 +69,22 @@ public class CheckpointRange {
       return false;
     }
     final CheckpointRange that = (CheckpointRange) o;
-    return Objects.equals(start, that.start) && Objects.equals(end, that.end);
+    return Objects.equals(syncTarget, that.syncTarget)
+        && Objects.equals(start, that.start)
+        && Objects.equals(end, that.end);
   }
 
   @Override
   public int hashCode() {
-    return Objects.hash(start, end);
+    return Objects.hash(syncTarget, start, end);
   }
 
   @Override
   public String toString() {
-    return MoreObjects.toStringHelper(this).add("start", start).add("end", end).toString();
+    return MoreObjects.toStringHelper(this)
+        .add("syncTarget", syncTarget)
+        .add("start", start)
+        .add("end", end)
+        .toString();
   }
 }

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

@@ -106,7 +106,7 @@ public class CheckpointRangeSource implements Iterator<CheckpointRange> {
     }
     if (checkpointFetcher.nextCheckpointEndsAtChainHead(peer, lastRangeEnd)) {
       reachedEndOfCheckpoints = true;
-      return new CheckpointRange(lastRangeEnd);
+      return new CheckpointRange(peer, lastRangeEnd);
     }
     pendingCheckpointsRequest = Optional.of(getNextCheckpointHeaders());
     return getCheckpointRangeFromPendingRequest();
@@ -147,7 +147,7 @@ public class CheckpointRangeSource implements Iterator<CheckpointRange> {
         requestFailureCount = 0;
       }
       for (final BlockHeader checkpointHeader : newCheckpointHeaders) {
-        retrievedRanges.add(new CheckpointRange(lastRangeEnd, checkpointHeader));
+        retrievedRanges.add(new CheckpointRange(peer, lastRangeEnd, checkpointHeader));
         lastRangeEnd = checkpointHeader;
       }
       return retrievedRanges.poll();

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

@@ -96,6 +96,7 @@ public class DownloadHeadersStep<C>
               checkpointRange.getStart().getNumber(),
               headerRequestSize,
               metricsSystem)
+          .assignPeer(checkpointRange.getSyncTarget())
           .run()
           .thenApply(PeerTaskResult::getResult);
     }

ethereum/eth/src/test/java/tech/pegasys/pantheon/ethereum/eth/sync/CheckpointHeaderValidationStepTest.java

@@ -24,6 +24,7 @@ import static tech.pegasys.pantheon.ethereum.mainnet.HeaderValidationMode.DETACH
 import tech.pegasys.pantheon.ethereum.ProtocolContext;
 import tech.pegasys.pantheon.ethereum.core.BlockDataGenerator;
 import tech.pegasys.pantheon.ethereum.core.BlockHeader;
+import tech.pegasys.pantheon.ethereum.eth.manager.EthPeer;
 import tech.pegasys.pantheon.ethereum.eth.sync.tasks.exceptions.InvalidBlockException;
 import tech.pegasys.pantheon.ethereum.mainnet.BlockHeaderValidator;
 import tech.pegasys.pantheon.ethereum.mainnet.ProtocolSchedule;
@@ -44,6 +45,7 @@ public class CheckpointHeaderValidationStepTest {
   @Mock private ProtocolContext<Void> protocolContext;
   @Mock private BlockHeaderValidator<Void> headerValidator;
   @Mock private ValidationPolicy validationPolicy;
+  @Mock private EthPeer syncTarget;
   private final BlockDataGenerator gen = new BlockDataGenerator();
   private CheckpointHeaderValidationStep<Void> validationStep;
 
@@ -51,7 +53,7 @@ public class CheckpointHeaderValidationStepTest {
   private final BlockHeader firstHeader = gen.header(11);
   private final CheckpointRangeHeaders rangeHeaders =
       new CheckpointRangeHeaders(
-          new CheckpointRange(checkpointStart, gen.header(13)),
+          new CheckpointRange(syncTarget, checkpointStart, gen.header(13)),
           asList(firstHeader, gen.header(12), gen.header(13)));
 
   @Before

ethereum/eth/src/test/java/tech/pegasys/pantheon/ethereum/eth/sync/CheckpointRangeSourceTest.java

@@ -172,18 +172,18 @@ public class CheckpointRangeSourceTest {
     when(checkpointFetcher.getNextCheckpointHeaders(peer, header(20)))
         .thenReturn(completedFuture(asList(header(25), header(30))));
 
-    assertThat(source.next()).isEqualTo(new CheckpointRange(commonAncestor, header(15)));
+    assertThat(source.next()).isEqualTo(new CheckpointRange(peer, commonAncestor, header(15)));
     verify(checkpointFetcher).getNextCheckpointHeaders(peer, commonAncestor);
     verify(checkpointFetcher).nextCheckpointEndsAtChainHead(peer, commonAncestor);
 
-    assertThat(source.next()).isEqualTo(new CheckpointRange(header(15), header(20)));
+    assertThat(source.next()).isEqualTo(new CheckpointRange(peer, header(15), header(20)));
     verifyNoMoreInteractions(checkpointFetcher);
 
-    assertThat(source.next()).isEqualTo(new CheckpointRange(header(20), header(25)));
+    assertThat(source.next()).isEqualTo(new CheckpointRange(peer, header(20), header(25)));
     verify(checkpointFetcher).getNextCheckpointHeaders(peer, header(20));
     verify(checkpointFetcher).nextCheckpointEndsAtChainHead(peer, header(20));
 
-    assertThat(source.next()).isEqualTo(new CheckpointRange(header(25), header(30)));
+    assertThat(source.next()).isEqualTo(new CheckpointRange(peer, header(25), header(30)));
     verifyNoMoreInteractions(checkpointFetcher);
   }
 
@@ -192,8 +192,8 @@ public class CheckpointRangeSourceTest {
     when(checkpointFetcher.getNextCheckpointHeaders(peer, commonAncestor))
         .thenReturn(completedFuture(asList(header(15), header(20))));
 
-    assertThat(source.next()).isEqualTo(new CheckpointRange(commonAncestor, header(15)));
-    assertThat(source.next()).isEqualTo(new CheckpointRange(header(15), header(20)));
+    assertThat(source.next()).isEqualTo(new CheckpointRange(peer, commonAncestor, header(15)));
+    assertThat(source.next()).isEqualTo(new CheckpointRange(peer, header(15), header(20)));
   }
 
   @Test
@@ -226,7 +226,7 @@ public class CheckpointRangeSourceTest {
     verify(checkpointFetcher).getNextCheckpointHeaders(peer, commonAncestor);
 
     future.complete(asList(header(15), header(20)));
-    assertThat(source.next()).isEqualTo(new CheckpointRange(commonAncestor, header(15)));
+    assertThat(source.next()).isEqualTo(new CheckpointRange(peer, commonAncestor, header(15)));
   }
 
   @Test
@@ -240,7 +240,7 @@ public class CheckpointRangeSourceTest {
     verify(checkpointFetcher).getNextCheckpointHeaders(peer, commonAncestor);
 
     // Then retries
-    assertThat(source.next()).isEqualTo(new CheckpointRange(commonAncestor, header(15)));
+    assertThat(source.next()).isEqualTo(new CheckpointRange(peer, commonAncestor, header(15)));
     verify(checkpointFetcher, times(2)).getNextCheckpointHeaders(peer, commonAncestor);
   }
 
@@ -251,7 +251,7 @@ public class CheckpointRangeSourceTest {
     when(checkpointFetcher.nextCheckpointEndsAtChainHead(peer, commonAncestor)).thenReturn(true);
 
     assertThat(source).hasNext();
-    assertThat(source.next()).isEqualTo(new CheckpointRange(commonAncestor));
+    assertThat(source.next()).isEqualTo(new CheckpointRange(peer, commonAncestor));
 
     // Once we've sent an open-ended range we shouldn't have any more ranges.
     assertThat(source).isExhausted();

ethereum/eth/src/test/java/tech/pegasys/pantheon/ethereum/eth/sync/DownloadHeadersStepTest.java

@@ -13,11 +13,13 @@
 package tech.pegasys.pantheon.ethereum.eth.sync;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
 import static tech.pegasys.pantheon.ethereum.eth.manager.RespondingEthPeer.blockchainResponder;
 
 import tech.pegasys.pantheon.ethereum.ProtocolContext;
 import tech.pegasys.pantheon.ethereum.chain.MutableBlockchain;
 import tech.pegasys.pantheon.ethereum.core.BlockHeader;
+import tech.pegasys.pantheon.ethereum.eth.manager.EthPeer;
 import tech.pegasys.pantheon.ethereum.eth.manager.EthProtocolManager;
 import tech.pegasys.pantheon.ethereum.eth.manager.EthProtocolManagerTestUtil;
 import tech.pegasys.pantheon.ethereum.eth.manager.RespondingEthPeer;
@@ -42,6 +44,7 @@ public class DownloadHeadersStepTest {
   private static ProtocolContext<Void> protocolContext;
   private static MutableBlockchain blockchain;
 
+  private final EthPeer syncTarget = mock(EthPeer.class);
   private EthProtocolManager ethProtocolManager;
   private DownloadHeadersStep<Void> downloader;
   private CheckpointRange checkpointRange;
@@ -70,7 +73,7 @@ public class DownloadHeadersStepTest {
 
     checkpointRange =
         new CheckpointRange(
-            blockchain.getBlockHeader(1).get(), blockchain.getBlockHeader(10).get());
+            syncTarget, blockchain.getBlockHeader(1).get(), blockchain.getBlockHeader(10).get());
   }
 
   @Test
@@ -103,7 +106,8 @@ public class DownloadHeadersStepTest {
   @Test
   public void shouldReturnOnlyEndHeaderWhenCheckpointRangeHasLengthOfOne() {
     final CheckpointRange checkpointRange =
-        new CheckpointRange(blockchain.getBlockHeader(3).get(), blockchain.getBlockHeader(4).get());
+        new CheckpointRange(
+            syncTarget, blockchain.getBlockHeader(3).get(), blockchain.getBlockHeader(4).get());
 
     final CompletableFuture<CheckpointRangeHeaders> result = this.downloader.apply(checkpointRange);
 
@@ -113,8 +117,9 @@ public class DownloadHeadersStepTest {
 
   @Test
   public void shouldGetRemainingHeadersWhenRangeHasNoEnd() {
-    final CheckpointRange checkpointRange = new CheckpointRange(blockchain.getBlockHeader(3).get());
     final RespondingEthPeer peer = EthProtocolManagerTestUtil.createPeer(ethProtocolManager, 1000);
+    final CheckpointRange checkpointRange =
+        new CheckpointRange(peer.getEthPeer(), blockchain.getBlockHeader(3).get());
 
     final CompletableFuture<CheckpointRangeHeaders> result = this.downloader.apply(checkpointRange);