Handle errors on reporting tasks (#7062)

Author: lucassaldanha

beacon/validator/build.gradle

@@ -37,6 +37,7 @@ dependencies {
   testImplementation testFixtures(project(':ethereum:networks'))
   testImplementation testFixtures(project(':storage'))
   testImplementation testFixtures(project(':infrastructure:async'))
+  testImplementation testFixtures(project(':infrastructure:logging'))
   testImplementation testFixtures(project(':infrastructure:metrics'))
   testImplementation testFixtures(project(':infrastructure:time'))
 

beacon/validator/src/main/java/tech/pegasys/teku/validator/coordinator/performance/DefaultPerformanceTracker.java

@@ -36,6 +36,8 @@
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.stream.Collectors;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
 import org.apache.tuweni.bytes.Bytes32;
 import tech.pegasys.teku.infrastructure.async.SafeFuture;
 import tech.pegasys.teku.infrastructure.logging.StatusLogger;
@@ -53,6 +55,7 @@
 import tech.pegasys.teku.validator.coordinator.ActiveValidatorTracker;
 
 public class DefaultPerformanceTracker implements PerformanceTracker {
+  private static final Logger LOG = LogManager.getLogger();
 
   @VisibleForTesting
   final NavigableMap<UInt64, Set<SlotAndBlockRoot>> producedBlocksByEpoch =
@@ -136,7 +139,7 @@ public void onSlot(UInt64 slot) {
       reportingTasks.add(reportSyncCommitteePerformance(currentEpoch));
     }
 
-    SafeFuture.allOf(reportingTasks.toArray(SafeFuture[]::new)).join();
+    SafeFuture.allOf(reportingTasks.toArray(SafeFuture[]::new)).handleException(LOG::error).join();
   }
 
   private SafeFuture<?> reportBlockPerformance(final UInt64 currentEpoch) {

beacon/validator/src/test/java/tech/pegasys/teku/validator/coordinator/performance/DefaultPerformanceTrackerTest.java

@@ -19,12 +19,14 @@
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoInteractions;
 import static org.mockito.Mockito.when;
 import static tech.pegasys.teku.validator.coordinator.performance.DefaultPerformanceTracker.ATTESTATION_INCLUSION_RANGE;
 
 import java.util.List;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
+import tech.pegasys.infrastructure.logging.LogCaptor;
 import tech.pegasys.teku.bls.BLSKeyGenerator;
 import tech.pegasys.teku.bls.BLSKeyPair;
 import tech.pegasys.teku.bls.BLSTestUtil;
@@ -362,6 +364,41 @@ void shouldReportSyncCommitteePerformance() {
     verify(validatorPerformanceMetrics).updateSyncCommitteePerformance(performance);
   }
 
+  @Test
+  void shouldHandleErrorsWhenReportTasksFail() {
+    chainUpdater.updateBestBlock(chainUpdater.advanceChainUntil(1));
+    final Attestation attestation = createAttestationForParentBlockOnSlot(1);
+    final UInt64 slot = spec.computeStartSlotAtEpoch(ATTESTATION_INCLUSION_RANGE);
+
+    performanceTracker.saveProducedAttestation(attestation);
+    when(validatorTracker.getNumberOfValidatorsForEpoch(any())).thenThrow(new RuntimeException());
+
+    try (LogCaptor logCaptor = LogCaptor.forClass(DefaultPerformanceTracker.class)) {
+      performanceTracker.onSlot(slot);
+
+      // No attestation performance report on status logger because task failed
+      verifyNoInteractions(log);
+      assertThat(logCaptor.getErrorLogs()).hasSize(1);
+    }
+  }
+
+  /**
+   * Creates an attestation voting for block on the slot provided. The attestation will be included
+   * in block slot + 1.
+   *
+   * @param slot the slot of the block being attested
+   * @return the created attestation
+   */
+  private Attestation createAttestationForParentBlockOnSlot(int slot) {
+    Attestation attestationForBlock1 = createAttestation(slot + 1, slot);
+    ChainBuilder.BlockOptions block2Options = ChainBuilder.BlockOptions.create();
+    block2Options.addAttestation(attestationForBlock1);
+    SignedBlockAndState latestBlockAndState = chainBuilder.generateBlockAtSlot(2, block2Options);
+    chainUpdater.saveBlock(latestBlockAndState);
+    chainUpdater.updateBestBlock(latestBlockAndState);
+    return attestationForBlock1;
+  }
+
   private Attestation createAttestation(
       ChainBuilder chainBuilder, int validForBlockAtSlot, int vouchingForBlockAtSlot) {
     return chainBuilder