feat: add engine_getBlobsV2 execution API (#7675)

**Motivation**

add `engine_getBlobsV2` to the execution API in preparation for
implementation of [distributed blob
publishing](https://github.com/ethereum/consensus-specs/blob/dev/specs/fulu/p2p-interface.md#distributed-blob-publishing-using-blobs-retrieved-from-local-execution-layer-client)

@dguenther and I wanted to get early feedback on the API change before
moving forward with the rest of the implementation

**Description**

upcoming spec changes will add `engine_getBlobsV2` to the execution API
to fetch blobs and cell proofs from the execution layer
(ethereum/execution-apis#630)

* add `engine_getBlobsV2` to execution API
* add type definition for `BlobAndProofV2`

**Not included**

We'll follow up with additional PR(s) for these as we move forward with
distributed blob publishing:

* fetch blobs from the EL in two places: on first seen block input
gossip and on unknown blocks during syncing
* reconstruct blobs from cell proofs
* publish data column sidecars on subscribed topics after

Relates to #7638

---------

Co-authored-by: Derek Guenther <[email protected]>

Author: hughy

packages/beacon-node/src/execution/engine/http.ts

@@ -1,7 +1,8 @@
 import {Logger} from "@lodestar/logger";
-import {ForkName, ForkSeq, SLOTS_PER_EPOCH} from "@lodestar/params";
+import {ForkName, ForkPostFulu, ForkPreFulu, ForkSeq, SLOTS_PER_EPOCH, isForkPostFulu} from "@lodestar/params";
 import {ExecutionPayload, ExecutionRequests, Root, RootHex, Wei} from "@lodestar/types";
 import {BlobAndProof} from "@lodestar/types/deneb";
+import {BlobAndProofV2} from "@lodestar/types/fulu";
 import {strip0xPrefix} from "@lodestar/utils";
 import {
   ErrorJsonRpcResponse,
@@ -35,6 +36,7 @@ import {
   ExecutionPayloadBody,
   assertReqSizeLimit,
   deserializeBlobAndProofs,
+  deserializeBlobAndProofsV2,
   deserializeExecutionPayloadBody,
   parseExecutionPayload,
   serializeBeaconBlockRoot,
@@ -99,6 +101,13 @@ export const defaultExecutionEngineHttpOpts: ExecutionEngineHttpOpts = {
  */
 const QUEUE_MAX_LENGTH = EPOCHS_PER_BATCH * SLOTS_PER_EPOCH * 2;
 
+/**
+ * Maximum number of version hashes that can be sent in a getBlobs request
+ * Clients must support at least 128 versionedHashes, so we avoid sending more
+ * https://github.com/ethereum/execution-apis/blob/main/src/engine/cancun.md#specification-3
+ */
+const MAX_VERSIONED_HASHES = 128;
+
 // Define static options once to prevent extra allocations
 const notifyNewPayloadOpts: ReqOpts = {routeId: "notifyNewPayload"};
 const forkchoiceUpdatedV1Opts: ReqOpts = {routeId: "forkchoiceUpdated"};
@@ -115,7 +124,7 @@ const getPayloadOpts: ReqOpts = {routeId: "getPayload"};
  */
 export class ExecutionEngineHttp implements IExecutionEngine {
   private logger: Logger;
-  private lastGetBlobsErrorTime = 0;
+  private lastGetBlobsV1ErrorTime = 0;
 
   // The default state is ONLINE, it will be updated to SYNCING once we receive the first payload
   // This assumption is better than the OFFLINE state, since we can't be sure if the EL is offline and being offline may trigger some notifications
@@ -466,55 +475,79 @@ export class ExecutionEngineHttp implements IExecutionEngine {
     return response.map(deserializeExecutionPayloadBody);
   }
 
-  async getBlobs(_fork: ForkName, versionedHashes: VersionedHashes): Promise<(BlobAndProof | null)[]> {
+  async getBlobs(fork: ForkPostFulu, versionedHashes: VersionedHashes): Promise<BlobAndProofV2[] | null>;
+  async getBlobs(fork: ForkPreFulu, versionedHashes: VersionedHashes): Promise<(BlobAndProof | null)[]>;
+  async getBlobs(
+    fork: ForkName,
+    versionedHashes: VersionedHashes
+  ): Promise<BlobAndProofV2[] | (BlobAndProof | null)[] | null> {
+    const method = isForkPostFulu(fork) ? "engine_getBlobsV2" : "engine_getBlobsV1";
+
+    // engine_getBlobsV2 is mandatory, but engine_getBlobsV1 is optional
+    const timeNow = Date.now() / 1000;
     // retry only after a day may be
     const GETBLOBS_RETRY_TIMEOUT = 256 * 32 * 12;
-    const timeNow = Date.now() / 1000;
-    const timeSinceLastFail = timeNow - this.lastGetBlobsErrorTime;
-    if (timeSinceLastFail < GETBLOBS_RETRY_TIMEOUT) {
-      // do not try getblobs since it might not be available
-      this.logger.debug(
-        `disabled engine_getBlobsV1 api call since last failed < GETBLOBS_RETRY_TIMEOUT=${GETBLOBS_RETRY_TIMEOUT}`,
-        timeSinceLastFail
-      );
-      throw Error(
-        `engine_getBlobsV1 call recently failed timeSinceLastFail=${timeSinceLastFail} < GETBLOBS_RETRY_TIMEOUT=${GETBLOBS_RETRY_TIMEOUT}`
-      );
+    if (method === "engine_getBlobsV1") {
+      const timeSinceLastFail = timeNow - this.lastGetBlobsV1ErrorTime;
+      if (timeSinceLastFail < GETBLOBS_RETRY_TIMEOUT) {
+        // do not try getblobs since it might not be available
+        this.logger.debug(
+          `disabled ${method} api call since last failed < GETBLOBS_RETRY_TIMEOUT=${GETBLOBS_RETRY_TIMEOUT}`,
+          timeSinceLastFail
+        );
+        throw Error(
+          `${method} call recently failed timeSinceLastFail=${timeSinceLastFail} < GETBLOBS_RETRY_TIMEOUT=${GETBLOBS_RETRY_TIMEOUT}`
+        );
+      }
     }
 
-    const method = "engine_getBlobsV1";
-    assertReqSizeLimit(versionedHashes.length, 128);
+    assertReqSizeLimit(versionedHashes.length, MAX_VERSIONED_HASHES);
     const versionedHashesHex = versionedHashes.map(bytesToData);
-    let response = await this.rpc
+    const response = await this.rpc
       .fetchWithRetries<EngineApiRpcReturnTypes[typeof method], EngineApiRpcParamTypes[typeof method]>({
         method,
         params: [versionedHashesHex],
       })
       .catch((e) => {
-        if (e instanceof ErrorJsonRpcResponse && parseJsonRpcErrorCode(e.response.error.code) === "Method not found") {
-          this.lastGetBlobsErrorTime = timeNow;
-          this.logger.debug("disabling engine_getBlobsV1 api call since engine responded with method not availeble", {
+        if (
+          method === "engine_getBlobsV1" &&
+          e instanceof ErrorJsonRpcResponse &&
+          parseJsonRpcErrorCode(e.response.error.code) === "Method not found"
+        ) {
+          if (method === "engine_getBlobsV1") {
+            this.lastGetBlobsV1ErrorTime = timeNow;
+          }
+          this.logger.debug(`disabling ${method} api call since engine responded with method not available`, {
             retryTimeout: GETBLOBS_RETRY_TIMEOUT,
           });
         }
         throw e;
       });
 
-    // handle nethermind buggy response
-    // see: https://discord.com/channels/595666850260713488/1293605631785304088/1298956894274060301
-    if (
-      (response as unknown as {blobsAndProofs: EngineApiRpcReturnTypes[typeof method]}).blobsAndProofs !== undefined
-    ) {
-      response = (response as unknown as {blobsAndProofs: EngineApiRpcReturnTypes[typeof method]}).blobsAndProofs;
-    }
+    // engine_getBlobsV2 does not return partial responses. It returns an empty array if any blob is not found
+    // TODO: Spec says to return null if any blob is not found, but reth and nethermind return empty arrays as of peerdas-devnet-6
+    const invalidLength =
+      method === "engine_getBlobsV2"
+        ? response && response.length !== 0 && response.length !== versionedHashes.length
+        : !response || response.length !== versionedHashes.length;
 
-    if (response.length !== versionedHashes.length) {
-      const error = `Invalid engine_getBlobsV1 response length=${response.length} versionedHashes=${versionedHashes.length}`;
+    if (invalidLength) {
+      const error = `Invalid ${method} response length=${response?.length ?? "null"} versionedHashes=${versionedHashes.length}`;
       this.logger.error(error);
       throw Error(error);
     }
 
-    return response.map(deserializeBlobAndProofs);
+    // engine_getBlobsV2 returns a list of cell proofs per blob, whereas engine_getBlobsV1 returns one proof per blob
+    switch (method) {
+      case "engine_getBlobsV1":
+        return (response as EngineApiRpcReturnTypes[typeof method]).map(deserializeBlobAndProofs);
+      case "engine_getBlobsV2": {
+        const castResponse = response as EngineApiRpcReturnTypes[typeof method];
+        // TODO: Spec says to return null if any blob is not found, but reth and nethermind return empty arrays as of peerdas-devnet-6
+        if (castResponse === null || castResponse.length === 0) return null;
+        return castResponse.map(deserializeBlobAndProofsV2);
+      }
+    }
   }
 
   private async getClientVersion(clientVersion: ClientVersion): Promise<ClientVersion[]> {

packages/beacon-node/src/execution/engine/interface.ts

@@ -1,6 +1,14 @@
-import {CONSOLIDATION_REQUEST_TYPE, DEPOSIT_REQUEST_TYPE, ForkName, WITHDRAWAL_REQUEST_TYPE} from "@lodestar/params";
+import {
+  CONSOLIDATION_REQUEST_TYPE,
+  DEPOSIT_REQUEST_TYPE,
+  ForkName,
+  ForkPostFulu,
+  ForkPreFulu,
+  WITHDRAWAL_REQUEST_TYPE,
+} from "@lodestar/params";
 import {ExecutionPayload, ExecutionRequests, Root, RootHex, Wei, capella} from "@lodestar/types";
 import {Blob, BlobAndProof, KZGCommitment, KZGProof} from "@lodestar/types/deneb";
+import {BlobAndProofV2} from "@lodestar/types/fulu";
 
 import {DATA} from "../../eth1/provider/utils.js";
 import {PayloadId, PayloadIdCache, WithdrawalV1} from "./payloadIdCache.js";
@@ -189,5 +197,6 @@ export interface IExecutionEngine {
 
   getPayloadBodiesByRange(fork: ForkName, start: number, count: number): Promise<(ExecutionPayloadBody | null)[]>;
 
-  getBlobs(fork: ForkName, versionedHashes: VersionedHashes): Promise<(BlobAndProof | null)[]>;
+  getBlobs(fork: ForkPostFulu, versionedHashes: VersionedHashes): Promise<BlobAndProofV2[] | null>;
+  getBlobs(fork: ForkPreFulu, versionedHashes: VersionedHashes): Promise<(BlobAndProof | null)[]>;
 }

packages/beacon-node/src/execution/engine/mock.ts

@@ -100,6 +100,7 @@ export class ExecutionEngineMockBackend implements JsonRpcBackend {
       engine_getPayloadBodiesByRangeV1: this.getPayloadBodiesByRange.bind(this),
       engine_getClientVersionV1: this.getClientVersionV1.bind(this),
       engine_getBlobsV1: this.getBlobs.bind(this),
+      engine_getBlobsV2: this.getBlobsV2.bind(this),
     };
   }
 
@@ -404,6 +405,12 @@ export class ExecutionEngineMockBackend implements JsonRpcBackend {
     return versionedHashes.map((_vh) => null);
   }
 
+  private getBlobsV2(
+    _versionedHashes: EngineApiRpcParamTypes["engine_getBlobsV2"][0]
+  ): EngineApiRpcReturnTypes["engine_getBlobsV2"] {
+    return [];
+  }
+
   private timestampToFork(timestamp: number): ForkPostBellatrix {
     if (timestamp > (this.opts.electraForkTimestamp ?? Infinity)) return ForkName.electra;
     if (timestamp > (this.opts.denebForkTimestamp ?? Infinity)) return ForkName.deneb;

packages/beacon-node/src/execution/engine/types.ts

@@ -10,6 +10,7 @@ import {
 } from "@lodestar/params";
 import {ExecutionPayload, ExecutionRequests, Root, Wei, bellatrix, capella, deneb, electra, ssz} from "@lodestar/types";
 import {BlobAndProof} from "@lodestar/types/deneb";
+import {BlobAndProofV2} from "@lodestar/types/fulu";
 
 import {
   DATA,
@@ -80,6 +81,7 @@ export type EngineApiRpcParamTypes = {
   engine_getClientVersionV1: [ClientVersionRpc];
 
   engine_getBlobsV1: [DATA[]];
+  engine_getBlobsV2: [DATA[]];
 };
 
 export type PayloadStatus = {
@@ -124,6 +126,7 @@ export type EngineApiRpcReturnTypes = {
   engine_getClientVersionV1: ClientVersionRpc[];
 
   engine_getBlobsV1: (BlobAndProofRpc | null)[];
+  engine_getBlobsV2: BlobAndProofV2Rpc[] | null;
 };
 
 type ExecutionPayloadRpcWithValue = {
@@ -191,6 +194,11 @@ export type BlobAndProofRpc = {
   proof: DATA;
 };
 
+export type BlobAndProofV2Rpc = {
+  blob: DATA;
+  proofs: DATA[];
+};
+
 export type VersionedHashesRpc = DATA[];
 
 export type PayloadAttributesRpc = {
@@ -563,6 +571,13 @@ export function deserializeBlobAndProofs(data: BlobAndProofRpc | null): BlobAndP
     : null;
 }
 
+export function deserializeBlobAndProofsV2(data: BlobAndProofV2Rpc): BlobAndProofV2 {
+  return {
+    blob: dataToBytes(data.blob, BYTES_PER_FIELD_ELEMENT * FIELD_ELEMENTS_PER_BLOB),
+    proofs: data.proofs.map((proof) => dataToBytes(proof, 48)),
+  };
+}
+
 export function assertReqSizeLimit(blockHashesReqCount: number, count: number): void {
   if (blockHashesReqCount > count) {
     throw new Error(`Requested blocks must not be > ${count}`);

packages/types/src/fulu/sszTypes.ts

@@ -16,6 +16,9 @@ import {ssz as primitiveSsz} from "../primitive/index.js";
 
 const {BLSSignature, Root, ColumnIndex, RowIndex, Bytes32, Slot, UintNum64} = primitiveSsz;
 
+export const KZGProof = denebSsz.KZGProof;
+export const Blob = denebSsz.Blob;
+
 export const Metadata = new ContainerType(
   {
     ...altariSsz.Metadata.fields,

packages/types/src/fulu/types.ts

@@ -1,6 +1,9 @@
 import {ValueOf} from "@chainsafe/ssz";
 import * as ssz from "./sszTypes.js";
 
+export type KZGProof = ValueOf<typeof ssz.KZGProof>;
+export type Blob = ValueOf<typeof ssz.Blob>;
+
 export type Metadata = ValueOf<typeof ssz.Metadata>;
 
 export type Cell = ValueOf<typeof ssz.Cell>;
@@ -44,3 +47,7 @@ export type LightClientStore = ValueOf<typeof ssz.LightClientStore>;
 export type BlockContents = ValueOf<typeof ssz.BlockContents>;
 export type SignedBlockContents = ValueOf<typeof ssz.SignedBlockContents>;
 export type Contents = Omit<BlockContents, "block">;
+export type BlobAndProofV2 = {
+  blob: Blob;
+  proofs: KZGProof[];
+};