Connect final eip-4844 networking parts

packages/beacon-node/src/network/gossip/gossipsub.ts

@@ -6,7 +6,7 @@ import {PeerScore, PeerScoreParams} from "@chainsafe/libp2p-gossipsub/score";
 import {MetricsRegister, TopicLabel, TopicStrToLabel} from "@chainsafe/libp2p-gossipsub/metrics";
 import {IBeaconConfig} from "@lodestar/config";
 import {ATTESTATION_SUBNET_COUNT, ForkName, SYNC_COMMITTEE_SUBNET_COUNT} from "@lodestar/params";
-import {allForks, altair, phase0} from "@lodestar/types";
+import {allForks, altair, phase0, eip4844} from "@lodestar/types";
 import {ILogger, Map2d, Map2dArr} from "@lodestar/utils";
 import {computeStartSlotAtEpoch} from "@lodestar/state-transition";
 
@@ -200,6 +200,14 @@ export class Eth2Gossipsub extends GossipSub {
     await this.publishObject<GossipType.beacon_block>({type: GossipType.beacon_block, fork}, signedBlock);
   }
 
+  async publishSignedBeaconBlockAndBlobsSidecar(item: eip4844.SignedBeaconBlockAndBlobsSidecar): Promise<void> {
+    const fork = this.config.getForkName(item.beaconBlock.message.slot);
+    await this.publishObject<GossipType.beacon_block_and_blobs_sidecar>(
+      {type: GossipType.beacon_block_and_blobs_sidecar, fork},
+      item
+    );
+  }
+
   async publishBeaconAggregateAndProof(aggregateAndProof: phase0.SignedAggregateAndProof): Promise<number> {
     const fork = this.config.getForkName(aggregateAndProof.message.aggregate.data.slot);
     return await this.publishObject<GossipType.beacon_aggregate_and_proof>(
@@ -421,6 +429,11 @@ function getMetricsTopicStrToLabel(config: IBeaconConfig): TopicStrToLabel {
     topics.push({fork, type: GossipType.proposer_slashing});
     topics.push({fork, type: GossipType.attester_slashing});
     topics.push({fork, type: GossipType.sync_committee_contribution_and_proof});
+
+    // TODO EIP-4844: It's an issue to pre-declare the topic here before the fork?
+    if (config.EIP4844_FORK_EPOCH < Infinity) {
+      topics.push({fork, type: GossipType.beacon_block_and_blobs_sidecar});
+    }
   }
 
   for (const topic of topics) {

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

@@ -8,7 +8,7 @@ import {ILogger, sleep} from "@lodestar/utils";
 import {ATTESTATION_SUBNET_COUNT, ForkName, ForkSeq, SYNC_COMMITTEE_SUBNET_COUNT} from "@lodestar/params";
 import {Discv5, ENR} from "@chainsafe/discv5";
 import {computeEpochAtSlot, computeTimeAtSlot} from "@lodestar/state-transition";
-import {altair, Epoch, phase0} from "@lodestar/types";
+import {altair, eip4844, Epoch, phase0} from "@lodestar/types";
 import {IMetrics} from "../metrics/index.js";
 import {ChainEvent, IBeaconChain, IBeaconClock} from "../chain/index.js";
 import {BlockInput, BlockInputType, getBlockInput} from "../chain/blocks/types.js";
@@ -207,17 +207,19 @@ export class Network implements INetwork {
     return this.peerManager.hasSomeConnectedPeer();
   }
 
-  publishBeaconBlockMaybeBlobs(blockImport: BlockInput): Promise<void> {
-    switch (blockImport.type) {
+  publishBeaconBlockMaybeBlobs(blockInput: BlockInput): Promise<void> {
+    switch (blockInput.type) {
       case BlockInputType.preEIP4844:
-        return this.gossip.publishBeaconBlock(blockImport.block);
+        return this.gossip.publishBeaconBlock(blockInput.block);
 
       case BlockInputType.postEIP4844:
-        // TODO EIP-4844: Implement SignedBeaconBlockAndBlobsSidecar publish topic
-        throw Error("SignedBeaconBlockAndBlobsSidecar publish not implemented");
+        return this.gossip.publishSignedBeaconBlockAndBlobsSidecar({
+          beaconBlock: blockInput.block as eip4844.SignedBeaconBlock,
+          blobsSidecar: blockInput.blobs,
+        });
 
       case BlockInputType.postEIP4844OldBlobs:
-        throw Error(`Attempting to broadcast old BlockImport slot ${blockImport.block.message.slot}`);
+        throw Error(`Attempting to broadcast old BlockImport slot ${blockInput.block.message.slot}`);
     }
   }
 
@@ -245,7 +247,7 @@ export class Network implements INetwork {
         throw Error(`blocks.length ${blocks.length} != blobsSidecars.length ${blobsSidecars.length}`);
       }
 
-      const blockImports: BlockInput[] = [];
+      const blockInput: BlockInput[] = [];
       for (let i = 0; i < blocks.length; i++) {
         const block = blocks[i];
         const blobsSidecar = blobsSidecars[i];
@@ -255,9 +257,9 @@ export class Network implements INetwork {
           throw Error(`blob does not match block slot ${block.message.slot} != ${blobsSidecar.beaconBlockSlot}`);
         }
 
-        blockImports.push(getBlockInput.postEIP4844(this.config, block, blobsSidecar));
+        blockInput.push(getBlockInput.postEIP4844(this.config, block, blobsSidecar));
       }
-      return blockImports;
+      return blockInput;
     }
 
     // Post EIP-4844 but old blobs
@@ -474,13 +476,20 @@ export class Network implements INetwork {
   private coreTopicsAtFork(fork: ForkName): GossipTopicTypeMap[keyof GossipTopicTypeMap][] {
     // Common topics for all forks
     const topics: GossipTopicTypeMap[keyof GossipTopicTypeMap][] = [
-      {type: GossipType.beacon_block},
+      // {type: GossipType.beacon_block}, // Handled below
       {type: GossipType.beacon_aggregate_and_proof},
       {type: GossipType.voluntary_exit},
       {type: GossipType.proposer_slashing},
       {type: GossipType.attester_slashing},
     ];
 
+    // After EIP4844 only track beacon_block_and_blobs_sidecar topic
+    if (ForkSeq[fork] < ForkSeq.eip4844) {
+      topics.push({type: GossipType.beacon_block});
+    } else {
+      topics.push({type: GossipType.beacon_block_and_blobs_sidecar});
+    }
+
     // Any fork after altair included
     if (ForkSeq[fork] >= ForkSeq.altair) {
       topics.push({type: GossipType.sync_committee_contribution_and_proof});

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

@@ -322,6 +322,13 @@ export class ReqRespBeaconNode extends ReqResp implements IReqRespBeaconNode {
       );
     }
 
+    if (ForkSeq[fork] >= ForkSeq.eip4844) {
+      protocols.push(
+        messages.BeaconBlockAndBlobsSidecarByRoot(modules, this.reqRespHandlers.onBeaconBlockAndBlobsSidecarByRoot),
+        messages.BlobsSidecarsByRange(modules, this.reqRespHandlers.onBlobsSidecarsByRange)
+      );
+    }
+
     return protocols;
   }
 

packages/beacon-node/src/network/reqresp/handlers/index.ts

@@ -4,6 +4,8 @@ import {IBeaconChain} from "../../../chain/index.js";
 import {IBeaconDb} from "../../../db/index.js";
 import {onBeaconBlocksByRange} from "./beaconBlocksByRange.js";
 import {onBeaconBlocksByRoot} from "./beaconBlocksByRoot.js";
+import {onBeaconBlockAndBlobsSidecarByRoot} from "./beaconBlockAndBlobsSidecarByRoot.js";
+import {onBlobsSidecarsByRange} from "./blobsSidecarsByRange.js";
 import {onLightClientBootstrap} from "./lightClientBootstrap.js";
 import {onLightClientFinalityUpdate} from "./lightClientFinalityUpdate.js";
 import {onLightClientOptimisticUpdate} from "./lightClientOptimisticUpdate.js";
@@ -14,6 +16,8 @@ export interface ReqRespHandlers {
   onStatus: HandlerTypeFromMessage<typeof messages.Status>;
   onBeaconBlocksByRange: HandlerTypeFromMessage<typeof messages.BeaconBlocksByRange>;
   onBeaconBlocksByRoot: HandlerTypeFromMessage<typeof messages.BeaconBlocksByRoot>;
+  onBeaconBlockAndBlobsSidecarByRoot: HandlerTypeFromMessage<typeof messages.BeaconBlockAndBlobsSidecarByRoot>;
+  onBlobsSidecarsByRange: HandlerTypeFromMessage<typeof messages.BlobsSidecarsByRange>;
   onLightClientBootstrap: HandlerTypeFromMessage<typeof messages.LightClientBootstrap>;
   onLightClientUpdatesByRange: HandlerTypeFromMessage<typeof messages.LightClientUpdatesByRange>;
   onLightClientFinalityUpdate: HandlerTypeFromMessage<typeof messages.LightClientFinalityUpdate>;
@@ -34,6 +38,12 @@ export function getReqRespHandlers({db, chain}: {db: IBeaconDb; chain: IBeaconCh
     async *onBeaconBlocksByRoot(req) {
       yield* onBeaconBlocksByRoot(req, chain, db);
     },
+    async *onBeaconBlockAndBlobsSidecarByRoot(req) {
+      yield* onBeaconBlockAndBlobsSidecarByRoot(req, chain, db);
+    },
+    async *onBlobsSidecarsByRange(req) {
+      yield* onBlobsSidecarsByRange(req, chain, db);
+    },
     async *onLightClientBootstrap(req) {
       yield* onLightClientBootstrap(req, chain);
     },

packages/beacon-node/test/e2e/network/reqresp.test.ts

@@ -97,6 +97,8 @@ describe("network / ReqResp", function () {
       } as HandlerTypeFromMessage<typeof messages.Status>,
       onBeaconBlocksByRange: notImplemented,
       onBeaconBlocksByRoot: notImplemented,
+      onBlobsSidecarsByRange: notImplemented,
+      onBeaconBlockAndBlobsSidecarByRoot: notImplemented,
       onLightClientBootstrap: notImplemented,
       onLightClientUpdatesByRange: notImplemented,
       onLightClientOptimisticUpdate: notImplemented,

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

@@ -138,10 +138,10 @@ export const forkChoiceTest: TestRunnerFn<ForkChoiceTestCase, void> = (fork) =>
               isValid,
             });
 
-            const blockImport = getBlockInput.preEIP4844(config, signedBlock);
+            const blockInput = getBlockInput.preEIP4844(config, signedBlock);
 
             try {
-              await chain.processBlock(blockImport, {seenTimestampSec: tickTime});
+              await chain.processBlock(blockInput, {seenTimestampSec: tickTime});
               if (!isValid) throw Error("Expect error since this is a negative test");
             } catch (e) {
               if (isValid) throw e;

packages/beacon-node/test/utils/node/beacon.ts

@@ -6,8 +6,8 @@ import {config as minimalConfig} from "@lodestar/config/default";
 import {createIBeaconConfig, createIChainForkConfig, IChainConfig} from "@lodestar/config";
 import {ILogger, RecursivePartial} from "@lodestar/utils";
 import {LevelDbController} from "@lodestar/db";
-import {phase0} from "@lodestar/types";
-import {GENESIS_SLOT} from "@lodestar/params";
+import {phase0, ssz} from "@lodestar/types";
+import {ForkSeq, GENESIS_SLOT} from "@lodestar/params";
 import {BeaconStateAllForks} from "@lodestar/state-transition";
 import {isPlainObject} from "@lodestar/utils";
 import {createKeypairFromPeerId, ENR} from "@chainsafe/discv5";
@@ -91,6 +91,12 @@ export async function getDevBeaconNode(
     const block = config.getForkTypes(GENESIS_SLOT).SignedBeaconBlock.defaultValue();
     block.message.stateRoot = state.hashTreeRoot();
     await db.blockArchive.add(block);
+
+    if (config.getForkSeq(GENESIS_SLOT) >= ForkSeq.eip4844) {
+      const blobsSidecar = ssz.eip4844.BlobsSidecar.defaultValue();
+      blobsSidecar.beaconBlockRoot = config.getForkTypes(GENESIS_SLOT).BeaconBlock.hashTreeRoot(block.message);
+      await db.blobsSidecar.add(blobsSidecar);
+    }
   }
 
   const beaconConfig = createIBeaconConfig(config, anchorState.genesisValidatorsRoot);