feat: data_column_sidecar sse event support

packages/api/src/beacon/routes/events.ts

@@ -1,4 +1,4 @@
-import {ContainerType, ValueOf} from "@chainsafe/ssz";
+import {ContainerType, ListBasicType, ValueOf} from "@chainsafe/ssz";
 import {ChainForkConfig} from "@lodestar/config";
 import {ForkName} from "@lodestar/params";
 import {
@@ -19,6 +19,7 @@ import {
   ssz,
   sszTypesFor,
 } from "@lodestar/types";
+import {MAX_BLOB_COMMITMENTS_PER_BLOCK} from "@lodestar/params";
 
 import {EmptyMeta, EmptyResponseCodec, EmptyResponseData} from "../../utils/codecs.js";
 import {getPostAltairForkTypes, getPostBellatrixForkTypes} from "../../utils/fork.js";
@@ -38,6 +39,18 @@ export const blobSidecarSSE = new ContainerType(
 );
 type BlobSidecarSSE = ValueOf<typeof blobSidecarSSE>;
 
+export const dataColumnSidecarSSE = new ContainerType(
+  {
+    blockRoot: stringType,
+    index: ssz.ColumnIndex,
+    slot: ssz.Slot,
+    kzgCommitments: new ListBasicType(stringType, MAX_BLOB_COMMITMENTS_PER_BLOCK),
+    versionedHashes: new ListBasicType(stringType, MAX_BLOB_COMMITMENTS_PER_BLOCK),
+  },
+  {typeName: "DataColumnSidecarSSE", jsonCase: "eth2"}
+);
+type DataColumnSidecarSSE = ValueOf<typeof dataColumnSidecarSSE>;
+
 export enum EventType {
   /**
    * The node has finished processing, resulting in a new head. previous_duty_dependent_root is
@@ -76,6 +89,8 @@ export enum EventType {
   payloadAttributes = "payload_attributes",
   /** The node has received a valid blobSidecar (from P2P or API) */
   blobSidecar = "blob_sidecar",
+  /** The node has received a valid DataColumnSidecar (from P2P or API) that passes all gossip validations */
+  dataColumnSidecar = "data_column_sidecar",
 }
 
 export const eventTypes: {[K in EventType]: K} = {
@@ -95,6 +110,7 @@ export const eventTypes: {[K in EventType]: K} = {
   [EventType.lightClientFinalityUpdate]: EventType.lightClientFinalityUpdate,
   [EventType.payloadAttributes]: EventType.payloadAttributes,
   [EventType.blobSidecar]: EventType.blobSidecar,
+  [EventType.dataColumnSidecar]: EventType.dataColumnSidecar,
 };
 
 export type EventData = {
@@ -143,6 +159,7 @@ export type EventData = {
   [EventType.lightClientFinalityUpdate]: {version: ForkName; data: LightClientFinalityUpdate};
   [EventType.payloadAttributes]: {version: ForkName; data: SSEPayloadAttributes};
   [EventType.blobSidecar]: BlobSidecarSSE;
+  [EventType.dataColumnSidecar]: DataColumnSidecarSSE;
 };
 
 export type BeaconEvent = {[K in EventType]: {type: K; message: EventData[K]}}[EventType];
@@ -296,6 +313,7 @@ export function getTypeByEvent(config: ChainForkConfig): {[K in EventType]: Type
     [EventType.contributionAndProof]: ssz.altair.SignedContributionAndProof,
     [EventType.payloadAttributes]: WithVersion((fork) => getPostBellatrixForkTypes(fork).SSEPayloadAttributes),
     [EventType.blobSidecar]: blobSidecarSSE,
+    [EventType.dataColumnSidecar]: dataColumnSidecarSSE,
 
     [EventType.lightClientOptimisticUpdate]: WithVersion(
       (fork) => getPostAltairForkTypes(fork).LightClientOptimisticUpdate

packages/api/test/unit/beacon/testData/events.ts

@@ -1,6 +1,6 @@
 import {ForkName} from "@lodestar/params";
 import {ssz} from "@lodestar/types";
-import {Endpoints, EventData, EventType, blobSidecarSSE} from "../../../../src/beacon/routes/events.js";
+import {Endpoints, EventData, EventType, blobSidecarSSE, dataColumnSidecarSSE} from "../../../../src/beacon/routes/events.js";
 import {GenericServerTestCases} from "../../../utils/genericServerTest.js";
 
 const abortController = new AbortController();
@@ -259,4 +259,13 @@ export const eventTestData: EventData = {
     slot: "1",
     versioned_hash: "0xcf8e0d4e9587369b2301d0790347320302cc0943d5a1884560367e8208d920f2",
   }),
+  [EventType.dataColumnSidecar]: dataColumnSidecarSSE.fromJson({
+    block_root: "0xcf8e0d4e9587369b2301d0790347320302cc0943d5a1884560367e8208d920f2",
+    index: "1",
+    slot: "1",
+    kzg_commitments: [
+      "0x1b66ac1fb663c9bc59509846d6ec05345bd908eda73e670af888da41af171505cc411d61252fb6cb3fa0017b679f8bb2305b26a285fa2737f175668d0dff91cc1b66ac1fb663c9bc59509846d6ec05345bd908eda73e670af888da41af171505",
+    ],
+    versioned_hashes: ["0xcf8e0d4e9587369b2301d0790347320302cc0943d5a1884560367e8208d920f2"],
+  }),
 };

packages/beacon-node/src/chain/blocks/blockInput/utils.ts

@@ -1,8 +1,13 @@
+import {routes} from "@lodestar/api";
 import {ChainForkConfig} from "@lodestar/config";
 import {ForkName, isForkPostDeneb} from "@lodestar/params";
 import {computeEpochAtSlot} from "@lodestar/state-transition";
 import {Epoch, Slot} from "@lodestar/types";
-import {BlobsSource, BlockSource as BlockSourceOld} from "../types.js";
+import {toHex, toHexString} from "@lodestar/utils";
+import {toRootHex} from "@lodestar/utils/lib/bytes/browser.js";
+import {kzgCommitmentToVersionedHash} from "../../../util/blobs.js";
+import {ChainEventEmitter} from "../../emitter.js";
+import {BlobsSource, BlockInput, BlockInputType, BlockSource as BlockSourceOld, NullBlockInput} from "../types.js";
 import {BlockInputSource as BlockSource} from "./types.js";
 
 export function isDaOutOfRange(
@@ -42,3 +47,29 @@ export function convertNewToOldBlobSource(source: BlockSource): BlobsSource {
       return BlobsSource.gossip;
   }
 }
+
+export function emitDataColumnSidecar(
+  emitter: ChainEventEmitter,
+  blockInput: BlockInput | NullBlockInput,
+  blockRoot: Uint8Array
+): void {
+  if (blockInput.block === null) return;
+  if (blockInput.type !== BlockInputType.availableData) return;
+  if (emitter.listenerCount(routes.events.EventType.dataColumnSidecar) === 0) return;
+
+  // TODO: Ideally it would be checked with ForkSeq > fulu but it's not returning right type
+  if (blockInput.blockData.fork !== ForkName.fulu) return;
+
+  const {dataColumns} = blockInput.blockData;
+  for (const dataColumnSidecar of dataColumns) {
+    const {index, kzgCommitments} = dataColumnSidecar;
+
+    emitter.emit(routes.events.EventType.dataColumnSidecar, {
+      blockRoot: toRootHex(blockRoot),
+      slot: blockInput.block.message.slot,
+      index,
+      kzgCommitments: kzgCommitments.map(toHex),
+      versionedHashes: kzgCommitments.map((commitment) => toHex(kzgCommitmentToVersionedHash(commitment))),
+    });
+  }
+}

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

@@ -30,6 +30,7 @@ import {ChainEvent, ReorgEventData} from "../emitter.js";
 import {ForkchoiceCaller} from "../forkChoice/index.js";
 import {REPROCESS_MIN_TIME_TO_NEXT_SLOT_SEC} from "../reprocess.js";
 import {toCheckpointHex} from "../stateCache/index.js";
+import {emitDataColumnSidecar} from "./blockInput/utils.js";
 import {AttestationImportOpt, BlockInputType, FullyVerifiedBlock, ImportBlockOpts} from "./types.js";
 import {getCheckpointFromState} from "./utils/checkpoint.js";
 import {writeBlockInputToDb} from "./writeBlockInputToDb.js";
@@ -139,7 +140,7 @@ export async function importBlock(
           });
         }
       } else if (blockData.fork === ForkName.fulu) {
-        // TODO peerDAS build and emit the event for the datacolumns
+        emitDataColumnSidecar(this.emitter, blockInput, blockRoot);
       }
     }
   });
