feat: add gloas execution payload envelope import pipeline

packages/api/src/beacon/routes/events.ts

@@ -88,6 +88,8 @@ export enum EventType {
   blobSidecar = "blob_sidecar",
   /** The node has received a valid DataColumnSidecar (from P2P or API) */
   dataColumnSidecar = "data_column_sidecar",
+  /** The node has verified that the execution payload and blobs for a block are available */
+  executionPayloadAvailable = "execution_payload_available",
 }
 
 export const eventTypes: {[K in EventType]: K} = {
@@ -108,6 +110,7 @@ export const eventTypes: {[K in EventType]: K} = {
   [EventType.payloadAttributes]: EventType.payloadAttributes,
   [EventType.blobSidecar]: EventType.blobSidecar,
   [EventType.dataColumnSidecar]: EventType.dataColumnSidecar,
+  [EventType.executionPayloadAvailable]: EventType.executionPayloadAvailable,
 };
 
 export type EventData = {
@@ -157,6 +160,10 @@ export type EventData = {
   [EventType.payloadAttributes]: {version: ForkName; data: SSEPayloadAttributes};
   [EventType.blobSidecar]: BlobSidecarSSE;
   [EventType.dataColumnSidecar]: DataColumnSidecarSSE;
+  [EventType.executionPayloadAvailable]: {
+    slot: Slot;
+    blockRoot: RootHex;
+  };
 };
 
 export type BeaconEvent = {[K in EventType]: {type: K; message: EventData[K]}}[EventType];
@@ -311,6 +318,13 @@ export function getTypeByEvent(config: ChainForkConfig): {[K in EventType]: Type
     [EventType.payloadAttributes]: WithVersion((fork) => getPostBellatrixForkTypes(fork).SSEPayloadAttributes),
     [EventType.blobSidecar]: blobSidecarSSE,
     [EventType.dataColumnSidecar]: dataColumnSidecarSSE,
+    [EventType.executionPayloadAvailable]: new ContainerType(
+      {
+        slot: ssz.Slot,
+        blockRoot: stringType,
+      },
+      {jsonCase: "eth2"}
+    ),
 
     [EventType.lightClientOptimisticUpdate]: WithVersion(
       (fork) => getPostAltairForkTypes(fork).LightClientOptimisticUpdate

packages/api/test/unit/beacon/oapiSpec.test.ts

@@ -78,7 +78,6 @@ runTestCheckAgainstSpec(openApiJson, definitions, testDatas, ignoredOperations,
 
 const ignoredTopics: string[] = [
   // TODO GLOAS: required by v5.0.0-alpha.0
-  "execution_payload_available",
   "execution_payload_bid",
   "payload_attestation_message",
 ];

packages/api/test/unit/beacon/testData/events.ts

@@ -274,4 +274,8 @@ export const eventTestData: EventData = {
       "0x1b66ac1fb663c9bc59509846d6ec05345bd908eda73e670af888da41af171505cc411d61252fb6cb3fa0017b679f8bb2305b26a285fa2737f175668d0dff91cc1b66ac1fb663c9bc59509846d6ec05345bd908eda73e670af888da41af171505",
     ],
   }),
+  [EventType.executionPayloadAvailable]: {
+    slot: 10,
+    blockRoot: "0x9a2fefd2fdb57f74993c7780ea5b9030d2897b615b89f808011ca5aebed54eaf",
+  },
 };

packages/beacon-node/src/api/impl/beacon/blocks/index.ts

@@ -35,6 +35,7 @@ import {
 } from "@lodestar/types";
 import {fromHex, sleep, toHex, toRootHex} from "@lodestar/utils";
 import {BlockInputSource, isBlockInputBlobs, isBlockInputColumns} from "../../../../chain/blocks/blockInput/index.js";
+import {PayloadEnvelopeInputSource} from "../../../../chain/blocks/payloadEnvelopeInput/index.js";
 import {ImportBlockOpts} from "../../../../chain/blocks/types.js";
 import {verifyBlocksInEpoch} from "../../../../chain/blocks/verifyBlock.js";
 import {BeaconChain} from "../../../../chain/chain.js";
@@ -48,6 +49,7 @@ import {
   ProduceFullGloas,
 } from "../../../../chain/produceBlock/index.js";
 import {validateGossipBlock} from "../../../../chain/validation/block.js";
+import {validateApiExecutionPayloadEnvelope} from "../../../../chain/validation/executionPayloadEnvelope.js";
 import {OpSource} from "../../../../chain/validatorMonitor.js";
 import {
   computePreFuluKzgCommitmentsInclusionProof,
@@ -659,6 +661,8 @@ export function getBeaconBlockApi({
         throw new ApiError(400, `Envelope slot ${slot} does not match block slot ${block.slot}`);
       }
 
+      await validateApiExecutionPayloadEnvelope(chain, signedExecutionPayloadEnvelope);
+
       const isSelfBuild = envelope.builderIndex === BUILDER_INDEX_SELF_BUILD;
       let dataColumnSidecars: gloas.DataColumnSidecars = [];
 
@@ -676,6 +680,7 @@ export function getBeaconBlockApi({
         }
 
         if (cachedResult.cells && cachedResult.blobsBundle.commitments.length > 0) {
+          const timer = metrics?.peerDas.dataColumnSidecarComputationTime.startTimer();
           const cellsAndProofs = cachedResult.cells.map((rowCells, rowIndex) => ({
             cells: rowCells,
             proofs: cachedResult.blobsBundle.proofs.slice(
@@ -685,24 +690,42 @@ export function getBeaconBlockApi({
           }));
 
           dataColumnSidecars = getDataColumnSidecarsForGloas(slot, envelope.beaconBlockRoot, cellsAndProofs);
+          timer?.();
         }
       } else {
         // TODO GLOAS: will this api be used by builders or only for self-building?
       }
 
-      // TODO GLOAS: Verify execution payload envelope signature
-      // For self-builds, the proposer signs with their own validator key
-      // For external builders, verify using the builder's registered pubkey
-      // Use verify_execution_payload_envelope_signature(state, signed_envelope)
+      // If called near a slot boundary (e.g. late in slot N-1), hold briefly so gossip aligns with slot N.
+      const msToBlockSlot = computeTimeAtSlot(config, slot, chain.genesisTime) * 1000 - Date.now();
+      if (msToBlockSlot <= MAX_API_CLOCK_DISPARITY_MS && msToBlockSlot > 0) {
+        await sleep(msToBlockSlot);
+      }
 
-      // TODO GLOAS: Process execution payload via state transition
-      // Call process_execution_payload(state, signed_envelope, execution_engine)
+      // TODO GLOAS: if block and payload are submitted in parallel, payloadInput may not yet exist.
+      // A queuing mechanism is needed to handle this case. See https://github.com/ChainSafe/lodestar/issues/8915
+      const payloadInput = chain.seenPayloadEnvelopeInputCache.get(blockRootHex);
+      if (!payloadInput) {
+        throw new ApiError(404, `PayloadEnvelopeInput not found for block root ${blockRootHex}`);
+      }
 
-      // TODO GLOAS: Update fork choice with the execution payload
-      // Call on_execution_payload(store, signed_envelope) to update fork choice state
+      payloadInput.addPayloadEnvelope({
+        envelope: signedExecutionPayloadEnvelope,
+        source: PayloadEnvelopeInputSource.api,
+        seenTimestampSec,
+        peerIdStr: undefined,
+      });
 
-      // TODO GLOAS: Add envelope and data columns to block input via seenBlockInputCache
-      // and trigger block import (Gloas block import requires both beacon block and envelope)
+      if (dataColumnSidecars.length > 0) {
+        for (const columnSidecar of dataColumnSidecars) {
+          payloadInput.addColumn({
+            columnSidecar,
+            source: PayloadEnvelopeInputSource.api,
+            seenTimestampSec,
+            peerIdStr: undefined,
+          });
+        }
+      }
 
       const valLogMeta = {
         slot,
@@ -712,23 +735,19 @@ export function getBeaconBlockApi({
         dataColumns: dataColumnSidecars.length,
       };
 
-      // If called near a slot boundary (e.g. late in slot N-1), hold briefly so gossip aligns with slot N.
-      const msToBlockSlot = computeTimeAtSlot(config, slot, chain.genesisTime) * 1000 - Date.now();
-      if (msToBlockSlot <= MAX_API_CLOCK_DISPARITY_MS && msToBlockSlot > 0) {
-        await sleep(msToBlockSlot);
-      }
-
       const delaySec = seenTimestampSec - computeTimeAtSlot(config, slot, chain.genesisTime);
       metrics?.gossipExecutionPayloadEnvelope.elapsedTimeTillReceived.observe({source: OpSource.api}, delaySec);
+      chain.validatorMonitor?.registerExecutionPayloadEnvelope(OpSource.api, delaySec, signedExecutionPayloadEnvelope);
 
       chain.logger.info("Publishing execution payload envelope", valLogMeta);
 
-      // Publish envelope and data columns
       const publishPromises = [
         // Gossip the signed execution payload envelope first
         () => network.publishSignedExecutionPayloadEnvelope(signedExecutionPayloadEnvelope),
         // For self-builds, publish all data column sidecars
         ...dataColumnSidecars.map((dataColumnSidecar) => () => network.publishDataColumnSidecar(dataColumnSidecar)),
+        // Import execution payload. Signature already verified above
+        () => chain.processExecutionPayload(payloadInput, {validSignature: true}),
       ];
 
       const sentPeersArr = await promiseAllMaybeAsync<number | void>(publishPromises);

packages/beacon-node/src/chain/blocks/blockInput/types.ts

@@ -1,5 +1,5 @@
 import {ForkName} from "@lodestar/params";
-import {ColumnIndex, RootHex, SignedBeaconBlock, Slot, deneb, fulu} from "@lodestar/types";
+import {ColumnIndex, DataColumnSidecars, RootHex, SignedBeaconBlock, Slot, deneb, fulu} from "@lodestar/types";
 import {VersionedHashes} from "../../../execution/index.js";
 
 export enum DAType {
@@ -107,9 +107,9 @@ export type MissingColumnMeta = {
 export interface IDataColumnsInput {
   readonly slot: Slot;
   readonly blockRootHex: string;
-  getCustodyColumns(): fulu.DataColumnSidecars;
+  getCustodyColumns(): DataColumnSidecars;
   hasComputedAllData(): boolean;
-  waitForComputedAllData(timeout: number, signal?: AbortSignal): Promise<fulu.DataColumnSidecars>;
+  waitForComputedAllData(timeout: number, signal?: AbortSignal): Promise<DataColumnSidecars>;
 }
 
 /**

packages/beacon-node/src/chain/blocks/importBlock.ts

@@ -9,7 +9,14 @@ import {
   getSafeExecutionBlockHash,
   isGloasBlock,
 } from "@lodestar/fork-choice";
-import {ForkPostAltair, ForkPostElectra, ForkSeq, MAX_SEED_LOOKAHEAD, SLOTS_PER_EPOCH} from "@lodestar/params";
+import {
+  ForkPostAltair,
+  ForkPostElectra,
+  ForkPostGloas,
+  ForkSeq,
+  MAX_SEED_LOOKAHEAD,
+  SLOTS_PER_EPOCH,
+} from "@lodestar/params";
 import {
   CachedBeaconStateAltair,
   EpochCache,
@@ -21,7 +28,17 @@ import {
   isStartSlotOfEpoch,
   isStateValidatorsNodesPopulated,
 } from "@lodestar/state-transition";
-import {Attestation, BeaconBlock, altair, capella, electra, isGloasBeaconBlock, phase0, ssz} from "@lodestar/types";
+import {
+  Attestation,
+  BeaconBlock,
+  SignedBeaconBlock,
+  altair,
+  capella,
+  electra,
+  isGloasBeaconBlock,
+  phase0,
+  ssz,
+} from "@lodestar/types";
 import {isErrorAborted, toRootHex} from "@lodestar/utils";
 import {ZERO_HASH_HEX} from "../../constants/index.js";
 import {callInNextEventLoop} from "../../util/eventLoop.js";
@@ -123,6 +140,23 @@ export async function importBlock(
   // processState manages both block state and payload state variants together for memory/disk management
   this.regen.processBlockState(blockRootHex, postState);
 
+  // For Gloas blocks, create PayloadEnvelopeInput so it's available for later payload import
+  if (fork >= ForkSeq.gloas) {
+    this.seenPayloadEnvelopeInputCache.add({
+      blockRootHex,
+      block: block as SignedBeaconBlock<ForkPostGloas>,
+      sampledColumns: this.custodyConfig.sampledColumns,
+      custodyColumns: this.custodyConfig.custodyColumns,
+      timeCreatedSec: fullyVerifiedBlock.seenTimestampSec,
+    });
+    this.logger.debug("Created PayloadEnvelopeInput for block", {
+      slot: blockSlot,
+      root: blockRootHex,
+      source: source.source,
+      ...(opts.seenTimestampSec !== undefined ? {recvToImport: Date.now() / 1000 - opts.seenTimestampSec} : {}),
+    });
+  }
+
   this.metrics?.importBlock.bySource.inc({source: source.source});
   this.logger.verbose("Added block to forkchoice and state cache", {slot: blockSlot, root: blockRootHex});