[ePBS] Block production (part 2) (Consensys#8598)

beacon/sync/src/test/java/tech/pegasys/teku/beacon/sync/historical/HistoricalBatchFetcherTest.java

@@ -148,7 +148,8 @@ public void setup() {
 
     when(signatureVerifier.verify(any(), any(), anyList()))
         .thenReturn(SafeFuture.completedFuture(true));
-    when(blobSidecarManager.createAvailabilityCheckerAndValidateImmediately(any(), anyList()))
+    when(blobSidecarManager.createAvailabilityCheckerAndValidateImmediately(
+            any(SignedBeaconBlock.class), anyList()))
         .thenAnswer(i -> BlobSidecarsAndValidationResult.validResult(i.getArgument(1)));
   }
 
@@ -208,7 +209,8 @@ public void run_returnAllBlocksAndBlobSidecarsOnFirstRequest() {
   @Test
   public void run_failsOnBlobSidecarsValidationFailure() {
     when(blobSidecarManager.isAvailabilityRequiredAtSlot(any())).thenReturn(true);
-    when(blobSidecarManager.createAvailabilityCheckerAndValidateImmediately(any(), anyList()))
+    when(blobSidecarManager.createAvailabilityCheckerAndValidateImmediately(
+            any(SignedBeaconBlock.class), anyList()))
         .thenAnswer(
             i ->
                 BlobSidecarsAndValidationResult.invalidResult(

beacon/validator/src/integrationTest/java/tech/pegasys/teku/validator/coordinator/ValidatorApiHandlerIntegrationTest.java

@@ -112,6 +112,8 @@ public class ValidatorApiHandlerIntegrationTest {
       mock(ExecutionPayloadHeaderPool.class);
   private final ExecutionPayloadManager executionPayloadManager =
       mock(ExecutionPayloadManager.class);
+  private final ExecutionPayloadAndBlobSidecarsRevealer executionPayloadAndBlobSidecarsRevealer =
+      mock(ExecutionPayloadAndBlobSidecarsRevealer.class);
 
   private final DutyMetrics dutyMetrics = mock(DutyMetrics.class);
   private final ValidatorApiHandler handler =
@@ -143,6 +145,7 @@ public class ValidatorApiHandlerIntegrationTest {
           syncCommitteeSubscriptionManager,
           executionPayloadHeaderPool,
           executionPayloadManager,
+          executionPayloadAndBlobSidecarsRevealer,
           new BlockProductionAndPublishingPerformanceFactory(
               new SystemTimeProvider(), __ -> UInt64.ZERO, true, 0, 0, 0, 0));
 

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

@@ -210,7 +210,8 @@ public Function<BeaconBlockBodyBuilder, SafeFuture<Void>> createSelector(
       // In `setExecutionData` the following fields are set:
       // Post-Bellatrix: Execution Payload / Execution Payload Header
       // Post-Deneb: KZG Commitments
-      // in ePBS, this section is skipped entirely because the bid is already available
+      // in ePBS, this section is skipped entirely because the bid is already available and local
+      // EL/builder have been called
       if (bodyBuilder.supportsExecutionPayload()) {
         final SchemaDefinitionsBellatrix schemaDefinitions =
             SchemaDefinitionsBellatrix.required(

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

@@ -0,0 +1,135 @@
+/*
+ * Copyright Consensys Software Inc., 2024
+ *
+ * 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.teku.validator.coordinator;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.stream.IntStream;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+import org.apache.tuweni.bytes.Bytes32;
+import tech.pegasys.teku.infrastructure.ssz.SszList;
+import tech.pegasys.teku.infrastructure.unsigned.UInt64;
+import tech.pegasys.teku.kzg.KZGProof;
+import tech.pegasys.teku.spec.Spec;
+import tech.pegasys.teku.spec.datastructures.blobs.versions.deneb.Blob;
+import tech.pegasys.teku.spec.datastructures.blobs.versions.deneb.BlobSidecar;
+import tech.pegasys.teku.spec.datastructures.blocks.SignedBeaconBlock;
+import tech.pegasys.teku.spec.datastructures.blocks.blockbody.versions.eip7732.BeaconBlockBodyEip7732;
+import tech.pegasys.teku.spec.datastructures.execution.BlobsBundle;
+import tech.pegasys.teku.spec.datastructures.execution.ExecutionPayloadEnvelope;
+import tech.pegasys.teku.spec.datastructures.execution.ExecutionPayloadHeader;
+import tech.pegasys.teku.spec.datastructures.execution.GetPayloadResponse;
+import tech.pegasys.teku.spec.datastructures.execution.versions.eip7732.ExecutionPayloadHeaderEip7732;
+import tech.pegasys.teku.spec.datastructures.state.beaconstate.BeaconState;
+import tech.pegasys.teku.spec.datastructures.type.SszKZGCommitment;
+import tech.pegasys.teku.spec.datastructures.type.SszKZGProof;
+import tech.pegasys.teku.spec.logic.versions.deneb.helpers.MiscHelpersDeneb;
+import tech.pegasys.teku.spec.schemas.SchemaDefinitionsEip7732;
+
+public class CachingExecutionPayloadAndBlobSidecarsRevealer
+    implements ExecutionPayloadAndBlobSidecarsRevealer {
+
+  private static final Logger LOG = LogManager.getLogger();
+
+  // EIP7732 TODO: should probably use more optimized data structure
+  private final Map<Bytes32, GetPayloadResponse> committedPayloads = new ConcurrentHashMap<>();
+
+  private final Spec spec;
+
+  public CachingExecutionPayloadAndBlobSidecarsRevealer(final Spec spec) {
+    this.spec = spec;
+  }
+
+  @Override
+  public void commit(
+      final ExecutionPayloadHeader header, final GetPayloadResponse getPayloadResponse) {
+    committedPayloads.put(header.hashTreeRoot(), getPayloadResponse);
+  }
+
+  @Override
+  public Optional<ExecutionPayloadEnvelope> revealExecutionPayload(
+      final SignedBeaconBlock block, final BeaconState state) {
+    final ExecutionPayloadHeaderEip7732 committedHeader =
+        ExecutionPayloadHeaderEip7732.required(
+            BeaconBlockBodyEip7732.required(block.getMessage().getBody())
+                .getSignedExecutionPayloadHeader()
+                .getMessage());
+    final GetPayloadResponse getPayloadResponse =
+        committedPayloads.get(committedHeader.hashTreeRoot());
+    if (getPayloadResponse == null) {
+      logMissingPayload(block);
+      return Optional.empty();
+    }
+    final SchemaDefinitionsEip7732 schemaDefinitions =
+        SchemaDefinitionsEip7732.required(spec.atSlot(block.getSlot()).getSchemaDefinitions());
+    final SszList<SszKZGCommitment> blobKzgCommitments =
+        schemaDefinitions
+            .getBlobKzgCommitmentsSchema()
+            .createFromBlobsBundle(getPayloadResponse.getBlobsBundle().orElseThrow());
+    final ExecutionPayloadEnvelope executionPayload =
+        schemaDefinitions
+            .getExecutionPayloadEnvelopeSchema()
+            .create(
+                getPayloadResponse.getExecutionPayload(),
+                committedHeader.getBuilderIndex(),
+                block.getRoot(),
+                blobKzgCommitments,
+                false,
+                state.hashTreeRoot());
+    return Optional.of(executionPayload);
+  }
+
+  @Override
+  public List<BlobSidecar> revealBlobSidecars(final SignedBeaconBlock block) {
+    final Bytes32 committedHeaderRoot =
+        BeaconBlockBodyEip7732.required(block.getMessage().getBody())
+            .getSignedExecutionPayloadHeader()
+            .getMessage()
+            .hashTreeRoot();
+    // clean up the cached payload since it would be no longer used
+    final GetPayloadResponse getPayloadResponse = committedPayloads.remove(committedHeaderRoot);
+    if (getPayloadResponse == null) {
+      logMissingPayload(block);
+      return List.of();
+    }
+    final MiscHelpersDeneb miscHelpersDeneb =
+        MiscHelpersDeneb.required(spec.atSlot(block.getSlot()).miscHelpers());
+
+    final BlobsBundle blobsBundle = getPayloadResponse.getBlobsBundle().orElseThrow();
+
+    final List<Blob> blobs = blobsBundle.getBlobs();
+    final List<KZGProof> proofs = blobsBundle.getProofs();
+
+    // EIP7732 TODO: need to modify the constructBlobSidecar method in ePBS
+    return IntStream.range(0, blobsBundle.getNumberOfBlobs())
+        .mapToObj(
+            index ->
+                miscHelpersDeneb.constructBlobSidecar(
+                    block,
+                    UInt64.valueOf(index),
+                    blobs.get(index),
+                    new SszKZGProof(proofs.get(index))))
+        .toList();
+  }
+
+  private void logMissingPayload(final SignedBeaconBlock block) {
+    LOG.error(
+        "There is no committed payload for block with slot {} and root {}",
+        block.getSlot(),
+        block.getRoot());
+  }
+}

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

@@ -0,0 +1,36 @@
+/*
+ * Copyright Consensys Software Inc., 2024
+ *
+ * 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.teku.validator.coordinator;
+
+import java.util.List;
+import java.util.Optional;
+import tech.pegasys.teku.spec.datastructures.blobs.versions.deneb.BlobSidecar;
+import tech.pegasys.teku.spec.datastructures.blocks.SignedBeaconBlock;
+import tech.pegasys.teku.spec.datastructures.execution.ExecutionPayloadEnvelope;
+import tech.pegasys.teku.spec.datastructures.execution.ExecutionPayloadHeader;
+import tech.pegasys.teku.spec.datastructures.execution.GetPayloadResponse;
+import tech.pegasys.teku.spec.datastructures.state.beaconstate.BeaconState;
+
+public interface ExecutionPayloadAndBlobSidecarsRevealer {
+
+  /** 1. builder commits to a header */
+  void commit(ExecutionPayloadHeader header, GetPayloadResponse getPayloadResponse);
+
+  /** 2. builder reveals the payload */
+  Optional<ExecutionPayloadEnvelope> revealExecutionPayload(
+      SignedBeaconBlock block, BeaconState state);
+
+  /** 3. builder reveals the blob sidecars */
+  List<BlobSidecar> revealBlobSidecars(SignedBeaconBlock block);
+}

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

@@ -37,14 +37,17 @@ public class ExecutionPayloadHeaderFactory {
   private final Spec spec;
   private final ForkChoiceNotifier forkChoiceNotifier;
   private final ExecutionLayerBlockProductionManager executionLayerBlockProductionManager;
+  private final ExecutionPayloadAndBlobSidecarsRevealer executionPayloadAndBlobSidecarsRevealer;
 
   public ExecutionPayloadHeaderFactory(
       final Spec spec,
       final ForkChoiceNotifier forkChoiceNotifier,
-      final ExecutionLayerBlockProductionManager executionLayerBlockProductionManager) {
+      final ExecutionLayerBlockProductionManager executionLayerBlockProductionManager,
+      final ExecutionPayloadAndBlobSidecarsRevealer executionPayloadAndBlobSidecarsRevealer) {
     this.spec = spec;
     this.forkChoiceNotifier = forkChoiceNotifier;
     this.executionLayerBlockProductionManager = executionLayerBlockProductionManager;
+    this.executionPayloadAndBlobSidecarsRevealer = executionPayloadAndBlobSidecarsRevealer;
   }
 
   public SafeFuture<ExecutionPayloadHeader> createUnsignedHeader(
@@ -80,13 +83,18 @@ public SafeFuture<ExecutionPayloadHeader> createUnsignedHeader(
               return executionPayloadResult
                   .getLocalPayloadResponseRequired()
                   .thenApply(
-                      getPayloadResponse ->
-                          createLocalBid(
-                              slot,
-                              builderIndex,
-                              parentRoot,
-                              executionPayloadContext,
-                              getPayloadResponse));
+                      getPayloadResponse -> {
+                        final ExecutionPayloadHeader localBid =
+                            createLocalBid(
+                                slot,
+                                builderIndex,
+                                parentRoot,
+                                executionPayloadContext,
+                                getPayloadResponse);
+                        executionPayloadAndBlobSidecarsRevealer.commit(
+                            localBid, getPayloadResponse);
+                        return localBid;
+                      });
             });
   }
 

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

@@ -162,6 +162,7 @@ public class ValidatorApiHandler implements ValidatorApiChannel {
   private final ExecutionPayloadHeaderPool executionPayloadHeaderPool;
   private final ProposersDataManager proposersDataManager;
   private final BlockPublisher blockPublisher;
+  private final ExecutionPayloadAndBlobSidecarsRevealer executionPayloadAndBlobSidecarsRevealer;
   private final ExecutionPayloadPublisher executionPayloadPublisher;
   private final AttesterDutiesGenerator attesterDutiesGenerator;
 
@@ -193,6 +194,7 @@ public ValidatorApiHandler(
       final SyncCommitteeSubscriptionManager syncCommitteeSubscriptionManager,
       final ExecutionPayloadHeaderPool executionPayloadHeaderPool,
       final ExecutionPayloadManager executionPayloadManager,
+      final ExecutionPayloadAndBlobSidecarsRevealer executionPayloadAndBlobSidecarsRevealer,
       final BlockProductionAndPublishingPerformanceFactory
           blockProductionAndPublishingPerformanceFactory) {
     this.blockProductionAndPublishingPerformanceFactory =
@@ -229,7 +231,13 @@ public ValidatorApiHandler(
             performanceTracker,
             dutyMetrics);
     this.executionPayloadPublisher =
-        new ExecutionPayloadPublisher(executionPayloadManager, executionPayloadGossipChannel);
+        new ExecutionPayloadPublisher(
+            executionPayloadManager,
+            blockBlobSidecarsTrackersPool,
+            executionPayloadGossipChannel,
+            executionPayloadAndBlobSidecarsRevealer,
+            blobSidecarGossipChannel);
+    this.executionPayloadAndBlobSidecarsRevealer = executionPayloadAndBlobSidecarsRevealer;
     this.attesterDutiesGenerator = new AttesterDutiesGenerator(spec);
   }
 
@@ -661,7 +669,17 @@ public SafeFuture<Optional<SyncCommitteeContribution>> createSyncCommitteeContri
   @Override
   public SafeFuture<Optional<ExecutionPayloadEnvelope>> getExecutionPayloadEnvelope(
       final UInt64 slot, final BLSPublicKey builderPublicKey) {
-    throw new UnsupportedOperationException("This method is not implemented by the Beacon Node");
+    return combinedChainDataClient
+        .getBlockAtSlotExact(slot)
+        .thenCombine(
+            combinedChainDataClient.getStateAtSlotExact(slot),
+            (maybeBlock, maybeState) -> {
+              if (maybeBlock.isEmpty() || maybeState.isEmpty()) {
+                return Optional.empty();
+              }
+              return executionPayloadAndBlobSidecarsRevealer.revealExecutionPayload(
+                  maybeBlock.get(), maybeState.get());
+            });
   }
 
   @Override
@@ -834,8 +852,19 @@ public SafeFuture<SendSignedBlockResult> sendSignedBlock(
   @Override
   public SafeFuture<Void> sendSignedExecutionPayloadEnvelope(
       final SignedExecutionPayloadEnvelope signedExecutionPayloadEnvelope) {
-    return executionPayloadPublisher
-        .sendExecutionPayload(signedExecutionPayloadEnvelope)
+    final Bytes32 blockRoot = signedExecutionPayloadEnvelope.getMessage().getBeaconBlockRoot();
+    return combinedChainDataClient
+        .getBlockByBlockRoot(blockRoot)
+        .thenCompose(
+            maybeBlock -> {
+              if (maybeBlock.isEmpty()) {
+                return SafeFuture.failedFuture(
+                    new IllegalArgumentException(
+                        "There is no block available with root " + blockRoot));
+              }
+              return executionPayloadPublisher.sendExecutionPayload(
+                  maybeBlock.get(), signedExecutionPayloadEnvelope);
+            })
         .thenAccept(
             result -> {
               if (result.isNotProcessable()) {