@@ -469,24 +470,23 @@ export async function importBlock(
       }
       if (
         blockInput.type === BlockInputType.availableData &&
-        this.emitter.listenerCount(routes.events.EventType.blobSidecar)
+        this.emitter.listenerCount(routes.events.EventType.blobSidecar) &&
+        (blockInput.blockData.fork === ForkName.deneb || blockInput.blockData.fork === ForkName.electra)
       ) {
-        if (blockInput.blockData.fork === ForkName.deneb || blockInput.blockData.fork === ForkName.electra) {
-          const {blobs} = blockInput.blockData;
-          for (const blobSidecar of blobs) {
-            const {index, kzgCommitment} = blobSidecar;
-            this.emitter.emit(routes.events.EventType.blobSidecar, {
-              blockRoot: blockRootHex,
-              slot: blockSlot,
-              index,
-              kzgCommitment: toHexString(kzgCommitment),
-              versionedHash: toHexString(kzgCommitmentToVersionedHash(kzgCommitment)),
-            });
-          }
-        } else {
-          // TODO add event for datacolumns
+        const {blobs} = blockInput.blockData;
+        for (const blobSidecar of blobs) {
+          const {index, kzgCommitment} = blobSidecar;
+          this.emitter.emit(routes.events.EventType.blobSidecar, {
+            blockRoot: blockRootHex,
+            slot: blockSlot,
+            index,
+            kzgCommitment: toHexString(kzgCommitment),
+            versionedHash: toHexString(kzgCommitmentToVersionedHash(kzgCommitment)),
+          });
         }
       }
+
+      emitDataColumnSidecar(this.emitter, blockInput, blockRoot);
     });
   }
 

