Produce block only once for a slot (#8773)

beacon/validator/src/main/java/tech/pegasys/teku/validator/coordinator/BlockOperationSelectorFactory.java

@@ -465,7 +465,7 @@ public Function<SignedBlockContainer, List<BlobSidecar>> createBlobSidecarsSelec
         // the blobs and the proofs wouldn't be part of the BlockContainer.
         final BuilderPayloadOrFallbackData builderPayloadOrFallbackData =
             executionLayerBlockProductionManager
-                .getCachedUnblindedPayload(block.getSlotAndBlockRoot())
+                .getCachedUnblindedPayload(slot)
                 .orElseThrow(
                     () ->
                         new IllegalStateException(

beacon/validator/src/main/java/tech/pegasys/teku/validator/coordinator/ValidatorApiHandler.java

@@ -34,7 +34,6 @@
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
-import java.util.Objects;
 import java.util.Optional;
 import java.util.Set;
 import org.apache.logging.log4j.LogManager;
@@ -122,8 +121,8 @@ public class ValidatorApiHandler implements ValidatorApiChannel {
    */
   private static final int DUTY_EPOCH_TOLERANCE = 1;
 
-  private final Map<UInt64, Bytes32> createdBlockRootsBySlotCache =
-      LimitedMap.createSynchronizedLRU(2);
+  private final Map<UInt64, SafeFuture<Optional<BlockContainerAndMetaData>>>
+      localBlockProductionBySlotCache = LimitedMap.createSynchronizedLRU(2);
 
   private final BlockProductionAndPublishingPerformanceFactory
       blockProductionAndPublishingPerformanceFactory;
@@ -330,12 +329,35 @@ public SafeFuture<Optional<Map<BLSPublicKey, ValidatorStatus>>> getValidatorStat
                                         StateValidatorData::getStatus))));
   }
 
