feat: add and use metadatav3 for peer custody subnet

fixes for metadata, working locally

change the condition to update metadata csc change

Author: g11tech

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

@@ -1,7 +1,7 @@
 import {BitArray} from "@chainsafe/ssz";
 import {ForkSeq} from "@lodestar/params";
 import {computeStartSlotAtEpoch} from "@lodestar/state-transition";
-import {altair, Epoch, phase0, ssz} from "@lodestar/types";
+import {altair, Epoch, phase0, ssz, electra} from "@lodestar/types";
 import {BeaconConfig} from "@lodestar/config";
 import {FAR_FUTURE_EPOCH} from "../constants/index.js";
 import {getCurrentAndNextFork} from "./forks.js";
@@ -19,7 +19,7 @@ export enum SubnetType {
 }
 
 export type MetadataOpts = {
-  metadata?: altair.Metadata;
+  metadata?: electra.Metadata;
 };
 
 export type MetadataModules = {
@@ -35,12 +35,15 @@ export type MetadataModules = {
 export class MetadataController {
   private onSetValue: (key: string, value: Uint8Array) => void;
   private config: BeaconConfig;
-  private _metadata: altair.Metadata;
+  private _metadata: electra.Metadata;
 
   constructor(opts: MetadataOpts, modules: MetadataModules) {
     this.config = modules.config;
     this.onSetValue = modules.onSetValue;
-    this._metadata = opts.metadata || ssz.altair.Metadata.defaultValue();
+    this._metadata = opts.metadata ?? {
+      ...ssz.electra.Metadata.defaultValue(),
+      csc: Math.max(this.config.CUSTODY_REQUIREMENT, this.config.NODE_CUSTODY_REQUIREMENT),
+    };
   }
 
   upstreamValues(currentEpoch: Epoch): void {
@@ -54,6 +57,10 @@ export class MetadataController {
       // adding syncnets to the ENR is not a problem, we will just have a useless field for a few hours.
       this.onSetValue(ENRKey.syncnets, ssz.phase0.AttestationSubnets.serialize(this._metadata.syncnets));
     }
+
+    if (this.config.getForkSeq(computeStartSlotAtEpoch(currentEpoch)) >= ForkSeq.electra) {
+      this.onSetValue(ENRKey.csc, ssz.Uint8.serialize(this._metadata.csc));
+    }
   }
 
   get seqNumber(): bigint {
@@ -79,8 +86,17 @@ export class MetadataController {
     this._metadata.attnets = attnets;
   }
 
+  get csc(): number {
+    return this._metadata.csc;
+  }
+
+  set csc(csc: number) {
+    this.onSetValue(ENRKey.csc, ssz.Uint8.serialize(csc));
+    this._metadata.csc = csc;
+  }
+
   /** Consumers that need the phase0.Metadata type can just ignore the .syncnets property */
-  get json(): altair.Metadata {
+  get json(): electra.Metadata {
     return this._metadata;
   }
 

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

@@ -684,6 +684,7 @@ export class Network implements INetwork {
   };
 
   private onPeerConnected = (data: NetworkEventData[NetworkEvent.peerConnected]): void => {
+    this.logger.warn("onPeerConnected", {peer: data.peer, dataColumns: data.dataColumns.join(",")});
     this.connectedPeers.set(data.peer, data.dataColumns);
   };
 

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

@@ -2,7 +2,7 @@ import {Connection, PeerId} from "@libp2p/interface";
 import {BitArray, toHexString} from "@chainsafe/ssz";
 import {SYNC_COMMITTEE_SUBNET_COUNT} from "@lodestar/params";
 import {BeaconConfig} from "@lodestar/config";
-import {Metadata, altair, phase0} from "@lodestar/types";
+import {Metadata, electra, phase0} from "@lodestar/types";
 import {withTimeout} from "@lodestar/utils";
 import {LoggerNode} from "@lodestar/logger/node";
 import {GoodByeReasonCode, GOODBYE_KNOWN_CODES, Libp2pEvent} from "../../constants/index.js";
@@ -308,6 +308,11 @@ export class PeerManager {
   private onPing(peer: PeerId, seqNumber: phase0.Ping): void {
     // if the sequence number is unknown update the peer's metadata
     const metadata = this.connectedPeers.get(peer.toString())?.metadata;
+    this.logger.warn("onPing", {
+      seqNumber,
+      metaSeqNumber: metadata?.seqNumber,
+      cond: !metadata || metadata.seqNumber < seqNumber,
+    });
     if (!metadata || metadata.seqNumber < seqNumber) {
       void this.requestMetadata(peer);
     }
@@ -320,12 +325,22 @@ export class PeerManager {
     // Store metadata always in case the peer updates attnets but not the sequence number
     // Trust that the peer always sends the latest metadata (From Lighthouse)
     const peerData = this.connectedPeers.get(peer.toString());
+    this.logger.warn("onMetadata", {peer: peer.toString(), peerData: peerData !== undefined});
+    console.log("onMetadata", metadata);
     if (peerData) {
+      const oldMetadata = peerData.metadata;
       peerData.metadata = {
         seqNumber: metadata.seqNumber,
         attnets: metadata.attnets,
-        syncnets: (metadata as Partial<altair.Metadata>).syncnets ?? BitArray.fromBitLen(SYNC_COMMITTEE_SUBNET_COUNT),
+        syncnets: (metadata as Partial<electra.Metadata>).syncnets ?? BitArray.fromBitLen(SYNC_COMMITTEE_SUBNET_COUNT),
+        csc:
+          (metadata as Partial<electra.Metadata>).csc ??
+          this.discovery?.["peerIdToCustodySubnetCount"].get(peer.toString()) ??
+          this.config.CUSTODY_REQUIREMENT,
       };
+      if (oldMetadata === null || oldMetadata.csc !== peerData.metadata.csc) {
+        void this.requestStatus(peer, this.statusCache.get());
+      }
     }
   }
 
@@ -392,10 +407,10 @@ export class PeerManager {
     }
     if (getConnection(this.libp2p, peer.toString())) {
       const nodeId = peerData?.nodeId ?? computeNodeId(peer);
-      const custodySubnetCount =
-        peerData?.custodySubnetCount ?? this.discovery?.["peerIdToCustodySubnetCount"].get(peer.toString());
+      console.log("onStatus", peerData?.metadata);
+      const custodySubnetCount = peerData?.metadata?.csc;
 
-      const peerCustodySubnetCount = custodySubnetCount ?? this.config.CUSTODY_REQUIREMENT;
+      const peerCustodySubnetCount = custodySubnetCount ?? 4;
       const peerCustodySubnets = getCustodyColumnSubnets(nodeId, peerCustodySubnetCount);
 
       const matchingSubnetsNum = this.custodySubnets.reduce(
@@ -407,6 +422,7 @@ export class PeerManager {
 
       this.logger.warn(`onStatus ${custodySubnetCount == undefined ? "undefined custody count assuming 4" : ""}`, {
         nodeId: toHexString(nodeId),
+        myNodeId: toHexString(this.nodeId),
         peerId: peer.toString(),
         custodySubnetCount,
         hasAllColumns,
@@ -438,14 +454,17 @@ export class PeerManager {
 
   private async requestMetadata(peer: PeerId): Promise<void> {
     try {
+      this.logger.warn("requestMetadata", {peer: peer.toString()});
       this.onMetadata(peer, await this.reqResp.sendMetadata(peer));
     } catch (e) {
+      console.log("requestMetadata", e);
       // TODO: Downvote peer here or in the reqResp layer
     }
   }
 
   private async requestPing(peer: PeerId): Promise<void> {
     try {
+      this.logger.warn("requestPing", {peer: peer.toString()});
       this.onPing(peer, await this.reqResp.sendPing(peer));
 
       // If peer replies a PING request also update lastReceivedMsg
@@ -643,7 +662,6 @@ export class PeerManager {
     // If that happens, it's okay. Only the "outbound" connection triggers immediate action
     const now = Date.now();
     const nodeId = computeNodeId(remotePeer);
-    const custodySubnetCount = this.discovery?.["peerIdToCustodySubnetCount"].get(remotePeer.toString()) ?? null;
     const peerData: PeerData = {
       lastReceivedMsgUnixTsMs: direction === "outbound" ? 0 : now,
       // If inbound, request after STATUS_INBOUND_GRACE_PERIOD
@@ -657,12 +675,11 @@ export class PeerManager {
       agentVersion: null,
       agentClient: null,
       encodingPreference: null,
-      custodySubnetCount,
     };
     this.connectedPeers.set(remotePeer.toString(), peerData);
 
     if (direction === "outbound") {
-      //this.pingAndStatusTimeouts();
+      // this.pingAndStatusTimeouts();
       void this.requestPing(remotePeer);
       void this.requestStatus(remotePeer, this.statusCache.get());
     }

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

@@ -1,5 +1,5 @@
 import {PeerId} from "@libp2p/interface";
-import {altair} from "@lodestar/types";
+import {electra} from "@lodestar/types";
 import {Encoding} from "@lodestar/reqresp";
 import {NodeId} from "../subnets/interface.js";
 import {ClientKind} from "./client.js";
@@ -20,11 +20,10 @@ export type PeerData = {
   direction: "inbound" | "outbound";
   peerId: PeerId;
   nodeId: NodeId | null;
-  metadata: altair.Metadata | null;
+  metadata: electra.Metadata | null;
   agentVersion: string | null;
   agentClient: ClientKind | null;
   encodingPreference: Encoding | null;
-  custodySubnetCount: number | null;
 };
 
 /**

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

@@ -192,7 +192,11 @@ export class ReqRespBeaconNode extends ReqResp {
         peerId,
         ReqRespMethod.Metadata,
         // Before altair, prioritize V2. After altair only request V2
-        this.currentRegisteredFork >= ForkSeq.altair ? [Version.V2] : [(Version.V2, Version.V1)],
+        this.currentRegisteredFork >= ForkSeq.electra
+          ? [Version.V3]
+          : this.currentRegisteredFork >= ForkSeq.altair
+            ? [Version.V3, Version.V2]
+            : [(Version.V3, Version.V2, Version.V1)],
         null
       ),
       responseSszTypeByMethod[ReqRespMethod.Metadata]
@@ -257,6 +261,7 @@ export class ReqRespBeaconNode extends ReqResp {
 
     if (ForkSeq[fork] >= ForkSeq.electra) {
       protocolsAtFork.push(
+        [protocols.MetadataV3(this.config), this.onMetadata.bind(this)],
         [protocols.DataColumnSidecarsByRoot(this.config), this.getHandler(ReqRespMethod.DataColumnSidecarsByRoot)],
         [protocols.DataColumnSidecarsByRange(this.config), this.getHandler(ReqRespMethod.DataColumnSidecarsByRange)]
       );

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

@@ -23,6 +23,12 @@ export const MetadataV2 = toProtocol({
   contextBytesType: ContextBytesType.Empty,
 });
 
+export const MetadataV3 = toProtocol({
+  method: ReqRespMethod.Metadata,
+  version: Version.V3,
+  contextBytesType: ContextBytesType.Empty,
+});
+
 export const Ping = toProtocol({
   method: ReqRespMethod.Ping,
   version: Version.V1,

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

@@ -97,7 +97,8 @@ export const responseSszTypeByMethod: {[K in ReqRespMethod]: ResponseTypeGetter<
   [ReqRespMethod.Status]: () => ssz.phase0.Status,
   [ReqRespMethod.Goodbye]: () => ssz.phase0.Goodbye,
   [ReqRespMethod.Ping]: () => ssz.phase0.Ping,
-  [ReqRespMethod.Metadata]: (_, version) => (version == Version.V1 ? ssz.phase0.Metadata : ssz.altair.Metadata),
+  [ReqRespMethod.Metadata]: (_, version) =>
+    version == Version.V1 ? ssz.phase0.Metadata : version == Version.V2 ? ssz.altair.Metadata : ssz.electra.Metadata,
   [ReqRespMethod.BeaconBlocksByRange]: blocksResponseType,
   [ReqRespMethod.BeaconBlocksByRoot]: blocksResponseType,
   [ReqRespMethod.BlobSidecarsByRange]: () => ssz.deneb.BlobSidecar,
@@ -126,6 +127,7 @@ export type RequestTypedContainer = {
 export enum Version {
   V1 = 1,
   V2 = 2,
+  V3 = 3,
 }
 
 export type OutgoingRequestArgs = {

packages/types/src/electra/sszTypes.ts

@@ -10,10 +10,19 @@ import {
 
 import {ssz as primitiveSsz} from "../primitive/index.js";
 import {ssz as phase0Ssz} from "../phase0/index.js";
+import {ssz as altariSsz} from "../altair/index.js";
 import {ssz as denebSsz} from "../deneb/index.js";
 
 const {BLSSignature, Root, ColumnIndex, Bytes32, Slot, UintNum64} = primitiveSsz;
 
+export const Metadata = new ContainerType(
+  {
+    ...altariSsz.Metadata.fields,
+    csc: UintNum64,
+  },
+  {typeName: "Metadata", jsonCase: "eth2"}
+);
+
 export const Cell = new ByteVectorType(BYTES_PER_FIELD_ELEMENT * FIELD_ELEMENTS_PER_CELL);
 export const DataColumn = new ListCompositeType(Cell, MAX_BLOB_COMMITMENTS_PER_BLOCK);
 export const ExtendedMatrix = new ListCompositeType(Cell, MAX_BLOB_COMMITMENTS_PER_BLOCK * NUMBER_OF_COLUMNS);
@@ -200,3 +209,17 @@ export const SSEPayloadAttributes = new ContainerType(
   },
   {typeName: "SSEPayloadAttributes", jsonCase: "eth2"}
 );
+
+export const BlockContents = new ContainerType(
+  {
+    ...denebSsz.BlockContents.fields,
+  },
+  {typeName: "BlockContents", jsonCase: "eth2"}
+);
+
+export const SignedBlockContents = new ContainerType(
+  {
+    ...denebSsz.SignedBlockContents.fields,
+  },
+  {typeName: "SignedBlockContents", jsonCase: "eth2"}
+);

packages/types/src/electra/types.ts

@@ -1,6 +1,8 @@
 import {ValueOf} from "@chainsafe/ssz";
 import * as ssz from "./sszTypes.js";
 
+export type Metadata = ValueOf<typeof ssz.Metadata>;
+
 export type Cell = ValueOf<typeof ssz.Cell>;
 export type DataColumn = ValueOf<typeof ssz.DataColumn>;
 export type ExtendedMatrix = ValueOf<typeof ssz.ExtendedMatrix>;
@@ -37,3 +39,7 @@ export type LightClientUpdate = ValueOf<typeof ssz.LightClientUpdate>;
 export type LightClientFinalityUpdate = ValueOf<typeof ssz.LightClientFinalityUpdate>;
 export type LightClientOptimisticUpdate = ValueOf<typeof ssz.LightClientOptimisticUpdate>;
 export type LightClientStore = ValueOf<typeof ssz.LightClientStore>;
+
+export type BlockContents = ValueOf<typeof ssz.BlockContents>;
+export type SignedBlockContents = ValueOf<typeof ssz.SignedBlockContents>;
+export type Contents = Omit<BlockContents, "block">;

packages/types/src/types.ts

@@ -141,7 +141,7 @@ type TypesByFork = {
     BeaconBlockBody: electra.BeaconBlockBody;
     BeaconState: electra.BeaconState;
     SignedBeaconBlock: electra.SignedBeaconBlock;
-    Metadata: altair.Metadata;
+    Metadata: electra.Metadata;
     LightClientHeader: electra.LightClientHeader;
     LightClientBootstrap: electra.LightClientBootstrap;
     LightClientUpdate: electra.LightClientUpdate;
@@ -156,12 +156,8 @@ type TypesByFork = {
     BuilderBid: electra.BuilderBid;
     SignedBuilderBid: electra.SignedBuilderBid;
     SSEPayloadAttributes: electra.SSEPayloadAttributes;
-    BlockContents: {block: BeaconBlock<ForkName.electra>; kzgProofs: deneb.KZGProofs; blobs: deneb.Blobs};
-    SignedBlockContents: {
-      signedBlock: SignedBeaconBlock<ForkName.electra>;
-      kzgProofs: deneb.KZGProofs;
-      blobs: deneb.Blobs;
-    };
+    BlockContents: electra.BlockContents;
+    SignedBlockContents: electra.SignedBlockContents;
     ExecutionPayloadAndBlobsBundle: deneb.ExecutionPayloadAndBlobsBundle;
     BlobsBundle: deneb.BlobsBundle;
     Contents: deneb.Contents;