packages/beacon-node/src/network/processor/gossipHandlers.ts

@@ -14,7 +14,8 @@ import {
   ssz,
   sszTypesFor,
 } from "@lodestar/types";
-import {LogLevel, Logger, prettyBytes, toRootHex} from "@lodestar/utils";
+import {LogLevel, Logger, prettyBytes, toHex, toRootHex} from "@lodestar/utils";
+import {emitDataColumnSidecar} from "../../chain/blocks/blockInput/utils.js";
 import {
   BlobSidecarValidation,
   BlockInput,
@@ -58,6 +59,7 @@ import {validateLightClientFinalityUpdate} from "../../chain/validation/lightCli
 import {validateLightClientOptimisticUpdate} from "../../chain/validation/lightClientOptimisticUpdate.js";
 import {OpSource} from "../../chain/validatorMonitor.js";
 import {Metrics} from "../../metrics/index.js";
+import {kzgCommitmentToVersionedHash} from "../../util/blobs.js";
 import {INetworkCore} from "../core/index.js";
 import {NetworkEvent, NetworkEventBus} from "../events.js";
 import {
@@ -311,6 +313,9 @@ function getSequentialHandlers(modules: ValidatorFnsModules, options: GossipHand
         validationTime,
       });
 
+      // Emit data column sidecar event for successfully validated gossip data column
+      emitDataColumnSidecar(chain.emitter, blockInput, blockRoot);
+
       return blockInput;
     } catch (e) {
       if (e instanceof DataColumnSidecarGossipError) {

packages/beacon-node/test/unit/api/impl/events/events.test.ts

@@ -1,6 +1,7 @@
 import {routes} from "@lodestar/api";
 import {config} from "@lodestar/config/default";
 import {ssz} from "@lodestar/types";
+import {toHexString} from "@lodestar/utils";
 import {MockedObject, afterEach, beforeEach, describe, expect, it, vi} from "vitest";
 import {getEventsApi} from "../../../../../src/api/impl/events/index.js";
 import {BeaconChain, ChainEventEmitter, HeadEventData} from "../../../../../src/chain/index.js";
@@ -70,5 +71,23 @@ describe("Events api impl", () => {
       expect(events[0].type).toBe(routes.events.EventType.head);
       expect(events[0].message).not.toBeNull();
     });
+
+    it("should emit dataColumnSidecar event", async () => {
+      const events = getEvents([routes.events.EventType.dataColumnSidecar]);
+
+      const mockDataColumnSidecarEvent = {
+        blockRoot: ZERO_HASH_HEX,
+        slot: 123,
+        index: 1,
+        kzgCommitments: [toHexString(ssz.deneb.KZGCommitment.defaultValue())],
+        versionedHashes: [ZERO_HASH_HEX],
+      };
+
+      chainEventEmmitter.emit(routes.events.EventType.dataColumnSidecar, mockDataColumnSidecarEvent);
+
+      expect(events).toHaveLength(1);
+      expect(events[0].type).toBe(routes.events.EventType.dataColumnSidecar);
+      expect(events[0].message).toEqual(mockDataColumnSidecarEvent);
+    });
   });
 });