+  /**
+   * Block would be produced only once per slot. Any additional calls to this method for the same
+   * slot would return the same {@link SafeFuture} as the first one. The only exception is when the
+   * block production fails. In this case, the next call would attempt to produce the block again.
+   */
   @Override
   public SafeFuture<Optional<BlockContainerAndMetaData>> createUnsignedBlock(
       final UInt64 slot,
       final BLSSignature randaoReveal,
       final Optional<Bytes32> graffiti,
       final Optional<UInt64> requestedBuilderBoostFactor) {
+    return localBlockProductionBySlotCache
+        .computeIfAbsent(
+            slot,
+            __ ->
+                createUnsignedBlockInternal(
+                    slot, randaoReveal, graffiti, requestedBuilderBoostFactor))
+        .whenException(
+            __ -> {
+              // allow further block production attempts for this slot
+              localBlockProductionBySlotCache.remove(slot);
+            });
+  }
+
+  public SafeFuture<Optional<BlockContainerAndMetaData>> createUnsignedBlockInternal(
+      final UInt64 slot,
+      final BLSSignature randaoReveal,
+      final Optional<Bytes32> graffiti,
+      final Optional<UInt64> requestedBuilderBoostFactor) {
     LOG.info("Creating unsigned block for slot {}", slot);
     performanceTracker.reportBlockProductionAttempt(spec.computeEpochAtSlot(slot));
     if (isSyncActive()) {
@@ -389,12 +411,7 @@ private SafeFuture<Optional<BlockContainerAndMetaData>> createBlock(
             graffiti,
             requestedBuilderBoostFactor,
             blockProductionPerformance)
-        .thenApply(
-            block -> {
-              final Bytes32 blockRoot = block.blockContainer().getBlock().getRoot();
-              createdBlockRootsBySlotCache.put(slot, blockRoot);
-              return Optional.of(block);
-            });
+        .thenApply(Optional::of);
   }
 
   @Override
@@ -866,11 +883,21 @@ private List<ProposerDuty> getProposalSlotsForEpoch(final BeaconState state, fin
     return proposerSlots;
   }
 
-  private boolean isLocallyCreatedBlock(final SignedBlockContainer blockContainer) {
-    final Bytes32 blockRoot = blockContainer.getSignedBlock().getMessage().getRoot();
-    final Bytes32 locallyCreatedBlockRoot =
-        createdBlockRootsBySlotCache.get(blockContainer.getSlot());
-    return Objects.equals(blockRoot, locallyCreatedBlockRoot);
+  private boolean isLocallyCreatedBlock(final SignedBlockContainer signedBlockContainer) {
+    final SafeFuture<Optional<BlockContainerAndMetaData>> localBlockProduction =
+        localBlockProductionBySlotCache.get(signedBlockContainer.getSlot());
+    if (localBlockProduction == null || !localBlockProduction.isCompletedNormally()) {
+      return false;
+    }
+    return localBlockProduction
+        .getImmediately()
+        .map(
+            blockContainerAndMetaData ->
+                blockContainerAndMetaData
+                    .blockContainer()
+                    .getRoot()
+                    .equals(signedBlockContainer.getRoot()))
+        .orElse(false);
   }
 
   @Override

beacon/validator/src/test/java/tech/pegasys/teku/validator/coordinator/AbstractBlockFactoryTest.java

@@ -374,8 +374,7 @@ protected BlockAndBlobSidecars createBlockAndBlobSidecars(
     }
 
     // simulate caching of the builder payload
-    when(executionLayer.getCachedUnblindedPayload(
-            signedBlockContainer.getSignedBlock().getSlotAndBlockRoot()))
+    when(executionLayer.getCachedUnblindedPayload(signedBlockContainer.getSlot()))
         .thenReturn(builderPayload.map(BuilderPayloadOrFallbackData::create));
 
     final List<BlobSidecar> blobSidecars =

beacon/validator/src/test/java/tech/pegasys/teku/validator/coordinator/BlockFactoryDenebTest.java

@@ -156,7 +156,7 @@ void shouldCreateValidBlobSidecarsForBlindedBlock() {
     final SignedBlockContainer block = blockAndBlobSidecars.block();
     final List<BlobSidecar> blobSidecars = blockAndBlobSidecars.blobSidecars();
 
-    verify(executionLayer).getCachedUnblindedPayload(block.getSlotAndBlockRoot());
+    verify(executionLayer).getCachedUnblindedPayload(block.getSlot());
 
     final SszList<SszKZGCommitment> expectedCommitments =
         block.getSignedBlock().getMessage().getBody().getOptionalBlobKzgCommitments().orElseThrow();

beacon/validator/src/test/java/tech/pegasys/teku/validator/coordinator/BlockOperationSelectorFactoryTest.java

@@ -50,7 +50,6 @@
 import tech.pegasys.teku.spec.datastructures.blocks.BeaconBlock;
 import tech.pegasys.teku.spec.datastructures.blocks.Eth1Data;
 import tech.pegasys.teku.spec.datastructures.blocks.SignedBeaconBlock;
-import tech.pegasys.teku.spec.datastructures.blocks.SlotAndBlockRoot;
 import tech.pegasys.teku.spec.datastructures.blocks.blockbody.BeaconBlockBody;
 import tech.pegasys.teku.spec.datastructures.blocks.blockbody.BeaconBlockBodyBuilder;
 import tech.pegasys.teku.spec.datastructures.blocks.blockbody.BeaconBlockBodySchema;
@@ -818,7 +817,7 @@ void shouldFailCreatingBlobSidecarsIfBuilderBlobsBundleCommitmentsRootIsNotConsi
         dataStructureUtil.randomBuilderBlobsBundle(3);
 
     prepareCachedBuilderPayload(
-        signedBlindedBeaconBlock.getSlotAndBlockRoot(),
+        signedBlindedBeaconBlock.getSlot(),
         dataStructureUtil.randomExecutionPayload(),
         blobsBundle);
 
@@ -843,7 +842,7 @@ void shouldFailCreatingBlobSidecarsIfBuilderBlobsBundleProofsIsNotConsistent() {
     when(blobsBundle.getBlobs()).thenReturn(dataStructureUtil.randomSszBlobs(2));
 
     prepareCachedBuilderPayload(
-        signedBlindedBeaconBlock.getSlotAndBlockRoot(),
+        signedBlindedBeaconBlock.getSlot(),
         dataStructureUtil.randomExecutionPayload(),
         blobsBundle);
 
@@ -868,7 +867,7 @@ void shouldFailCreatingBlobSidecarsIfBuilderBlobsBundleBlobsIsNotConsistent() {
     when(blobsBundle.getProofs()).thenReturn(dataStructureUtil.randomSszKZGProofs(2));
 
     prepareCachedBuilderPayload(
-        signedBlindedBeaconBlock.getSlotAndBlockRoot(),
+        signedBlindedBeaconBlock.getSlot(),
         dataStructureUtil.randomExecutionPayload(),
         blobsBundle);
 
@@ -902,14 +901,10 @@ void shouldCreateBlobSidecarsForBlindedBlock(final boolean useLocalFallback) {
                   .toList(),
               blobsBundle.getProofs().stream().map(SszKZGProof::getKZGProof).toList(),
               blobsBundle.getBlobs().stream().toList());
-      prepareCachedFallbackData(
-          signedBlindedBeaconBlock.getSlotAndBlockRoot(),
-          executionPayload,
-          localFallbackBlobsBundle);
+      prepareCachedFallbackData(slot, executionPayload, localFallbackBlobsBundle);
     } else {
 
-      prepareCachedBuilderPayload(
-          signedBlindedBeaconBlock.getSlotAndBlockRoot(), executionPayload, blobsBundle);
+      prepareCachedBuilderPayload(slot, executionPayload, blobsBundle);
     }
 
     final List<BlobSidecar> blobSidecars =
@@ -1286,23 +1281,20 @@ private void prepareCachedPayloadHeaderWithFallbackResult(
   }
 
   private void prepareCachedBuilderPayload(
-      final SlotAndBlockRoot slotAndBlockRoot,
+      final UInt64 slot,
       final ExecutionPayload executionPayload,
       final tech.pegasys.teku.spec.datastructures.builder.BlobsBundle blobsBundle) {
     final BuilderPayload builderPayload =
-        SchemaDefinitionsDeneb.required(
-                spec.atSlot(slotAndBlockRoot.getSlot()).getSchemaDefinitions())
+        SchemaDefinitionsDeneb.required(spec.atSlot(slot).getSchemaDefinitions())
             .getExecutionPayloadAndBlobsBundleSchema()
             .create(executionPayload, blobsBundle);
-    when(executionLayer.getCachedUnblindedPayload(slotAndBlockRoot))
+    when(executionLayer.getCachedUnblindedPayload(slot))
         .thenReturn(Optional.of(BuilderPayloadOrFallbackData.create(builderPayload)));
   }
 
   private void prepareCachedFallbackData(
-      final SlotAndBlockRoot slotAndBlockRoot,
-      final ExecutionPayload executionPayload,
-      final BlobsBundle blobsBundle) {
-    when(executionLayer.getCachedUnblindedPayload(slotAndBlockRoot))
+      final UInt64 slot, final ExecutionPayload executionPayload, final BlobsBundle blobsBundle) {
+    when(executionLayer.getCachedUnblindedPayload(slot))
         .thenReturn(
             Optional.of(
                 BuilderPayloadOrFallbackData.create(

beacon/validator/src/test/java/tech/pegasys/teku/validator/coordinator/ValidatorApiHandlerTest.java

@@ -24,6 +24,7 @@
 import static org.mockito.Mockito.inOrder;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.verifyNoInteractions;
 import static org.mockito.Mockito.verifyNoMoreInteractions;
@@ -543,10 +544,20 @@ public void createUnsignedBlock_shouldCreateBlock() {
     // even if passing a non-empty requestedBlinded and requestedBuilderBoostFactor isn't a valid
     // combination,
     // we still want to check that all parameters are passed down the line to the block factory
-    final SafeFuture<Optional<BlockContainerAndMetaData>> result =
+    SafeFuture<Optional<BlockContainerAndMetaData>> result =
+        validatorApiHandler.createUnsignedBlock(
+            newSlot, randaoReveal, Optional.empty(), Optional.of(ONE));
+
+    assertThat(result).isCompletedWithValue(Optional.of(blockContainerAndMetaData));
+
+    // further calls in the same slot should return the same block
+    result =
         validatorApiHandler.createUnsignedBlock(
             newSlot, randaoReveal, Optional.empty(), Optional.of(ONE));
 
+    assertThat(result).isCompletedWithValue(Optional.of(blockContainerAndMetaData));
+
+    // only produced once
     verify(blockFactory)
         .createUnsignedBlock(
             blockSlotState,
@@ -555,7 +566,51 @@ public void createUnsignedBlock_shouldCreateBlock() {
             Optional.empty(),
             Optional.of(ONE),
             BlockProductionPerformance.NOOP);
+  }
+
+  @Test
+  public void createUnsignedBlock_shouldAllowProducingBlockTwiceIfFirstAttemptFailed() {
+    final UInt64 newSlot = UInt64.valueOf(25);
+    final BeaconState blockSlotState = dataStructureUtil.randomBeaconState(newSlot);
+    final BLSSignature randaoReveal = dataStructureUtil.randomSignature();
+    final BlockContainerAndMetaData blockContainerAndMetaData =
+        dataStructureUtil.randomBlockContainerAndMetaData(newSlot);
+
+    when(chainDataClient.getStateForBlockProduction(newSlot, false))
+        .thenReturn(SafeFuture.completedFuture(Optional.of(blockSlotState)));
+    when(blockFactory.createUnsignedBlock(
+            blockSlotState,
+            newSlot,
+            randaoReveal,
+            Optional.empty(),
+            Optional.of(ONE),
+            BlockProductionPerformance.NOOP))
+        .thenThrow(new IllegalStateException("oopsy"))
+        .thenReturn(SafeFuture.completedFuture(blockContainerAndMetaData));
+
+    // first call should fail
+    SafeFuture<Optional<BlockContainerAndMetaData>> result =
+        validatorApiHandler.createUnsignedBlock(
+            newSlot, randaoReveal, Optional.empty(), Optional.of(ONE));
+
+    assertThat(result).isCompletedExceptionally();
+
+    // second call in the same slot should succeed and return the block
+    result =
+        validatorApiHandler.createUnsignedBlock(
+            newSlot, randaoReveal, Optional.empty(), Optional.of(ONE));
+
     assertThat(result).isCompletedWithValue(Optional.of(blockContainerAndMetaData));
+
+    // attempted to produce twice
+    verify(blockFactory, times(2))
+        .createUnsignedBlock(
+            blockSlotState,
+            newSlot,
+            randaoReveal,
+            Optional.empty(),
+            Optional.of(ONE),
+            BlockProductionPerformance.NOOP);
   }
 
   @Test

ethereum/executionlayer/src/main/java/tech/pegasys/teku/ethereum/executionlayer/ExecutionLayerBlockProductionManagerImpl.java

@@ -22,7 +22,6 @@
 import tech.pegasys.teku.infrastructure.async.SafeFuture;
 import tech.pegasys.teku.infrastructure.unsigned.UInt64;
 import tech.pegasys.teku.spec.datastructures.blocks.SignedBeaconBlock;
-import tech.pegasys.teku.spec.datastructures.blocks.SlotAndBlockRoot;
 import tech.pegasys.teku.spec.datastructures.execution.BuilderBidOrFallbackData;
 import tech.pegasys.teku.spec.datastructures.execution.BuilderPayloadOrFallbackData;
 import tech.pegasys.teku.spec.datastructures.execution.ExecutionPayloadContext;
@@ -41,7 +40,7 @@ public class ExecutionLayerBlockProductionManagerImpl
   private final NavigableMap<UInt64, ExecutionPayloadResult> executionResultCache =
       new ConcurrentSkipListMap<>();
 
-  private final NavigableMap<SlotAndBlockRoot, BuilderPayloadOrFallbackData> builderResultCache =
+  private final NavigableMap<UInt64, BuilderPayloadOrFallbackData> builderResultCache =
       new ConcurrentSkipListMap<>();
 
   private final ExecutionLayerChannel executionLayerChannel;
@@ -56,13 +55,9 @@ public void onSlot(final UInt64 slot) {
     executionResultCache
         .headMap(slot.minusMinZero(EXECUTION_RESULT_CACHE_RETENTION_SLOTS), false)
         .clear();
-    final UInt64 slotMax = slot.minusMinZero(BUILDER_RESULT_CACHE_RETENTION_SLOTS);
-    builderResultCache.keySet().removeIf(key -> key.getSlot().isLessThan(slotMax));
-  }
-
-  @Override
-  public Optional<ExecutionPayloadResult> getCachedPayloadResult(final UInt64 slot) {
-    return Optional.ofNullable(executionResultCache.get(slot));
+    builderResultCache
+        .headMap(slot.minusMinZero(BUILDER_RESULT_CACHE_RETENTION_SLOTS), false)
+        .clear();
   }
 
   @Override
@@ -84,6 +79,11 @@ public ExecutionPayloadResult initiateBlockProduction(
     return result;
   }
 
+  @Override
+  public Optional<ExecutionPayloadResult> getCachedPayloadResult(final UInt64 slot) {
+    return Optional.ofNullable(executionResultCache.get(slot));
+  }
+
   @Override
   public SafeFuture<BuilderPayloadOrFallbackData> getUnblindedPayload(
       final SignedBeaconBlock signedBeaconBlock,
@@ -92,15 +92,13 @@ public SafeFuture<BuilderPayloadOrFallbackData> getUnblindedPayload(
         .builderGetPayload(signedBeaconBlock, this::getCachedPayloadResult)
         .thenPeek(
             builderPayloadOrFallbackData ->
-                builderResultCache.put(
-                    signedBeaconBlock.getSlotAndBlockRoot(), builderPayloadOrFallbackData))
+                builderResultCache.put(signedBeaconBlock.getSlot(), builderPayloadOrFallbackData))
         .alwaysRun(blockPublishingPerformance::builderGetPayload);
   }
 
   @Override
-  public Optional<BuilderPayloadOrFallbackData> getCachedUnblindedPayload(
-      final SlotAndBlockRoot slotAndBlockRoot) {
-    return Optional.ofNullable(builderResultCache.get(slotAndBlockRoot));
+  public Optional<BuilderPayloadOrFallbackData> getCachedUnblindedPayload(final UInt64 slot) {
+    return Optional.ofNullable(builderResultCache.get(slot));
   }
 
   private ExecutionPayloadResult executeLocalFlow(

ethereum/spec/src/main/java/tech/pegasys/teku/spec/datastructures/blocks/BlockContainer.java

@@ -14,6 +14,7 @@
 package tech.pegasys.teku.spec.datastructures.blocks;
 
 import java.util.Optional;
+import org.apache.tuweni.bytes.Bytes32;
 import tech.pegasys.teku.infrastructure.ssz.SszContainer;
 import tech.pegasys.teku.infrastructure.ssz.SszData;
 import tech.pegasys.teku.infrastructure.ssz.SszList;
@@ -35,6 +36,10 @@ default UInt64 getSlot() {
     return getBlock().getSlot();
   }
 
+  default Bytes32 getRoot() {
+    return getBlock().getRoot();
+  }
+
   default Optional<SszList<SszKZGProof>> getKzgProofs() {
     return Optional.empty();
   }

ethereum/spec/src/main/java/tech/pegasys/teku/spec/datastructures/blocks/SignedBeaconBlock.java

@@ -137,11 +137,6 @@ public UInt64 getSlot() {
     return getMessage().getSlot();
   }
 
-  @Override
-  public SlotAndBlockRoot getSlotAndBlockRoot() {
-    return getMessage().getSlotAndBlockRoot();
-  }
-
   @Override
   public Bytes32 getParentRoot() {
     return getMessage().getParentRoot();

ethereum/spec/src/main/java/tech/pegasys/teku/spec/datastructures/blocks/SignedBlockContainer.java

@@ -40,10 +40,6 @@ default Bytes32 getRoot() {
     return getSignedBlock().getRoot();
   }
 
-  default SlotAndBlockRoot getSlotAndBlockRoot() {
-    return getSignedBlock().getSlotAndBlockRoot();
-  }
-
   default Optional<SszList<SszKZGProof>> getKzgProofs() {
     return Optional.empty();
   }