feat: refactor block input (#8200)

**Motivation**

Implement new IBlockInput on `unstable` version of peer das code.

---------

Co-authored-by: Cayman <[email protected]>
Co-authored-by: twoeths <[email protected]>
Co-authored-by: Tuyen Nguyen <[email protected]>
Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>
Co-authored-by: Nazar Hussain <[email protected]>
Co-authored-by: Nico Flaig <[email protected]>

Author: 6 people

dashboards/lodestar_block_processor.json

@@ -28,20 +28,11 @@ import {
   sszTypesFor,
 } from "@lodestar/types";
 import {fromAsync, fromHex, sleep, toHex, toRootHex} from "@lodestar/utils";
-import {
-  BlobsSource,
-  BlockInput,
-  BlockInputAvailableData,
-  BlockInputBlobs,
-  BlockInputDataColumns,
-  BlockInputType,
-  BlockSource,
-  DataColumnsSource,
-  ImportBlockOpts,
-  getBlockInput,
-} from "../../../../chain/blocks/types.js";
+import {BlockInputSource, isBlockInputBlobs, isBlockInputColumns} from "../../../../chain/blocks/blockInput/index.js";
+import {ImportBlockOpts} from "../../../../chain/blocks/types.js";
 import {verifyBlocksInEpoch} from "../../../../chain/blocks/verifyBlock.js";
 import {BeaconChain} from "../../../../chain/chain.js";
+import {ChainEvent} from "../../../../chain/emitter.js";
 import {BlockError, BlockErrorCode, BlockGossipError} from "../../../../chain/errors/index.js";
 import {
   BlockType,
@@ -51,7 +42,6 @@ import {
 } from "../../../../chain/produceBlock/index.js";
 import {validateGossipBlock} from "../../../../chain/validation/block.js";
 import {OpSource} from "../../../../chain/validatorMonitor.js";
-import {NetworkEvent} from "../../../../network/index.js";
 import {getBlobSidecars, kzgCommitmentToVersionedHash, reconstructBlobs} from "../../../../util/blobs.js";
 import {getDataColumnSidecarsFromBlock} from "../../../../util/dataColumns.js";
 import {isOptimisticBlock} from "../../../../util/forkChoice.js";
@@ -95,10 +85,15 @@ export function getBeaconBlockApi({
     const fork = config.getForkName(slot);
     const blockRoot = toRootHex(chain.config.getForkTypes(slot).BeaconBlock.hashTreeRoot(signedBlock.message));
 
-    let blockForImport: BlockInput, blobSidecars: deneb.BlobSidecars, dataColumnSidecars: fulu.DataColumnSidecars;
+    const blockForImport = chain.seenBlockInputCache.getByBlock({
+      block: signedBlock,
+      source: BlockInputSource.api,
+      seenTimestampSec,
+      blockRootHex: blockRoot,
+    });
+    let blobSidecars: deneb.BlobSidecars, dataColumnSidecars: fulu.DataColumnSidecars;
 
     if (isDenebBlockContents(signedBlockContents)) {
-      let blockData: BlockInputAvailableData;
       if (isForkPostFulu(fork)) {
         const timer = metrics?.peerDas.dataColumnSidecarComputationTime.startTimer();
         // If the block was produced by this node, we will already have computed cells
@@ -116,30 +111,36 @@ export function getBeaconBlockApi({
           cellsAndProofs
         );
         timer?.();
-        blockData = {
-          fork,
-          dataColumns: dataColumnSidecars,
-          dataColumnsBytes: dataColumnSidecars.map(() => null),
-          dataColumnsSource: DataColumnsSource.api,
-        } as BlockInputDataColumns;
         blobSidecars = [];
       } else if (isForkPostDeneb(fork)) {
         blobSidecars = getBlobSidecars(config, signedBlock, signedBlockContents.blobs, signedBlockContents.kzgProofs);
-        blockData = {
-          fork,
-          blobs: blobSidecars,
-          blobsSource: BlobsSource.api,
-        } as BlockInputBlobs;
         dataColumnSidecars = [];
       } else {
         throw Error(`Invalid data fork=${fork} for publish`);
       }
-
-      blockForImport = getBlockInput.availableData(config, signedBlock, BlockSource.api, blockData);
     } else {
       blobSidecars = [];
       dataColumnSidecars = [];
-      blockForImport = getBlockInput.preData(config, signedBlock, BlockSource.api);
+    }
+
+    if (isBlockInputColumns(blockForImport)) {
+      for (const dataColumnSidecar of dataColumnSidecars) {
+        blockForImport.addColumn({
+          blockRootHex: blockRoot,
+          columnSidecar: dataColumnSidecar,
+          source: BlockInputSource.api,
+          seenTimestampSec,
+        });
+      }
+    } else if (isBlockInputBlobs(blockForImport)) {
+      for (const blobSidecar of blobSidecars) {
+        blockForImport.addBlob({
+          blockRootHex: blockRoot,
+          blobSidecar,
+          source: BlockInputSource.api,
+          seenTimestampSec,
+        });
+      }
     }
 
     // check what validations have been requested before broadcasting and publishing the block
@@ -184,9 +185,10 @@ export function getBeaconBlockApi({
         if (!blockLocallyProduced) {
           const parentBlock = chain.forkChoice.getBlock(signedBlock.message.parentRoot);
           if (parentBlock === null) {
-            network.events.emit(NetworkEvent.unknownBlockParent, {
+            chain.emitter.emit(ChainEvent.unknownParent, {
               blockInput: blockForImport,
               peer: IDENTITY_PEER_ID,
+              source: BlockInputSource.api,
             });
             chain.persistInvalidSszValue(
               chain.config.getForkTypes(slot).SignedBeaconBlock,
@@ -247,18 +249,16 @@ export function getBeaconBlockApi({
 
     // Simple implementation of a pending block queue. Keeping the block here recycles the API logic, and keeps the
     // REST request promise without any extra infrastructure.
-    const msToBlockSlot =
-      computeTimeAtSlot(config, blockForImport.block.message.slot, chain.genesisTime) * 1000 - Date.now();
+    const msToBlockSlot = computeTimeAtSlot(config, slot, chain.genesisTime) * 1000 - Date.now();
     if (msToBlockSlot <= MAX_API_CLOCK_DISPARITY_MS && msToBlockSlot > 0) {
       // If block is a bit early, hold it in a promise. Equivalent to a pending queue.
       await sleep(msToBlockSlot);
     }
 
     // TODO: Validate block
-    const delaySec =
-      seenTimestampSec - (chain.genesisTime + blockForImport.block.message.slot * config.SECONDS_PER_SLOT);
+    const delaySec = seenTimestampSec - (chain.genesisTime + slot * config.SECONDS_PER_SLOT);
     metrics?.gossipBlock.elapsedTimeTillReceived.observe({source: OpSource.api}, delaySec);
-    chain.validatorMonitor?.registerBeaconBlock(OpSource.api, delaySec, blockForImport.block.message);
+    chain.validatorMonitor?.registerBeaconBlock(OpSource.api, delaySec, signedBlock.message);
 
     chain.logger.info("Publishing block", valLogMeta);
     const publishPromises = [
@@ -280,9 +280,10 @@ export function getBeaconBlockApi({
           .processBlock(blockForImport, {...opts, eagerPersistBlock: false})
           .catch((e) => {
             if (e instanceof BlockError && e.type.code === BlockErrorCode.PARENT_UNKNOWN) {
-              network.events.emit(NetworkEvent.unknownBlockParent, {
+              chain.emitter.emit(ChainEvent.unknownParent, {
                 blockInput: blockForImport,
                 peer: IDENTITY_PEER_ID,
+                source: BlockInputSource.api,
               });
             }
             throw e;
@@ -315,38 +316,34 @@ export function getBeaconBlockApi({
 
     chain.emitter.emit(routes.events.EventType.blockGossip, {slot, block: blockRoot});
 
-    if (blockForImport.type === BlockInputType.availableData) {
-      if (isForkPostFulu(blockForImport.blockData.fork)) {
-        const {dataColumns} = blockForImport.blockData as BlockInputDataColumns;
-        metrics?.dataColumns.bySource.inc({source: DataColumnsSource.api}, dataColumns.length);
-
-        if (chain.emitter.listenerCount(routes.events.EventType.dataColumnSidecar)) {
-          for (const dataColumnSidecar of dataColumns) {
-            chain.emitter.emit(routes.events.EventType.dataColumnSidecar, {
-              blockRoot,
-              slot,
-              index: dataColumnSidecar.index,
-              kzgCommitments: dataColumnSidecar.kzgCommitments.map(toHex),
-            });
-          }
-        }
-      } else if (
-        isForkPostDeneb(blockForImport.blockData.fork) &&
-        chain.emitter.listenerCount(routes.events.EventType.blobSidecar)
-      ) {
-        const {blobs} = blockForImport.blockData as BlockInputBlobs;
-
-        for (const blobSidecar of blobs) {
-          const {index, kzgCommitment} = blobSidecar;
-          chain.emitter.emit(routes.events.EventType.blobSidecar, {
+    if (isBlockInputColumns(blockForImport)) {
+      const dataColumns = blockForImport.getAllColumns();
+      metrics?.dataColumns.bySource.inc({source: BlockInputSource.api}, dataColumns.length);
+
+      if (chain.emitter.listenerCount(routes.events.EventType.dataColumnSidecar)) {
+        for (const dataColumnSidecar of dataColumns) {
+          chain.emitter.emit(routes.events.EventType.dataColumnSidecar, {
             blockRoot,
             slot,
-            index,
-            kzgCommitment: toHex(kzgCommitment),
-            versionedHash: toHex(kzgCommitmentToVersionedHash(kzgCommitment)),
+            index: dataColumnSidecar.index,
+            kzgCommitments: dataColumnSidecar.kzgCommitments.map(toHex),
           });
         }
       }
+    } else if (isBlockInputBlobs(blockForImport) && chain.emitter.listenerCount(routes.events.EventType.blobSidecar)) {
+      const blobSidecars = blockForImport.getBlobs();
+      const versionedHashes = blockForImport.getVersionedHashes();
+
+      for (const blobSidecar of blobSidecars) {
+        const {index, kzgCommitment} = blobSidecar;
+        chain.emitter.emit(routes.events.EventType.blobSidecar, {
+          blockRoot,
+          slot,
+          index,
+          kzgCommitment: toHex(kzgCommitment),
+          versionedHash: toHex(versionedHashes[index]),
+        });
+      }
     }
   };
 

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

@@ -115,7 +115,7 @@ export function getLodestarApi({
         data: (chain as BeaconChain)["blockProcessor"].jobQueue.getItems().map((item) => {
           const [blockInputs, opts] = item.args;
           return {
-            blockSlots: blockInputs.map((blockInput) => blockInput.block.message.slot),
+            blockSlots: blockInputs.map((blockInput) => blockInput.slot),
             jobOpts: opts,
             addedTimeMs: item.addedTimeMs,
           };

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

@@ -57,6 +57,7 @@ import {
   toRootHex,
 } from "@lodestar/utils";
 import {MAX_BUILDER_BOOST_FACTOR} from "@lodestar/validator";
+import {BlockInputSource} from "../../../chain/blocks/blockInput/types.js";
 import {
   AttestationError,
   AttestationErrorCode,
@@ -978,7 +979,7 @@ export function getValidatorApi(
       // see https://github.com/ChainSafe/lodestar/issues/5063
       if (!chain.forkChoice.hasBlock(beaconBlockRoot)) {
         const rootHex = toRootHex(beaconBlockRoot);
-        network.searchUnknownSlotRoot({slot, root: rootHex});
+        network.searchUnknownSlotRoot({slot, root: rootHex}, BlockInputSource.api);
         // if result of this call is false, i.e. block hasn't seen after 1 slot then the below notOnOptimisticBlockRoot call will throw error
         await chain.waitForBlock(slot, rootHex);
       }

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

@@ -0,0 +1,72 @@
+import {Logger, sleep} from "@lodestar/utils";
+import {ChainEventEmitter} from "./emitter.js";
+import {Metrics} from "../metrics/metrics.js";
+import {ChainForkConfig} from "@lodestar/config";
+import {BlockInputColumns} from "./blocks/blockInput/index.js";
+import {recoverDataColumnSidecars} from "../util/dataColumns.js";
+
+/**
+ * Minimum time to wait before attempting reconstruction
+ */
+const RECONSTRUCTION_DELAY_MIN_MS = 800;
+
+/**
+ * Maximum time to wait before attempting reconstruction
+ */
+const RECONSTRUCTION_DELAY_MAX_MS = 1200;
+
+export type ColumnReconstructionTrackerInit = {
+  logger: Logger;
+  emitter: ChainEventEmitter;
+  metrics: Metrics | null;
+  config: ChainForkConfig;
+};
+
+/**
+ * Tracks column reconstruction attempts to avoid duplicate and multiple in-flight calls
+ */
+export class ColumnReconstructionTracker {
+  logger: Logger;
+  emitter: ChainEventEmitter;
+  metrics: Metrics | null;
+  config: ChainForkConfig;
+
+  /**
+   * Track last attempted block root
+   *
+   * This is sufficient to avoid duplicate calls since we only call this
+   * function when we see a new data column sidecar from gossip.
+   */
+  lastBlockRootHex: string | null = null;
+  /** Track if a reconstruction attempt is in-flight */
+  running = false;
+
+  constructor(init: ColumnReconstructionTrackerInit) {
+    this.logger = init.logger;
+    this.emitter = init.emitter;
+    this.metrics = init.metrics;
+    this.config = init.config;
+  }
+
+  triggerColumnReconstruction(blockInput: BlockInputColumns): void {
+    if (this.running) {
+      return;
+    }
+
+    if (this.lastBlockRootHex === blockInput.blockRootHex) {
+      return;
+    }
+
+    // We don't care about the outcome of this call,
+    // just that it has been triggered for this block root.
+    this.running = true;
+    this.lastBlockRootHex = blockInput.blockRootHex;
+    const delay =
+      RECONSTRUCTION_DELAY_MIN_MS + Math.random() * (RECONSTRUCTION_DELAY_MAX_MS - RECONSTRUCTION_DELAY_MIN_MS);
+    sleep(delay).then(() => {
+      recoverDataColumnSidecars(blockInput, this.emitter, this.metrics).finally(() => {
+        this.running = false;
+      });
+    });
+  }
+}