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

This reverts commit 2e5265b.

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 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>,
+      // 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)),
+      () => network.publishBeaconBlock(signedBlock) as Promise<unknown>,
       () =>
         // there is no rush to persist block since we published it to gossip anyway
         chain

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

@@ -107,7 +107,7 @@ export async function verifyBlocksInEpoch(
           } as SegmentExecStatus),
 
       // data availability for the blobs
-      verifyBlocksDataAvailability(this, blocksInput, abortController.signal, opts),
+      verifyBlocksDataAvailability(this, blocksInput, 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 {ErrorAborted, Logger} from "@lodestar/utils";
+import {Logger} from "@lodestar/utils";
 import {Metrics} from "../../metrics/metrics.js";
 import {BlockError, BlockErrorCode} from "../errors/index.js";
 import {validateBlobSidecars} from "../validation/blobSidecar.js";
@@ -27,7 +27,6 @@ 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[];
@@ -44,12 +43,9 @@ 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, signal, opts);
+    const {dataAvailabilityStatus, availableBlockInput} = await maybeValidateBlobs(chain, blockInput, opts);
     dataAvailabilityStatuses.push(dataAvailabilityStatus);
     availableBlockInputs.push(availableBlockInput);
   }
@@ -73,7 +69,6 @@ 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) {
@@ -97,7 +92,7 @@ async function maybeValidateBlobs(
       const blobsData =
         blockInput.type === BlockInputType.availableData
           ? blockInput.blockData
-          : await raceWithCutoff(chain, blockInput, blockInput.cachedData.availabilityPromise, signal);
+          : await raceWithCutoff(chain, blockInput, blockInput.cachedData.availabilityPromise);
       const {blobs} = blobsData;
 
       const {blobKzgCommitments} = (block as deneb.SignedBeaconBlock).message.body;
@@ -127,21 +122,16 @@ async function maybeValidateBlobs(
 async function raceWithCutoff<T>(
   chain: {config: ChainForkConfig; genesisTime: UintNum64; logger: Logger},
   blockInput: BlockInput,
-  availabilityPromise: Promise<T>,
-  signal: AbortSignal
+  availabilityPromise: Promise<T>
 ): Promise<T> {
   const {block} = blockInput;
   const blockSlot = block.message.slot;
 
-  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"));
+  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));
   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
  */
-export function parseJsonRpcErrorCode(code: number): string {
+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,8 +28,4 @@ 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,17 +1,15 @@
 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 {strip0xPrefix} from "@lodestar/utils";
 import {
   ErrorJsonRpcResponse,
   HttpRpcError,
   IJsonRpcHttpClient,
   JsonRpcHttpClientEvent,
   ReqOpts,
-  parseJsonRpcErrorCode,
 } from "../../eth1/provider/jsonRpcHttpClient.js";
-import {bytesToData, numToQuantity} from "../../eth1/provider/utils.js";
+import {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";
@@ -34,7 +32,6 @@ import {
   EngineApiRpcReturnTypes,
   ExecutionPayloadBody,
   assertReqSizeLimit,
-  deserializeBlobAndProofs,
   deserializeExecutionPayloadBody,
   parseExecutionPayload,
   serializeBeaconBlockRoot,
@@ -115,7 +112,6 @@ 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
@@ -466,57 +462,6 @@ 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, BlobAndProof, KZGCommitment, KZGProof} from "@lodestar/types/deneb";
+import {Blob, KZGCommitment, KZGProof} from "@lodestar/types/deneb";
 
 import {DATA} from "../../eth1/provider/utils.js";
 import {PayloadId, PayloadIdCache, WithdrawalV1} from "./payloadIdCache.js";
@@ -179,6 +179,4 @@ 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,7 +99,6 @@ 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),
     };
   }
 
@@ -398,12 +397,6 @@ 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,7 +6,6 @@ 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,
@@ -68,8 +67,6 @@ export type EngineApiRpcParamTypes = {
    * Object - Instance of ClientVersion
    */
   engine_getClientVersionV1: [ClientVersionRpc];
-
-  engine_getBlobsV1: [DATA[]];
 };
 
 export type PayloadStatus = {
@@ -112,8 +109,6 @@ export type EngineApiRpcReturnTypes = {
   engine_getPayloadBodiesByRangeV1: (ExecutionPayloadBodyRpc | null)[];
 
   engine_getClientVersionV1: ClientVersionRpc[];
-
-  engine_getBlobsV1: (BlobAndProofRpc | null)[];
 };
 
 type ExecutionPayloadRpcWithValue = {
@@ -176,11 +171,6 @@ 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 = {
@@ -472,15 +462,6 @@ 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}`);