Download receipts.

ethereum/eth/src/main/java/tech/pegasys/pantheon/ethereum/eth/sync/fastsync/BlockWithReceipts.java

@@ -17,6 +17,9 @@
 import tech.pegasys.pantheon.ethereum.core.TransactionReceipt;
 
 import java.util.List;
+import java.util.Objects;
+
+import com.google.common.base.MoreObjects;
 
 class BlockWithReceipts {
   private final Block block;
@@ -38,4 +41,29 @@ public Block getBlock() {
   public List<TransactionReceipt> getReceipts() {
     return receipts;
   }
+
+  @Override
+  public boolean equals(final Object o) {
+    if (this == o) {
+      return true;
+    }
+    if (o == null || getClass() != o.getClass()) {
+      return false;
+    }
+    final BlockWithReceipts that = (BlockWithReceipts) o;
+    return Objects.equals(block, that.block) && Objects.equals(receipts, that.receipts);
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(block, receipts);
+  }
+
+  @Override
+  public String toString() {
+    return MoreObjects.toStringHelper(this)
+        .add("block", block)
+        .add("receipts", receipts)
+        .toString();
+  }
 }

ethereum/eth/src/main/java/tech/pegasys/pantheon/ethereum/eth/sync/fastsync/DownloadReceiptsStep.java

@@ -0,0 +1,64 @@
+/*
+ * Copyright 2019 ConsenSys AG.
+ *
+ * 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.
+ */
+package tech.pegasys.pantheon.ethereum.eth.sync.fastsync;
+
+import static java.util.Collections.emptyList;
+import static java.util.stream.Collectors.toList;
+
+import tech.pegasys.pantheon.ethereum.core.Block;
+import tech.pegasys.pantheon.ethereum.core.BlockHeader;
+import tech.pegasys.pantheon.ethereum.core.TransactionReceipt;
+import tech.pegasys.pantheon.ethereum.eth.manager.EthContext;
+import tech.pegasys.pantheon.ethereum.eth.sync.tasks.GetReceiptsForHeadersTask;
+import tech.pegasys.pantheon.metrics.MetricsSystem;
+import tech.pegasys.pantheon.util.FutureUtils;
+
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.function.Function;
+
+public class DownloadReceiptsStep
+    implements Function<List<Block>, CompletableFuture<List<BlockWithReceipts>>> {
+  private final EthContext ethContext;
+  private final MetricsSystem metricsSystem;
+
+  public DownloadReceiptsStep(final EthContext ethContext, final MetricsSystem metricsSystem) {
+    this.ethContext = ethContext;
+    this.metricsSystem = metricsSystem;
+  }
+
+  @Override
+  public CompletableFuture<List<BlockWithReceipts>> apply(final List<Block> blocks) {
+    final List<BlockHeader> headers = blocks.stream().map(Block::getHeader).collect(toList());
+    final CompletableFuture<Map<BlockHeader, List<TransactionReceipt>>> getReceipts =
+        GetReceiptsForHeadersTask.forHeaders(ethContext, headers, metricsSystem).run();
+    final CompletableFuture<List<BlockWithReceipts>> combineWithBlocks =
+        getReceipts.thenApply(
+            receiptsByHeader -> combineBlocksAndReceipts(blocks, receiptsByHeader));
+    FutureUtils.propagateCancellation(combineWithBlocks, getReceipts);
+    return combineWithBlocks;
+  }
+
+  private List<BlockWithReceipts> combineBlocksAndReceipts(
+      final List<Block> blocks, final Map<BlockHeader, List<TransactionReceipt>> receiptsByHeader) {
+    return blocks.stream()
+        .map(
+            block -> {
+              final List<TransactionReceipt> receipts =
+                  receiptsByHeader.getOrDefault(block.getHeader(), emptyList());
+              return new BlockWithReceipts(block, receipts);
+            })
+        .collect(toList());
+  }
+}

ethereum/eth/src/main/java/tech/pegasys/pantheon/ethereum/eth/sync/fastsync/FastSyncDownloadPipelineFactory.java

@@ -36,7 +36,6 @@
 import tech.pegasys.pantheon.services.pipeline.PipelineBuilder;
 
 import java.time.Duration;
-import java.util.function.Function;
 
 public class FastSyncDownloadPipelineFactory<C> implements DownloadPipelineFactory {
   private final SynchronizerConfiguration syncConfig;
@@ -117,7 +116,10 @@ public Pipeline<?> createDownloadPipelineForSyncTarget(final SyncTarget target)
         .thenFlatMap("validateHeadersJoin", validateHeadersJoinUpStep, singleHeaderBufferSize)
         .inBatches(headerRequestSize)
         .thenProcessAsync("downloadBodies", downloadBodiesStep, downloaderParallelism)
-        .thenProcess("downloadReceipts", Function.identity())
+        .thenProcessAsync(
+            "downloadReceipts",
+            new DownloadReceiptsStep(ethContext, metricsSystem),
+            downloaderParallelism)
         .andFinishWith("complete", result -> {});
   }
 

ethereum/eth/src/test/java/tech/pegasys/pantheon/ethereum/eth/sync/fastsync/DownloadReceiptsStepTest.java

@@ -0,0 +1,88 @@
+/*
+ * Copyright 2019 ConsenSys AG.
+ *
+ * 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.
+ */
+package tech.pegasys.pantheon.ethereum.eth.sync.fastsync;
+
+import static java.util.Arrays.asList;
+import static org.assertj.core.api.Assertions.assertThat;
+
+import tech.pegasys.pantheon.ethereum.ProtocolContext;
+import tech.pegasys.pantheon.ethereum.chain.MutableBlockchain;
+import tech.pegasys.pantheon.ethereum.core.Block;
+import tech.pegasys.pantheon.ethereum.core.BlockHeader;
+import tech.pegasys.pantheon.ethereum.core.TransactionReceipt;
+import tech.pegasys.pantheon.ethereum.eth.manager.EthProtocolManager;
+import tech.pegasys.pantheon.ethereum.eth.manager.EthProtocolManagerTestUtil;
+import tech.pegasys.pantheon.ethereum.eth.manager.RespondingEthPeer;
+import tech.pegasys.pantheon.ethereum.eth.manager.ethtaskutils.BlockchainSetupUtil;
+import tech.pegasys.pantheon.metrics.noop.NoOpMetricsSystem;
+
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+public class DownloadReceiptsStepTest {
+
+  private static ProtocolContext<Void> protocolContext;
+  private static MutableBlockchain blockchain;
+
+  private EthProtocolManager ethProtocolManager;
+  private DownloadReceiptsStep downloadReceiptsStep;
+
+  @BeforeClass
+  public static void setUpClass() {
+    final BlockchainSetupUtil<Void> setupUtil = BlockchainSetupUtil.forTesting();
+    setupUtil.importFirstBlocks(20);
+    protocolContext = setupUtil.getProtocolContext();
+    blockchain = setupUtil.getBlockchain();
+  }
+
+  @Before
+  public void setUp() {
+    ethProtocolManager =
+        EthProtocolManagerTestUtil.create(blockchain, protocolContext.getWorldStateArchive());
+    downloadReceiptsStep =
+        new DownloadReceiptsStep(ethProtocolManager.ethContext(), new NoOpMetricsSystem());
+  }
+
+  @Test
+  public void shouldDownloadReceiptsForBlocks() {
+    final RespondingEthPeer peer = EthProtocolManagerTestUtil.createPeer(ethProtocolManager, 1000);
+
+    final List<Block> blocks = asList(block(1), block(2), block(3), block(4));
+    final CompletableFuture<List<BlockWithReceipts>> result = downloadReceiptsStep.apply(blocks);
+
+    peer.respond(RespondingEthPeer.blockchainResponder(blockchain));
+
+    assertThat(result)
+        .isCompletedWithValue(
+            asList(
+                blockWithReceipts(1),
+                blockWithReceipts(2),
+                blockWithReceipts(3),
+                blockWithReceipts(4)));
+  }
+
+  private Block block(final long number) {
+    final BlockHeader header = blockchain.getBlockHeader(number).get();
+    return new Block(header, blockchain.getBlockBody(header.getHash()).get());
+  }
+
+  private BlockWithReceipts blockWithReceipts(final long number) {
+    final Block block = block(number);
+    final List<TransactionReceipt> receipts = blockchain.getTxReceipts(block.getHash()).get();
+    return new BlockWithReceipts(block, receipts);
+  }
+}