Select new pivot block when world state becomes unavailable (#869)

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

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

@@ -131,6 +131,7 @@ class FastSynchronizer<C> {
   }
 
   public CompletableFuture<FastSyncState> start() {
+    LOG.info("Fast sync enabled");
     return fastSyncDownloader.start(initialSyncState);
   }
 

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

@@ -12,7 +12,12 @@
  */
 package tech.pegasys.pantheon.ethereum.eth.sync.fastsync;
 
+import static tech.pegasys.pantheon.util.FutureUtils.completedExceptionally;
+import static tech.pegasys.pantheon.util.FutureUtils.exceptionallyCompose;
+
 import tech.pegasys.pantheon.ethereum.eth.sync.worldstate.WorldStateDownloader;
+import tech.pegasys.pantheon.ethereum.eth.sync.worldstate.WorldStateUnavailableException;
+import tech.pegasys.pantheon.util.ExceptionUtils;
 
 import java.util.concurrent.CompletableFuture;
 
@@ -35,13 +40,24 @@ public class FastSyncDownloader<C> {
   }
 
   public CompletableFuture<FastSyncState> start(final FastSyncState fastSyncState) {
-    LOG.info("Fast sync enabled");
-    return fastSyncActions
-        .waitForSuitablePeers(fastSyncState)
-        .thenCompose(fastSyncActions::selectPivotBlock)
-        .thenCompose(fastSyncActions::downloadPivotBlockHeader)
-        .thenApply(this::storeState)
-        .thenCompose(this::downloadChainAndWorldState);
+    return exceptionallyCompose(
+        fastSyncActions
+            .waitForSuitablePeers(fastSyncState)
+            .thenCompose(fastSyncActions::selectPivotBlock)
+            .thenCompose(fastSyncActions::downloadPivotBlockHeader)
+            .thenApply(this::storeState)
+            .thenCompose(this::downloadChainAndWorldState),
+        this::handleWorldStateUnavailable);
+  }
+
+  private CompletableFuture<FastSyncState> handleWorldStateUnavailable(final Throwable error) {
+    if (ExceptionUtils.rootCause(error) instanceof WorldStateUnavailableException) {
+      LOG.warn(
+          "Fast sync was unable to download the world state. Retrying with a new pivot block.");
+      return start(FastSyncState.EMPTY_SYNC_STATE);
+    } else {
+      return completedExceptionally(error);
+    }
   }
 
   private FastSyncState storeState(final FastSyncState state) {

ethereum/eth/src/main/java/tech/pegasys/pantheon/ethereum/eth/sync/worldstate/WorldStateUnavailableException.java

@@ -0,0 +1,15 @@
+/*
+ * 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.worldstate;
+
+public class WorldStateUnavailableException extends RuntimeException {}

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

@@ -15,6 +15,7 @@ package tech.pegasys.pantheon.ethereum.eth.sync.fastsync;
 import static java.util.concurrent.CompletableFuture.completedFuture;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.verifyNoMoreInteractions;
 import static org.mockito.Mockito.when;
@@ -26,6 +27,7 @@ import static tech.pegasys.pantheon.ethereum.eth.sync.fastsync.FastSyncState.EMP
 import tech.pegasys.pantheon.ethereum.core.BlockHeader;
 import tech.pegasys.pantheon.ethereum.core.BlockHeaderTestFixture;
 import tech.pegasys.pantheon.ethereum.eth.sync.worldstate.WorldStateDownloader;
+import tech.pegasys.pantheon.ethereum.eth.sync.worldstate.WorldStateUnavailableException;
 
 import java.util.concurrent.CompletableFuture;
 
@@ -247,6 +249,66 @@ public class FastSyncDownloaderTest {
     assertThat(result).isNotDone();
   }
 
+  @SuppressWarnings("unchecked")
+  @Test
+  public void shouldResetFastSyncStateAndRestartProcessIfWorldStateIsUnavailable() {
+    final CompletableFuture<Void> firstWorldStateFuture = new CompletableFuture<>();
+    final CompletableFuture<Void> secondWorldStateFuture = new CompletableFuture<>();
+    final CompletableFuture<FastSyncState> chainFuture = new CompletableFuture<>();
+    final FastSyncState selectPivotBlockState = new FastSyncState(50);
+    final FastSyncState secondSelectPivotBlockState = new FastSyncState(90);
+    final BlockHeader pivotBlockHeader = new BlockHeaderTestFixture().number(50).buildHeader();
+    final BlockHeader secondPivotBlockHeader =
+        new BlockHeaderTestFixture().number(90).buildHeader();
+    final FastSyncState downloadPivotBlockHeaderState = new FastSyncState(pivotBlockHeader);
+    final FastSyncState secondDownloadPivotBlockHeaderState =
+        new FastSyncState(secondPivotBlockHeader);
+    // First attempt
+    when(fastSyncActions.waitForSuitablePeers(EMPTY_SYNC_STATE)).thenReturn(COMPLETE);
+    when(fastSyncActions.selectPivotBlock(EMPTY_SYNC_STATE))
+        .thenReturn(
+            completedFuture(selectPivotBlockState), completedFuture(secondSelectPivotBlockState));
+    when(fastSyncActions.downloadPivotBlockHeader(selectPivotBlockState))
+        .thenReturn(completedFuture(downloadPivotBlockHeaderState));
+    when(fastSyncActions.downloadChain(downloadPivotBlockHeaderState)).thenReturn(chainFuture);
+    when(worldStateDownloader.run(pivotBlockHeader)).thenReturn(firstWorldStateFuture);
+
+    // Second attempt with new pivot block
+    when(fastSyncActions.downloadPivotBlockHeader(secondSelectPivotBlockState))
+        .thenReturn(completedFuture(secondDownloadPivotBlockHeaderState));
+    when(fastSyncActions.downloadChain(secondDownloadPivotBlockHeaderState))
+        .thenReturn(completedFuture(secondDownloadPivotBlockHeaderState));
+    when(worldStateDownloader.run(secondPivotBlockHeader)).thenReturn(secondWorldStateFuture);
+
+    final CompletableFuture<FastSyncState> result = downloader.start(EMPTY_SYNC_STATE);
+
+    verify(fastSyncActions).waitForSuitablePeers(EMPTY_SYNC_STATE);
+    verify(fastSyncActions).selectPivotBlock(EMPTY_SYNC_STATE);
+    verify(fastSyncActions).downloadPivotBlockHeader(selectPivotBlockState);
+    verify(storage).storeState(downloadPivotBlockHeaderState);
+    verify(fastSyncActions).downloadChain(downloadPivotBlockHeaderState);
+    verify(worldStateDownloader).run(pivotBlockHeader);
+    verifyNoMoreInteractions(fastSyncActions, worldStateDownloader, storage);
+
+    assertThat(result).isNotDone();
+
+    firstWorldStateFuture.completeExceptionally(new WorldStateUnavailableException());
+    assertThat(result).isNotDone();
+    assertThat(chainFuture).isCancelled();
+
+    verify(fastSyncActions, times(2)).waitForSuitablePeers(EMPTY_SYNC_STATE);
+    verify(fastSyncActions, times(2)).selectPivotBlock(EMPTY_SYNC_STATE);
+    verify(fastSyncActions).downloadPivotBlockHeader(secondSelectPivotBlockState);
+    verify(storage).storeState(secondDownloadPivotBlockHeaderState);
+    verify(fastSyncActions).downloadChain(secondDownloadPivotBlockHeaderState);
+    verify(worldStateDownloader).run(secondPivotBlockHeader);
+    verifyNoMoreInteractions(fastSyncActions, worldStateDownloader, storage);
+
+    secondWorldStateFuture.complete(null);
+
+    assertThat(result).isCompletedWithValue(secondDownloadPivotBlockHeaderState);
+  }
+
   private <T> CompletableFuture<T> completedExceptionally(final Throwable error) {
     final CompletableFuture<T> result = new CompletableFuture<>();
     result.completeExceptionally(error);