Move blobs availability check to on_execution_payload

eth-reference-tests/src/referenceTest/java/tech/pegasys/teku/reference/phase0/forkchoice/StubBlobSidecarManager.java

@@ -29,6 +29,7 @@
 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.deneb.BeaconBlockBodyDeneb;
+import tech.pegasys.teku.spec.datastructures.execution.ExecutionPayloadEnvelope;
 import tech.pegasys.teku.spec.datastructures.type.SszKZGCommitment;
 import tech.pegasys.teku.spec.logic.versions.deneb.blobs.BlobSidecarsAndValidationResult;
 import tech.pegasys.teku.spec.logic.versions.deneb.blobs.BlobSidecarsAvailabilityChecker;
@@ -125,6 +126,12 @@ private BlobSidecarsAndValidationResult validateImmediately(
     };
   }
 
+  @Override
+  public BlobSidecarsAvailabilityChecker createAvailabilityChecker(
+      final SignedBeaconBlock block, final ExecutionPayloadEnvelope executionPayloadEnvelope) {
+    throw new UnsupportedOperationException("Not yet implemented");
+  }
+
   @Override
   public BlobSidecarsAndValidationResult createAvailabilityCheckerAndValidateImmediately(
       final SignedBeaconBlock block, final List<BlobSidecar> blobSidecars) {

ethereum/statetransition/src/main/java/tech/pegasys/teku/statetransition/blobs/BlobSidecarManager.java

@@ -19,6 +19,7 @@
 import tech.pegasys.teku.infrastructure.unsigned.UInt64;
 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.logic.versions.deneb.blobs.BlobSidecarsAndValidationResult;
 import tech.pegasys.teku.spec.logic.versions.deneb.blobs.BlobSidecarsAvailabilityChecker;
 import tech.pegasys.teku.statetransition.validation.InternalValidationResult;
@@ -52,6 +53,13 @@ public BlobSidecarsAvailabilityChecker createAvailabilityChecker(
           return BlobSidecarsAvailabilityChecker.NOOP;
         }
 
+        @Override
+        public BlobSidecarsAvailabilityChecker createAvailabilityChecker(
+            final SignedBeaconBlock block,
+            final ExecutionPayloadEnvelope executionPayloadEnvelope) {
+          return BlobSidecarsAvailabilityChecker.NOOP;
+        }
+
         @Override
         public BlobSidecarsAndValidationResult createAvailabilityCheckerAndValidateImmediately(
             final SignedBeaconBlock block, final List<BlobSidecar> blobSidecars) {
@@ -70,6 +78,10 @@ SafeFuture<InternalValidationResult> validateAndPrepareForBlockImport(
 
   BlobSidecarsAvailabilityChecker createAvailabilityChecker(SignedBeaconBlock block);
 
+  // ePBS
+  BlobSidecarsAvailabilityChecker createAvailabilityChecker(
+      SignedBeaconBlock block, ExecutionPayloadEnvelope executionPayloadEnvelope);
+
   BlobSidecarsAndValidationResult createAvailabilityCheckerAndValidateImmediately(
       SignedBeaconBlock block, List<BlobSidecar> blobSidecars);
 

ethereum/statetransition/src/main/java/tech/pegasys/teku/statetransition/blobs/BlobSidecarManagerImpl.java

@@ -26,6 +26,7 @@
 import tech.pegasys.teku.spec.Spec;
 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.logic.versions.deneb.blobs.BlobSidecarsAndValidationResult;
 import tech.pegasys.teku.spec.logic.versions.deneb.blobs.BlobSidecarsAvailabilityChecker;
 import tech.pegasys.teku.statetransition.forkchoice.ForkChoiceBlobSidecarsAvailabilityChecker;
@@ -132,6 +133,17 @@ public BlobSidecarsAvailabilityChecker createAvailabilityChecker(final SignedBea
         spec, asyncRunner, recentChainData, blockBlobSidecarsTracker, kzg);
   }
 
+  @Override
+  public BlobSidecarsAvailabilityChecker createAvailabilityChecker(
+      final SignedBeaconBlock block, final ExecutionPayloadEnvelope executionPayloadEnvelope) {
+    final BlockBlobSidecarsTracker blockBlobSidecarsTracker =
+        blockBlobSidecarsTrackersPool.getOrCreateBlockBlobSidecarsTracker(
+            block, executionPayloadEnvelope);
+
+    return new ForkChoiceBlobSidecarsAvailabilityChecker(
+        spec, asyncRunner, recentChainData, blockBlobSidecarsTracker, kzg);
+  }
+
   @Override
   public BlobSidecarsAndValidationResult createAvailabilityCheckerAndValidateImmediately(
       final SignedBeaconBlock block, final List<BlobSidecar> blobSidecars) {

ethereum/statetransition/src/main/java/tech/pegasys/teku/statetransition/blobs/BlockBlobSidecarsTracker.java

@@ -36,6 +36,7 @@
 import tech.pegasys.teku.spec.datastructures.blocks.SignedBeaconBlock;
 import tech.pegasys.teku.spec.datastructures.blocks.SlotAndBlockRoot;
 import tech.pegasys.teku.spec.datastructures.blocks.blockbody.versions.deneb.BeaconBlockBodyDeneb;
+import tech.pegasys.teku.spec.datastructures.execution.ExecutionPayloadEnvelope;
 import tech.pegasys.teku.spec.datastructures.networking.libp2p.rpc.BlobIdentifier;
 import tech.pegasys.teku.statetransition.block.BlockImportChannel;
 
@@ -52,6 +53,9 @@ public class BlockBlobSidecarsTracker {
   private final AtomicReference<Optional<SignedBeaconBlock>> block =
       new AtomicReference<>(Optional.empty());
 
+  private final AtomicReference<Optional<ExecutionPayloadEnvelope>> executionPayloadEnvelope =
+      new AtomicReference<>(Optional.empty());
+
   private final AtomicBoolean blockImportOnCompletionEnabled = new AtomicBoolean(false);
 
   private final NavigableMap<UInt64, BlobSidecar> blobSidecars = new ConcurrentSkipListMap<>();
@@ -101,6 +105,10 @@ public Optional<SignedBeaconBlock> getBlock() {
     return block.get();
   }
 
+  public Optional<ExecutionPayloadEnvelope> getExecutionPayloadEnvelope() {
+    return executionPayloadEnvelope.get();
+  }
+
   public boolean containsBlobSidecar(final BlobIdentifier blobIdentifier) {
     return Optional.ofNullable(blobSidecars.get(blobIdentifier.getIndex()))
         .map(blobSidecar -> blobSidecar.getBlockRoot().equals(blobIdentifier.getBlockRoot()))
@@ -193,6 +201,29 @@ public boolean setBlock(final SignedBeaconBlock block) {
     return true;
   }
 
+  public boolean setBlockAndExecutionPayloadEnvelope(
+      final SignedBeaconBlock block, final ExecutionPayloadEnvelope executionPayloadEnvelope) {
+    checkArgument(block.getSlotAndBlockRoot().equals(slotAndBlockRoot), "Wrong block");
+    final Optional<SignedBeaconBlock> oldBlock = this.block.getAndSet(Optional.of(block));
+    final Optional<ExecutionPayloadEnvelope> oldExecutionPayloadEnvelope =
+        this.executionPayloadEnvelope.getAndSet(Optional.of(executionPayloadEnvelope));
+    if (oldBlock.isPresent() || oldExecutionPayloadEnvelope.isPresent()) {
+      return false;
+    }
+
+    LOG.debug(
+        "Block and execution payload envelope received for {}", slotAndBlockRoot::toLogString);
+    maybeDebugTimings.ifPresent(
+        debugTimings -> {
+          debugTimings.put(BLOCK_ARRIVAL_TIMING_IDX, System.currentTimeMillis());
+        });
+
+    pruneExcessiveBlobSidecars();
+    checkCompletion();
+
+    return true;
+  }
+
   public void enableBlockImportOnCompletion(final BlockImportChannel blockImportChannel) {
     final boolean alreadyEnabled = blockImportOnCompletionEnabled.getAndSet(true);
     if (alreadyEnabled) {
@@ -279,7 +310,13 @@ private Optional<Integer> getBlobKzgCommitmentsCount() {
             b ->
                 BeaconBlockBodyDeneb.required(b.getMessage().getBody())
                     .getOptionalBlobKzgCommitments()
-                    .map(SszList::size));
+                    .map(SszList::size))
+        // ePBS
+        .or(
+            () ->
+                executionPayloadEnvelope
+                    .get()
+                    .map(envelope -> envelope.getBlobKzgCommitments().size()));
   }
 
   /**

ethereum/statetransition/src/main/java/tech/pegasys/teku/statetransition/blobs/BlockBlobSidecarsTrackersPool.java

@@ -22,6 +22,7 @@
 import tech.pegasys.teku.infrastructure.unsigned.UInt64;
 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.networking.libp2p.rpc.BlobIdentifier;
 import tech.pegasys.teku.statetransition.blobs.BlobSidecarManager.RemoteOrigin;
 
@@ -73,6 +74,13 @@ public BlockBlobSidecarsTracker getOrCreateBlockBlobSidecarsTracker(
           throw new UnsupportedOperationException();
         }
 
+        @Override
+        public BlockBlobSidecarsTracker getOrCreateBlockBlobSidecarsTracker(
+            final SignedBeaconBlock block,
+            final ExecutionPayloadEnvelope executionPayloadEnvelope) {
+          throw new UnsupportedOperationException();
+        }
+
         @Override
         public Optional<BlockBlobSidecarsTracker> getBlockBlobSidecarsTracker(
             final SignedBeaconBlock block) {
@@ -127,6 +135,10 @@ public void subscribeNewBlobSidecar(NewBlobSidecarSubscriber newBlobSidecarSubsc
 
   BlockBlobSidecarsTracker getOrCreateBlockBlobSidecarsTracker(SignedBeaconBlock block);
 
+  // ePBS
+  BlockBlobSidecarsTracker getOrCreateBlockBlobSidecarsTracker(
+      SignedBeaconBlock block, ExecutionPayloadEnvelope executionPayloadEnvelope);
+
   Optional<BlockBlobSidecarsTracker> getBlockBlobSidecarsTracker(SignedBeaconBlock block);
 
   void enableBlockImportOnCompletion(SignedBeaconBlock block);

ethereum/statetransition/src/main/java/tech/pegasys/teku/statetransition/forkchoice/ForkChoice.java

@@ -555,7 +555,7 @@ private SafeFuture<Void> onExecutionPayload(
     final ForkChoicePayloadExecutorEip7732 payloadExecutor =
         ForkChoicePayloadExecutorEip7732.create(executionLayer, block.getSlot());
     final BlobSidecarsAvailabilityChecker blobSidecarsAvailabilityChecker =
-        blobSidecarManager.createAvailabilityChecker(block);
+        blobSidecarManager.createAvailabilityChecker(block, signedEnvelope.getMessage());
 
     blobSidecarsAvailabilityChecker.initiateDataAvailabilityCheck();
 

ethereum/statetransition/src/main/java/tech/pegasys/teku/statetransition/forkchoice/ForkChoiceBlobSidecarsAvailabilityChecker.java

@@ -378,7 +378,15 @@ static Supplier<List<KZGCommitment>> createLazyKzgCommitmentsSupplier(
                         .flatMap(SignedBeaconBlock::getBeaconBlock)
                         .map(BeaconBlock::getBody)
                         .orElseThrow())
-                .getBlobKzgCommitments()
+                .getOptionalBlobKzgCommitments()
+                // ePBS
+                .orElseGet(
+                    () ->
+                        availabilityChecker
+                            .blockBlobSidecarsTracker
+                            .getExecutionPayloadEnvelope()
+                            .orElseThrow()
+                            .getBlobKzgCommitments())
                 .stream()
                 .map(SszKZGCommitment::getKZGCommitment)
                 .toList());

ethereum/statetransition/src/main/java/tech/pegasys/teku/statetransition/util/BlockBlobSidecarsTrackersPoolImpl.java

@@ -53,6 +53,7 @@
 import tech.pegasys.teku.spec.datastructures.blocks.SlotAndBlockRoot;
 import tech.pegasys.teku.spec.datastructures.blocks.blockbody.versions.deneb.BeaconBlockBodyDeneb;
 import tech.pegasys.teku.spec.datastructures.execution.BlobAndProof;
+import tech.pegasys.teku.spec.datastructures.execution.ExecutionPayloadEnvelope;
 import tech.pegasys.teku.spec.datastructures.networking.libp2p.rpc.BlobIdentifier;
 import tech.pegasys.teku.spec.datastructures.type.SszKZGCommitment;
 import tech.pegasys.teku.spec.datastructures.type.SszKZGProof;
@@ -72,6 +73,7 @@ public class BlockBlobSidecarsTrackersPoolImpl extends AbstractIgnoringFutureHis
   private static final Logger LOG = LogManager.getLogger();
 
   static final String COUNTER_BLOCK_TYPE = "block";
+  static final String COUNTER_EXECUTION_PAYLOAD_ENVELOPE_TYPE = "execution_payload_envelope";
   static final String COUNTER_SIDECAR_TYPE = "blob_sidecar";
 
   static final String COUNTER_GOSSIP_SUBTYPE = "gossip";
@@ -290,6 +292,13 @@ public synchronized BlockBlobSidecarsTracker getOrCreateBlockBlobSidecarsTracker
     return internalOnNewBlock(block, Optional.empty());
   }
 
+  @Override
+  public BlockBlobSidecarsTracker getOrCreateBlockBlobSidecarsTracker(
+      final SignedBeaconBlock block, final ExecutionPayloadEnvelope executionPayloadEnvelope) {
+    return internalOnNewBlockAndExecutionPayloadEnvelope(
+        block, executionPayloadEnvelope, Optional.empty());
+  }
+
   @Override
   public synchronized Optional<BlockBlobSidecarsTracker> getBlockBlobSidecarsTracker(
       final SignedBeaconBlock block) {
@@ -511,6 +520,62 @@ private BlockBlobSidecarsTracker internalOnNewBlock(
     return tracker;
   }
 
+  private BlockBlobSidecarsTracker internalOnNewBlockAndExecutionPayloadEnvelope(
+      final SignedBeaconBlock block,
+      final ExecutionPayloadEnvelope executionPayloadEnvelope,
+      final Optional<RemoteOrigin> remoteOrigin) {
+    final SlotAndBlockRoot slotAndBlockRoot = block.getSlotAndBlockRoot();
+
+    final BlockBlobSidecarsTracker tracker =
+        getOrCreateBlobSidecarsTracker(
+            slotAndBlockRoot,
+            newTracker -> {
+              newTracker.setBlockAndExecutionPayloadEnvelope(block, executionPayloadEnvelope);
+              countBlock(remoteOrigin);
+              onFirstSeen(slotAndBlockRoot);
+            },
+            existingTracker -> {
+              if (!existingTracker.setBlockAndExecutionPayloadEnvelope(
+                  block, executionPayloadEnvelope)) {
+                // block and execution envelope were already set
+                countDuplicateBlock(remoteOrigin);
+                return;
+              }
+
+              countBlock(remoteOrigin);
+
+              if (existingTracker.isRpcFetchTriggered()) {
+                // block has been set for the first time and we previously triggered fetching of
+                // missing blobSidecars. So we may have requested to fetch more sidecars
+                // than the block actually requires. Let's drop them.
+                existingTracker
+                    .getUnusedBlobSidecarsForBlock()
+                    .forEach(
+                        blobIdentifier ->
+                            requiredBlobSidecarDroppedSubscribers.deliver(
+                                RequiredBlobSidecarDroppedSubscriber::onRequiredBlobSidecarDropped,
+                                blobIdentifier));
+
+                // if we attempted to fetch via RPC, we missed the opportunity to complete the
+                // tracker via local EL (local El fetch requires the block to be known)
+                // Let's try now
+                if (!existingTracker.isLocalElFetchTriggered() && !existingTracker.isCompleted()) {
+                  fetchMissingContentFromLocalEL(slotAndBlockRoot)
+                      .finish(
+                          error ->
+                              LOG.error(
+                                  "An occurred while attempting to fetch blobs via local EL"));
+                }
+              }
+            });
+
+    if (orderedBlobSidecarsTrackers.add(slotAndBlockRoot)) {
+      sizeGauge.set(orderedBlobSidecarsTrackers.size(), GAUGE_BLOB_SIDECARS_TRACKERS_LABEL);
+    }
+
+    return tracker;
+  }
+
   private void countBlock(final Optional<RemoteOrigin> maybeRemoteOrigin) {
     maybeRemoteOrigin.ifPresent(
         remoteOrigin -> {
@@ -571,13 +636,13 @@ private void onFirstSeen(final SlotAndBlockRoot slotAndBlockRoot) {
     asyncRunner
         .runAfterDelay(
             () ->
-                this.fetchMissingContentFromLocalEL(slotAndBlockRoot)
+                fetchMissingContentFromLocalEL(slotAndBlockRoot)
                     .handleException(
                         error ->
                             LOG.warn(
                                 "Local EL blobs lookup failed: {}",
                                 ExceptionUtils.getMessage(error)))
-                    .thenRun(() -> this.fetchMissingContentFromRemotePeers(slotAndBlockRoot)),
+                    .thenRun(() -> fetchMissingContentFromRemotePeers(slotAndBlockRoot)),
             fetchDelay)
         .finish(
             error ->
@@ -639,19 +704,26 @@ private synchronized SafeFuture<Void> fetchMissingContentFromLocalEL(
         specVersion.getSchemaDefinitions().toVersionDeneb().orElseThrow().getBlobSidecarSchema();
     final MiscHelpersDeneb miscHelpersDeneb =
         specVersion.miscHelpers().toVersionDeneb().orElseThrow();
-    final SignedBeaconBlockHeader signedBeaconBlockHeader =
-        blockBlobSidecarsTracker.getBlock().get().asHeader();
-    final BeaconBlockBodyDeneb beaconBlockBodyDeneb =
+    final BeaconBlockBodyDeneb beaconBlockBody =
         blockBlobSidecarsTracker
             .getBlock()
             .get()
             .getMessage()
             .getBody()
             .toVersionDeneb()
             .orElseThrow();
+    final SignedBeaconBlockHeader signedBeaconBlockHeader =
+        blockBlobSidecarsTracker.getBlock().get().asHeader();
 
     final SszList<SszKZGCommitment> sszKZGCommitments =
-        beaconBlockBodyDeneb.getBlobKzgCommitments();
+        beaconBlockBody
+            .getOptionalBlobKzgCommitments()
+            .orElseGet(
+                () ->
+                    blockBlobSidecarsTracker
+                        .getExecutionPayloadEnvelope()
+                        .orElseThrow()
+                        .getBlobKzgCommitments());
 
     final List<BlobIdentifier> missingBlobsIdentifiers =
         blockBlobSidecarsTracker.getMissingBlobSidecars().toList();
@@ -687,13 +759,15 @@ private synchronized SafeFuture<Void> fetchMissingContentFromLocalEL(
                   continue;
                 }
 
+                final BlobIdentifier blobIdentifier = missingBlobsIdentifiers.get(index);
                 final BlobSidecar blobSidecar =
                     createBlobSidecarFromBlobAndProof(
                         blobSidecarSchema,
                         miscHelpersDeneb,
-                        missingBlobsIdentifiers.get(index),
+                        blobIdentifier,
                         blobAndProof.get(),
-                        beaconBlockBodyDeneb,
+                        sszKZGCommitments.get(blobIdentifier.getIndex().intValue()),
+                        beaconBlockBody,
                         signedBeaconBlockHeader);
 
                 onNewBlobSidecar(blobSidecar, LOCAL_EL);
@@ -706,12 +780,10 @@ private BlobSidecar createBlobSidecarFromBlobAndProof(
       final MiscHelpersDeneb miscHelpersDeneb,
       final BlobIdentifier blobIdentifier,
       final BlobAndProof blobAndProof,
+      final SszKZGCommitment sszKZGCommitment,
       final BeaconBlockBodyDeneb beaconBlockBodyDeneb,
       final SignedBeaconBlockHeader signedBeaconBlockHeader) {
 
-    final SszKZGCommitment sszKZGCommitment =
-        beaconBlockBodyDeneb.getBlobKzgCommitments().get(blobIdentifier.getIndex().intValue());
-
     return blobSidecarSchema.create(
         blobIdentifier.getIndex(),
         blobAndProof.blob(),