feat: add and track blob source for metrics

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

@@ -4,7 +4,7 @@ import {computeEpochAtSlot, computeTimeAtSlot, reconstructFullBlockOrContents} f
 import {SLOTS_PER_HISTORICAL_ROOT} from "@lodestar/params";
 import {sleep, toHex} from "@lodestar/utils";
 import {allForks, deneb, isSignedBlockContents, ProducedBlockSource} from "@lodestar/types";
-import {BlockSource, getBlockInput, ImportBlockOpts, BlockInput} from "../../../../chain/blocks/types.js";
+import {BlockSource, getBlockInput, ImportBlockOpts, BlockInput, BlobsSource} from "../../../../chain/blocks/types.js";
 import {promiseAllMaybeAsync} from "../../../../util/promises.js";
 import {isOptimisticBlock} from "../../../../util/forkChoice.js";
 import {computeBlobSidecars} from "../../../../util/blobs.js";
@@ -52,6 +52,7 @@ export function getBeaconBlockApi({
         signedBlock,
         BlockSource.api,
         blobSidecars,
+        BlobsSource.api,
         // don't bundle any bytes for block and blobs
         null,
         blobSidecars.map(() => null)

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

@@ -95,15 +95,20 @@ export async function importBlock(
   this.logger.verbose("Added block to forkchoice and state cache", {slot: blockSlot, root: blockRootHex});
 
   // We want to import block asap so call all event handler in the next event loop
-  setTimeout(() => {
+  setTimeout(async () => {
     this.emitter.emit(routes.events.EventType.block, {
       block: blockRootHex,
       slot: blockSlot,
       executionOptimistic: blockSummary != null && isOptimisticBlock(blockSummary),
     });
 
-    if (blockInput.type === BlockInputType.postDeneb) {
-      for (const blobSidecar of blockInput.blobs) {
+    if (blockInput.type === BlockInputType.postDeneb || blockInput.type === BlockInputType.blobsPromise) {
+      const blobsData =
+        blockInput.type === BlockInputType.postDeneb ? blockInput : await blockInput.availabilityPromise;
+      const {blobsSource, blobs} = blobsData;
+
+      this.metrics?.importBlock.blobBySource.inc({blobsSource});
+      for (const blobSidecar of blobs) {
         const {index, kzgCommitment} = blobSidecar;
         this.emitter.emit(routes.events.EventType.blobSidecar, {
           blockRoot: blockRootHex,

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

@@ -18,13 +18,21 @@ export enum BlockSource {
   byRoot = "req_resp_by_root",
 }
 
+/** Enum to represent where blobs come from */
+export enum BlobsSource {
+  gossip = "gossip",
+  api = "api",
+  byRange = "req_resp_by_range",
+  byRoot = "req_resp_by_root",
+}
+
 export enum GossipedInputType {
   block = "block",
   blob = "blob",
 }
 
 export type BlobsCache = Map<number, {blobSidecar: deneb.BlobSidecar; blobBytes: Uint8Array | null}>;
-export type BlockInputBlobs = {blobs: deneb.BlobSidecars; blobsBytes: (Uint8Array | null)[]};
+export type BlockInputBlobs = {blobs: deneb.BlobSidecars; blobsBytes: (Uint8Array | null)[]; blobsSource: BlobsSource};
 type CachedBlobs = {
   blobsCache: BlobsCache;
   availabilityPromise: Promise<BlockInputBlobs>;
@@ -34,6 +42,7 @@ type CachedBlobs = {
 export type BlockInput = {block: allForks.SignedBeaconBlock; source: BlockSource; blockBytes: Uint8Array | null} & (
   | {type: BlockInputType.preDeneb}
   | ({type: BlockInputType.postDeneb} & BlockInputBlobs)
+  // the blobsSource here is added to BlockInputBlobs when availability is resolved
   | ({type: BlockInputType.blobsPromise} & CachedBlobs)
 );
 export type NullBlockInput = {block: null; blockRootHex: RootHex; blockInputPromise: Promise<BlockInput>} & CachedBlobs;
@@ -69,6 +78,7 @@ export const getBlockInput = {
     block: allForks.SignedBeaconBlock,
     source: BlockSource,
     blobs: deneb.BlobSidecars,
+    blobsSource: BlobsSource,
     blockBytes: Uint8Array | null,
     blobsBytes: (Uint8Array | null)[]
   ): BlockInput {
@@ -80,6 +90,7 @@ export const getBlockInput = {
       block,
       source,
       blobs,
+      blobsSource,
       blockBytes,
       blobsBytes,
     };
@@ -109,7 +120,7 @@ export const getBlockInput = {
   },
 };
 
-export function getBlockInputBlobs(blobsCache: BlobsCache): BlockInputBlobs {
+export function getBlockInputBlobs(blobsCache: BlobsCache, blobsSource: BlobsSource): BlockInputBlobs {
   const blobs = [];
   const blobsBytes = [];
 
@@ -122,7 +133,7 @@ export function getBlockInputBlobs(blobsCache: BlobsCache): BlockInputBlobs {
     blobs.push(blobSidecar);
     blobsBytes.push(blobBytes);
   }
-  return {blobs, blobsBytes};
+  return {blobs, blobsBytes, blobsSource};
 }
 
 export enum AttestationImportOpt {

packages/beacon-node/src/chain/seenCache/seenGossipBlockInput.ts

@@ -13,6 +13,7 @@ import {
   BlobsCache,
   GossipedInputType,
   getBlockInputBlobs,
+  BlobsSource,
 } from "../blocks/types.js";
 import {Metrics} from "../../metrics/index.js";
 
@@ -134,15 +135,16 @@ export class SeenGossipBlockInput {
       }
 
       if (blobKzgCommitments.length === blobsCache.size) {
-        const allBlobs = getBlockInputBlobs(blobsCache);
+        const allBlobs = getBlockInputBlobs(blobsCache, BlobsSource.gossip);
         resolveAvailability(allBlobs);
         metrics?.syncUnknownBlock.resolveAvailabilitySource.inc({source: BlockInputAvailabilitySource.GOSSIP});
-        const {blobs, blobsBytes} = allBlobs;
+        const {blobs, blobsBytes, blobsSource} = allBlobs;
         const blockInput = getBlockInput.postDeneb(
           config,
           signedBlock,
           BlockSource.gossip,
           blobs,
+          blobsSource,
           blockBytes ?? null,
           blobsBytes
         );

packages/beacon-node/src/metrics/metrics/lodestar.ts

@@ -1,6 +1,6 @@
 import {EpochTransitionStep, StateCloneSource, StateHashTreeRootSource} from "@lodestar/state-transition";
 import {allForks} from "@lodestar/types";
-import {BlockSource} from "../../chain/blocks/types.js";
+import {BlockSource, BlobsSource} from "../../chain/blocks/types.js";
 import {JobQueueItemType} from "../../chain/bls/index.js";
 import {BlockErrorCode} from "../../chain/errors/index.js";
 import {InsertOutcome} from "../../chain/opPools/types.js";
@@ -800,6 +800,11 @@ export function createLodestarMetrics(
         help: "Total number of imported blocks by source",
         labelNames: ["source"],
       }),
+      blobBySource: register.gauge<{blobsSource: BlobsSource}>({
+        name: "lodestar_import_blob_by_source_total",
+        help: "Total number of imported blobs by source",
+        labelNames: ["blobsSource"],
+      }),
     },
     engineNotifyNewPayloadResult: register.gauge<{result: ExecutionPayloadStatus}>({
       name: "lodestar_execution_engine_notify_new_payload_result_total",

packages/beacon-node/src/network/reqresp/beaconBlocksMaybeBlobsByRange.ts

@@ -3,7 +3,7 @@ import {deneb, Epoch, phase0, allForks, Slot} from "@lodestar/types";
 import {ForkSeq} from "@lodestar/params";
 import {computeEpochAtSlot} from "@lodestar/state-transition";
 
-import {BlockInput, BlockSource, getBlockInput} from "../../chain/blocks/types.js";
+import {BlobsSource, BlockInput, BlockSource, getBlockInput} from "../../chain/blocks/types.js";
 import {PeerIdStr} from "../../util/peerId.js";
 import {INetwork, WithBytes} from "../interface.js";
 
@@ -43,7 +43,7 @@ export async function beaconBlocksMaybeBlobsByRange(
       network.sendBlobSidecarsByRange(peerId, request),
     ]);
 
-    return matchBlockWithBlobs(config, allBlocks, allBlobSidecars, endSlot, BlockSource.byRange);
+    return matchBlockWithBlobs(config, allBlocks, allBlobSidecars, endSlot, BlockSource.byRange, BlobsSource.byRange);
   }
 
   // Post Deneb but old blobs
@@ -58,7 +58,8 @@ export function matchBlockWithBlobs(
   allBlocks: WithBytes<allForks.SignedBeaconBlock>[],
   allBlobSidecars: deneb.BlobSidecar[],
   endSlot: Slot,
-  blockSource: BlockSource
+  blockSource: BlockSource,
+  blobsSource: BlobsSource
 ): BlockInput[] {
   const blockInputs: BlockInput[] = [];
   let blobSideCarIndex = 0;
@@ -101,6 +102,7 @@ export function matchBlockWithBlobs(
           block.data,
           blockSource,
           blobSidecars,
+          blobsSource,
           null,
           Array.from({length: blobKzgCommitmentsLen}, () => null)
         )

packages/beacon-node/src/network/reqresp/beaconBlocksMaybeBlobsByRoot.ts

@@ -9,6 +9,7 @@ import {
   getBlockInputBlobs,
   getBlockInput,
   NullBlockInput,
+  BlobsSource,
 } from "../../chain/blocks/types.js";
 import {PeerIdStr} from "../../util/peerId.js";
 import {INetwork} from "../interface.js";
@@ -47,7 +48,7 @@ export async function beaconBlocksMaybeBlobsByRoot(
 
   // The last arg is to provide slot to which all blobs should be exausted in matching
   // and here it should be infinity since all bobs should match
-  return matchBlockWithBlobs(config, allBlocks, allBlobSidecars, Infinity, BlockSource.byRoot);
+  return matchBlockWithBlobs(config, allBlocks, allBlobSidecars, Infinity, BlockSource.byRoot, BlobsSource.byRoot);
 }
 
 export async function unavailableBeaconBlobsByRoot(
@@ -98,13 +99,13 @@ export async function unavailableBeaconBlobsByRoot(
 
   // check and see if all blobs are now available and in that case resolve availability
   // if not this will error and the leftover blobs will be tried from another peer
-  const allBlobs = getBlockInputBlobs(blobsCache);
-  const {blobs, blobsBytes} = allBlobs;
+  const allBlobs = getBlockInputBlobs(blobsCache, BlobsSource.byRoot);
+  const {blobs, blobsBytes, blobsSource} = allBlobs;
   if (blobs.length !== blobKzgCommitmentsLen) {
     throw Error(`Not all blobs fetched missingBlobs=${blobKzgCommitmentsLen - blobs.length}`);
   }
 
   resolveAvailability(allBlobs);
   metrics?.syncUnknownBlock.resolveAvailabilitySource.inc({source: BlockInputAvailabilitySource.UNKNOWN_SYNC});
-  return getBlockInput.postDeneb(config, block, BlockSource.byRoot, blobs, blockBytes, blobsBytes);
+  return getBlockInput.postDeneb(config, block, BlockSource.byRoot, blobs, blobsSource, blockBytes, blobsBytes);
 }

packages/beacon-node/test/spec/presets/fork_choice.test.ts

@@ -27,6 +27,7 @@ import {
   AttestationImportOpt,
   BlockSource,
   BlobSidecarValidation,
+  BlobsSource,
 } from "../../../src/chain/blocks/types.js";
 import {ZERO_HASH_HEX} from "../../../src/constants/constants.js";
 import {PowMergeBlock} from "../../../src/eth1/interface.js";
@@ -209,9 +210,15 @@ const forkChoiceTest =
                     };
                   });
 
-                  blockImport = getBlockInput.postDeneb(config, signedBlock, BlockSource.gossip, blobSidecars, null, [
+                  blockImport = getBlockInput.postDeneb(
+                    config,
+                    signedBlock,
+                    BlockSource.gossip,
+                    blobSidecars,
+                    BlobsSource.gossip,
                     null,
-                  ]);
+                    [null]
+                  );
                 } else {
                   blockImport = getBlockInput.preDeneb(config, signedBlock, BlockSource.gossip, null);
                 }

packages/beacon-node/test/unit/network/beaconBlocksMaybeBlobsByRange.test.ts

@@ -3,7 +3,7 @@ import {ssz, deneb} from "@lodestar/types";
 import {createBeaconConfig, createChainForkConfig, defaultChainConfig} from "@lodestar/config";
 
 import {beaconBlocksMaybeBlobsByRange} from "../../../src/network/reqresp/index.js";
-import {BlockInputType, BlockSource} from "../../../src/chain/blocks/types.js";
+import {BlockInputType, BlockSource, BlobsSource} from "../../../src/chain/blocks/types.js";
 import {initCKZG, loadEthereumTrustedSetup} from "../../../src/util/kzg.js";
 import {INetwork} from "../../../src/network/interface.js";
 import {ZERO_HASH} from "../../../src/constants/constants.js";
@@ -104,6 +104,7 @@ describe("beaconBlocksMaybeBlobsByRange", () => {
           block,
           source: BlockSource.byRange,
           blobs,
+          blobsSource: BlobsSource.byRange,
           blockBytes: null,
           blobsBytes: blobs.map(() => null),
         };