Revalidate future blocks on slot replay

ethereum/statetransition/src/main/java/tech/pegasys/teku/statetransition/block/BlockManager.java

@@ -188,7 +188,9 @@ public SafeFuture<InternalValidationResult> validateAndImportBlock(
   @Override
   public void onSlot(final UInt64 slot) {
     futureBlocks.onSlot(slot);
-    futureBlocks.prune(slot).forEach(this::importBlockIgnoringResult);
+    futureBlocks
+        .prune(slot)
+        .forEach(block -> validateAndImportBlock(block, Optional.empty()).finishError(LOG));
   }
 
   public void subscribeFailedPayloadExecution(final FailedPayloadExecutionSubscriber subscriber) {

ethereum/statetransition/src/test/java/tech/pegasys/teku/statetransition/block/BlockManagerTest.java

@@ -65,6 +65,7 @@
 import tech.pegasys.teku.infrastructure.async.ExceptionThrowingFutureSupplier;
 import tech.pegasys.teku.infrastructure.async.SafeFuture;
 import tech.pegasys.teku.infrastructure.async.SafeFutureAssert;
+import tech.pegasys.teku.infrastructure.async.Waiter;
 import tech.pegasys.teku.infrastructure.async.eventthread.InlineEventThread;
 import tech.pegasys.teku.infrastructure.collections.LimitedMap;
 import tech.pegasys.teku.infrastructure.logging.EventLogger;
@@ -547,6 +548,32 @@ public void onProposedBlock_futureBlock() {
     verifyNoInteractions(blobSidecarManager);
   }
 
+  @Test
+  public void onProposedBlock_futureBlock_shouldRerunGossipValidationOnRetry() {
+    incrementSlot();
+    final UInt64 nextSlot = currentSlot.plus(UInt64.ONE);
+    final SignedBeaconBlock futureBlock =
+        localChain.chainBuilder().generateBlockAtSlot(nextSlot).getBlock();
+
+    when(blockValidator.validateGossip(eq(futureBlock)))
+        .thenReturn(SafeFuture.completedFuture(InternalValidationResult.SAVE_FOR_FUTURE))
+        .thenReturn(
+            SafeFuture.completedFuture(InternalValidationResult.reject("retry validation failed")));
+
+    assertThatSafeFuture(blockManager.validateAndImportBlock(futureBlock, Optional.empty()))
+        .isCompletedWithValue(InternalValidationResult.SAVE_FOR_FUTURE);
+    Waiter.waitFor(() -> assertThat(futureBlocks.size()).isEqualTo(1));
+    assertThat(futureBlocks.contains(futureBlock)).isTrue();
+    assertThat(invalidBlockRoots).isEmpty();
+
+    incrementSlot();
+
+    Waiter.waitFor(() -> verify(blockValidator, times(2)).validateGossip(eq(futureBlock)));
+    Waiter.waitFor(() -> assertThat(invalidBlockRoots).isEmpty());
+    Waiter.waitFor(() -> assertThat(futureBlocks.size()).isEqualTo(0));
+    verify(blockEventsListenerRouter).removeAllForBlock(futureBlock.getSlotAndBlockRoot());
+  }
+
   @Test
   public void onBlockImported_withPendingBlocks() {
     final int blockCount = 3;