From 196c4c177ceac032f27d6ec1de249e6318538cef Mon Sep 17 00:00:00 2001
From: Lucas Saldanha <lucas.saldanha@consensys.net>
Date: Fri, 21 Apr 2023 11:17:44 +1200
Subject: [PATCH] Handle errors on reporting tasks (#7062)

---
 beacon/validator/build.gradle                 |  1 +
 .../DefaultPerformanceTracker.java            |  5 ++-
 .../DefaultPerformanceTrackerTest.java        | 37 +++++++++++++++++++
 3 files changed, 42 insertions(+), 1 deletion(-)

diff --git a/beacon/validator/build.gradle b/beacon/validator/build.gradle
index dcf619a98e7..34e3e559db3 100644
--- a/beacon/validator/build.gradle
+++ b/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'))
 
diff --git a/beacon/validator/src/main/java/tech/pegasys/teku/validator/coordinator/performance/DefaultPerformanceTracker.java b/beacon/validator/src/main/java/tech/pegasys/teku/validator/coordinator/performance/DefaultPerformanceTracker.java
index 1c77adae682..cadba313d3e 100644
--- a/beacon/validator/src/main/java/tech/pegasys/teku/validator/coordinator/performance/DefaultPerformanceTracker.java
+++ b/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) {
diff --git a/beacon/validator/src/test/java/tech/pegasys/teku/validator/coordinator/performance/DefaultPerformanceTrackerTest.java b/beacon/validator/src/test/java/tech/pegasys/teku/validator/coordinator/performance/DefaultPerformanceTrackerTest.java
index 45dad15ff54..26effa4f0ab 100644
--- a/beacon/validator/src/test/java/tech/pegasys/teku/validator/coordinator/performance/DefaultPerformanceTrackerTest.java
+++ b/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