feat: add and use getBlobsV1 to expedite gossip import (#7134)

* hookup the getblobs api to get bob and proof data from el

remove unused

fix import

metrics overhault, test, debugging testing, some feeback

fix

add nethermind bug dicussion link

fix

resolve conflicts

* deblobs timeout

* fix metric

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

@@ -215,12 +215,12 @@ export function getBeaconBlockApi({
       // specification is very clear that this is the desired behaviour.
       //
       // i) Publish blobs and block before importing so that network can see them asap
-      // ii) publish blobs first because
-      //     a) by the times nodes see block, they might decide to pull blobs
-      //     b) they might require more hops to reach recipients in peerDAS kind of setup where
-      //        blobs might need to hop between nodes because of partial subnet subscription
-      ...blobSidecars.map((blobSidecar) => () => network.publishBlobSidecar(blobSidecar)),
+      // ii) publish block first because
+      //     a) as soon as node sees block they can start processing it while blobs arrive
+      //     b) getting block first allows nodes to use getBlobs from local ELs and save
+      //        import latency and hopefully bandwidth
       () => network.publishBeaconBlock(signedBlock) as Promise<unknown>,
+      ...blobSidecars.map((blobSidecar) => () => network.publishBlobSidecar(blobSidecar)),
       () =>
         // there is no rush to persist block since we published it to gossip anyway
         chain

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

@@ -106,7 +106,7 @@ export async function verifyBlocksInEpoch(
           } as SegmentExecStatus),
 
       // data availability for the blobs
-      verifyBlocksDataAvailability(this, blocksInput, opts),
+      verifyBlocksDataAvailability(this, blocksInput, abortController.signal, opts),
 
       // Run state transition only
       // TODO: Ensure it yields to allow flushing to workers and engine API

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

@@ -2,7 +2,7 @@ import {ChainForkConfig} from "@lodestar/config";
 import {DataAvailabilityStatus} from "@lodestar/fork-choice";
 import {computeTimeAtSlot} from "@lodestar/state-transition";
 import {UintNum64, deneb} from "@lodestar/types";
-import {Logger} from "@lodestar/utils";
+import {ErrorAborted, Logger} from "@lodestar/utils";
 import {Metrics} from "../../metrics/metrics.js";
 import {BlockError, BlockErrorCode} from "../errors/index.js";
 import {validateBlobSidecars} from "../validation/blobSidecar.js";
@@ -27,6 +27,7 @@ const BLOB_AVAILABILITY_TIMEOUT = 12_000;
 export async function verifyBlocksDataAvailability(
   chain: {config: ChainForkConfig; genesisTime: UintNum64; logger: Logger; metrics: Metrics | null},
   blocks: BlockInput[],
+  signal: AbortSignal,
   opts: ImportBlockOpts
 ): Promise<{
   dataAvailabilityStatuses: DataAvailabilityStatus[];
@@ -43,9 +44,12 @@ export async function verifyBlocksDataAvailability(
   const availableBlockInputs: BlockInput[] = [];
 
   for (const blockInput of blocks) {
+    if (signal.aborted) {
+      throw new ErrorAborted("verifyBlocksDataAvailability");
+    }
     // Validate status of only not yet finalized blocks, we don't need yet to propogate the status
     // as it is not used upstream anywhere
-    const {dataAvailabilityStatus, availableBlockInput} = await maybeValidateBlobs(chain, blockInput, opts);
+    const {dataAvailabilityStatus, availableBlockInput} = await maybeValidateBlobs(chain, blockInput, signal, opts);
     dataAvailabilityStatuses.push(dataAvailabilityStatus);
     availableBlockInputs.push(availableBlockInput);
   }
@@ -69,6 +73,7 @@ export async function verifyBlocksDataAvailability(
 async function maybeValidateBlobs(
   chain: {config: ChainForkConfig; genesisTime: UintNum64; logger: Logger},
   blockInput: BlockInput,
+  signal: AbortSignal,
   opts: ImportBlockOpts
 ): Promise<{dataAvailabilityStatus: DataAvailabilityStatus; availableBlockInput: BlockInput}> {
   switch (blockInput.type) {
@@ -92,7 +97,7 @@ async function maybeValidateBlobs(
       const blobsData =
         blockInput.type === BlockInputType.availableData
           ? blockInput.blockData
-          : await raceWithCutoff(chain, blockInput, blockInput.cachedData.availabilityPromise);
+          : await raceWithCutoff(chain, blockInput, blockInput.cachedData.availabilityPromise, signal);
       const {blobs} = blobsData;
 
       const {blobKzgCommitments} = (block as deneb.SignedBeaconBlock).message.body;
@@ -122,16 +127,21 @@ async function maybeValidateBlobs(
 async function raceWithCutoff<T>(
   chain: {config: ChainForkConfig; genesisTime: UintNum64; logger: Logger},
   blockInput: BlockInput,
-  availabilityPromise: Promise<T>
+  availabilityPromise: Promise<T>,
+  signal: AbortSignal
 ): Promise<T> {
   const {block} = blockInput;
   const blockSlot = block.message.slot;
 
-  const cutoffTime = Math.max(
-    computeTimeAtSlot(chain.config, blockSlot, chain.genesisTime) * 1000 + BLOB_AVAILABILITY_TIMEOUT - Date.now(),
-    0
-  );
-  const cutoffTimeout = new Promise((_resolve, reject) => setTimeout(reject, cutoffTime));
+  const cutoffTime =
+    computeTimeAtSlot(chain.config, blockSlot, chain.genesisTime) * 1000 + BLOB_AVAILABILITY_TIMEOUT - Date.now();
+  const cutoffTimeout =
+    cutoffTime > 0
+      ? new Promise((_resolve, reject) => {
+          setTimeout(() => reject(new Error("Timeout exceeded")), cutoffTime);
+          signal.addEventListener("abort", () => reject(signal.reason));
+        })
+      : Promise.reject(new Error("Cutoff time must be greater than 0"));
   chain.logger.debug("Racing for blob availabilityPromise", {blockSlot, cutoffTime});
 
   try {

packages/beacon-node/src/eth1/provider/jsonRpcHttpClient.ts

@@ -372,7 +372,7 @@ export class HttpRpcError extends Error {
 /**
  * JSON RPC spec errors https://www.jsonrpc.org/specification#response_object
  */
-function parseJsonRpcErrorCode(code: number): string {
+export function parseJsonRpcErrorCode(code: number): string {
   if (code === -32700) return "Parse request error";
   if (code === -32600) return "Invalid request object";
   if (code === -32601) return "Method not found";

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

@@ -28,4 +28,8 @@ export class ExecutionEngineDisabled implements IExecutionEngine {
   getPayloadBodiesByRange(): Promise<never> {
     throw Error("Execution engine disabled");
   }
+
+  getBlobs(): Promise<never> {
+    throw Error("Execution engine disabled");
+  }
 }

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

@@ -1,14 +1,16 @@
 import {Logger} from "@lodestar/logger";
 import {ForkName, ForkSeq, SLOTS_PER_EPOCH} from "@lodestar/params";
 import {ExecutionPayload, ExecutionRequests, Root, RootHex, Wei} from "@lodestar/types";
+import {BlobAndProof} from "@lodestar/types/deneb";
 import {
   ErrorJsonRpcResponse,
   HttpRpcError,
   IJsonRpcHttpClient,
   JsonRpcHttpClientEvent,
   ReqOpts,
+  parseJsonRpcErrorCode,
 } from "../../eth1/provider/jsonRpcHttpClient.js";
-import {numToQuantity} from "../../eth1/provider/utils.js";
+import {bytesToData, numToQuantity} from "../../eth1/provider/utils.js";
 import {Metrics} from "../../metrics/index.js";
 import {EPOCHS_PER_BATCH} from "../../sync/constants.js";
 import {getLodestarClientVersion} from "../../util/metadata.js";
@@ -31,6 +33,7 @@ import {
   EngineApiRpcReturnTypes,
   ExecutionPayloadBody,
   assertReqSizeLimit,
+  deserializeBlobAndProofs,
   deserializeExecutionPayloadBody,
   parseExecutionPayload,
   serializeBeaconBlockRoot,
@@ -111,6 +114,7 @@ const getPayloadOpts: ReqOpts = {routeId: "getPayload"};
  */
 export class ExecutionEngineHttp implements IExecutionEngine {
   private logger: Logger;
+  private lastGetBlobsErrorTime = 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
@@ -461,6 +465,57 @@ export class ExecutionEngineHttp implements IExecutionEngine {
     return response.map(deserializeExecutionPayloadBody);
   }
 
+  async getBlobs(_fork: ForkName, versionedHashes: VersionedHashes): Promise<(BlobAndProof | null)[]> {
+    // 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}`
+      );
+    }
+
+    const method = "engine_getBlobsV1";
+    assertReqSizeLimit(versionedHashes.length, 128);
+    const versionedHashesHex = versionedHashes.map(bytesToData);
+    let 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", {
+            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;
+    }
+
+    if (response.length !== versionedHashes.length) {
+      const error = `Invalid engine_getBlobsV1 response length=${response.length} versionedHashes=${versionedHashes.length}`;
+      this.logger.error(error);
+      throw Error(error);
+    }
+
+    return response.map(deserializeBlobAndProofs);
+  }
+
   private async getClientVersion(clientVersion: ClientVersion): Promise<ClientVersion[]> {
     const method = "engine_getClientVersionV1";
 

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

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

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

@@ -99,6 +99,7 @@ export class ExecutionEngineMockBackend implements JsonRpcBackend {
       engine_getPayloadBodiesByHashV1: this.getPayloadBodiesByHash.bind(this),
       engine_getPayloadBodiesByRangeV1: this.getPayloadBodiesByRange.bind(this),
       engine_getClientVersionV1: this.getClientVersionV1.bind(this),
+      engine_getBlobsV1: this.getBlobs.bind(this),
     };
   }
 
@@ -397,6 +398,12 @@ export class ExecutionEngineMockBackend implements JsonRpcBackend {
     return [{code: ClientCode.XX, name: "mock", version: "", commit: ""}];
   }
 
+  private getBlobs(
+    versionedHashes: EngineApiRpcParamTypes["engine_getBlobsV1"][0]
+  ): EngineApiRpcReturnTypes["engine_getBlobsV1"] {
+    return versionedHashes.map((_vh) => null);
+  }
+
   private timestampToFork(timestamp: number): ForkExecution {
     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

@@ -6,6 +6,7 @@ import {
   ForkSeq,
 } from "@lodestar/params";
 import {ExecutionPayload, ExecutionRequests, Root, Wei, bellatrix, capella, deneb, electra, ssz} from "@lodestar/types";
+import {BlobAndProof} from "@lodestar/types/deneb";
 
 import {
   DATA,
@@ -67,6 +68,8 @@ export type EngineApiRpcParamTypes = {
    * Object - Instance of ClientVersion
    */
   engine_getClientVersionV1: [ClientVersionRpc];
+
+  engine_getBlobsV1: [DATA[]];
 };
 
 export type PayloadStatus = {
@@ -109,6 +112,8 @@ export type EngineApiRpcReturnTypes = {
   engine_getPayloadBodiesByRangeV1: (ExecutionPayloadBodyRpc | null)[];
 
   engine_getClientVersionV1: ClientVersionRpc[];
+
+  engine_getBlobsV1: (BlobAndProofRpc | null)[];
 };
 
 type ExecutionPayloadRpcWithValue = {
@@ -171,6 +176,11 @@ export type DepositRequestsRpc = DATA;
 export type WithdrawalRequestsRpc = DATA;
 export type ConsolidationRequestsRpc = DATA;
 
+export type BlobAndProofRpc = {
+  blob: DATA;
+  proof: DATA;
+};
+
 export type VersionedHashesRpc = DATA[];
 
 export type PayloadAttributesRpc = {
@@ -462,6 +472,15 @@ export function serializeExecutionPayloadBody(data: ExecutionPayloadBody | null)
     : null;
 }
 
+export function deserializeBlobAndProofs(data: BlobAndProofRpc | null): BlobAndProof | null {
+  return data
+    ? {
+        blob: dataToBytes(data.blob, BYTES_PER_FIELD_ELEMENT * FIELD_ELEMENTS_PER_BLOB),
+        proof: dataToBytes(data.proof, 48),
+      }
+    : null;
+}
+
 export function assertReqSizeLimit(blockHashesReqCount: number, count: number): void {
   if (blockHashesReqCount > count) {
     throw new Error(`Requested blocks must not be > ${count}`);