refactor(reqresp)!: support byte based handlers

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

@@ -68,10 +68,6 @@ export class MetadataController {
     return this._metadata.seqNumber;
   }
 
-  get currentSlot(): number {
-    return this.chain.clock.currentSlot;
-  }
-
   get syncnets(): BitArray {
     return this._metadata.syncnets;
   }

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

@@ -1,16 +1,15 @@
-import {ContextBytesType, EncodedPayloadBytes, EncodedPayloadType, ProtocolDescriptor} from "@lodestar/reqresp";
+import {ResponseOutgoing} from "@lodestar/reqresp";
 import {deneb} from "@lodestar/types";
 import {toHex} from "@lodestar/utils";
 import {IBeaconChain} from "../../../chain/index.js";
 import {IBeaconDb} from "../../../db/index.js";
 import {getSlotFromSignedBeaconBlockSerialized} from "../../../util/sszBytes.js";
 
 export async function* onBeaconBlockAndBlobsSidecarByRoot(
-  _protocol: ProtocolDescriptor<deneb.BeaconBlockAndBlobsSidecarByRootRequest, deneb.SignedBeaconBlockAndBlobsSidecar>,
   requestBody: deneb.BeaconBlockAndBlobsSidecarByRootRequest,
   chain: IBeaconChain,
   db: IBeaconDb
-): AsyncIterable<EncodedPayloadBytes> {
+): AsyncIterable<ResponseOutgoing> {
   const finalizedSlot = chain.forkChoice.getFinalizedBlock().slot;
 
   for (const blockRoot of requestBody) {
@@ -40,15 +39,10 @@ export async function* onBeaconBlockAndBlobsSidecarByRoot(
     if (forkSlot === null) {
       throw Error(`Invalid block bytes for block ${blockRootHex}`);
     }
-    const fork = chain.config.getForkName(forkSlot);
 
     yield {
-      type: EncodedPayloadType.bytes,
-      bytes: signedBeaconBlockAndBlobsSidecarFromBytes(blockBytes, blobsSidecarBytes),
-      contextBytes: {
-        type: ContextBytesType.ForkDigest,
-        fork,
-      },
+      data: signedBeaconBlockAndBlobsSidecarFromBytes(blockBytes, blobsSidecarBytes),
+      fork: chain.config.getForkName(forkSlot),
     };
   }
 }

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

@@ -1,45 +1,31 @@
 import {GENESIS_SLOT, MAX_REQUEST_BLOCKS} from "@lodestar/params";
-import {
-  ContextBytesType,
-  EncodedPayloadBytes,
-  EncodedPayloadType,
-  ProtocolDescriptor,
-  ResponseError,
-  RespStatus,
-} from "@lodestar/reqresp";
-import {allForks, deneb, phase0} from "@lodestar/types";
+import {ResponseError, ResponseOutgoing, RespStatus} from "@lodestar/reqresp";
+import {deneb, phase0} from "@lodestar/types";
 import {fromHex} from "@lodestar/utils";
 import {IBeaconChain} from "../../../chain/index.js";
 import {IBeaconDb} from "../../../db/index.js";
 
 // TODO: Unit test
 
 export function onBeaconBlocksByRange(
-  protocol:
-    | ProtocolDescriptor<phase0.BeaconBlocksByRangeRequest, allForks.SignedBeaconBlock>
-    | ProtocolDescriptor<deneb.BlobsSidecarsByRangeRequest, allForks.SignedBeaconBlock>,
   request: phase0.BeaconBlocksByRangeRequest,
   chain: IBeaconChain,
   db: IBeaconDb
-): AsyncIterable<EncodedPayloadBytes> {
-  return onBlocksOrBlobsSidecarsByRange(protocol, request, chain, {
+): AsyncIterable<ResponseOutgoing> {
+  return onBlocksOrBlobsSidecarsByRange(request, chain, {
     finalized: db.blockArchive,
     unfinalized: db.block,
   });
 }
 
 export async function* onBlocksOrBlobsSidecarsByRange(
-  protocol:
-    | ProtocolDescriptor<deneb.BlobsSidecarsByRangeRequest, allForks.SignedBeaconBlock>
-    | ProtocolDescriptor<deneb.BlobsSidecarsByRangeRequest, deneb.SignedBeaconBlockAndBlobsSidecar>
-    | ProtocolDescriptor<deneb.BlobsSidecarsByRangeRequest, deneb.BlobsSidecar>,
   request: deneb.BlobsSidecarsByRangeRequest,
   chain: IBeaconChain,
   db: {
     finalized: Pick<IBeaconDb["blockArchive"], "binaryEntriesStream" | "decodeKey">;
     unfinalized: Pick<IBeaconDb["block"], "getBinary">;
   }
-): AsyncIterable<EncodedPayloadBytes> {
+): AsyncIterable<ResponseOutgoing> {
   const {startSlot, count} = validateBeaconBlocksByRangeRequest(request);
   const endSlot = startSlot + count;
 
@@ -57,15 +43,8 @@ export async function* onBlocksOrBlobsSidecarsByRange(
     // Chain of blobs won't change
     for await (const {key, value} of db.finalized.binaryEntriesStream({gte: startSlot, lt: endSlot})) {
       yield {
-        type: EncodedPayloadType.bytes,
-        bytes: value,
-        contextBytes:
-          protocol.contextBytes.type === ContextBytesType.Empty
-            ? {type: ContextBytesType.Empty}
-            : {
-                type: ContextBytesType.ForkDigest,
-                fork: chain.config.getForkName(db.finalized.decodeKey(key)),
-              },
+        data: value,
+        fork: chain.config.getForkName(db.finalized.decodeKey(key)),
       };
     }
   }
@@ -95,15 +74,8 @@ export async function* onBlocksOrBlobsSidecarsByRange(
         }
 
         yield {
-          type: EncodedPayloadType.bytes,
-          bytes: blockBytes,
-          contextBytes:
-            protocol.contextBytes.type === ContextBytesType.Empty
-              ? {type: ContextBytesType.Empty}
-              : {
-                  type: ContextBytesType.ForkDigest,
-                  fork: chain.config.getForkName(block.slot),
-                },
+          data: blockBytes,
+          fork: chain.config.getForkName(block.slot),
         };
       }
 

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

@@ -1,16 +1,15 @@
 import {toHexString} from "@chainsafe/ssz";
-import {ContextBytesType, EncodedPayloadBytes, EncodedPayloadType, ProtocolDescriptor} from "@lodestar/reqresp";
-import {Slot, allForks, phase0} from "@lodestar/types";
+import {ResponseOutgoing} from "@lodestar/reqresp";
+import {Slot, phase0} from "@lodestar/types";
 import {IBeaconChain} from "../../../chain/index.js";
 import {IBeaconDb} from "../../../db/index.js";
 import {getSlotFromSignedBeaconBlockSerialized} from "../../../util/sszBytes.js";
 
 export async function* onBeaconBlocksByRoot(
-  protocol: ProtocolDescriptor<phase0.BeaconBlocksByRootRequest, allForks.SignedBeaconBlock>,
   requestBody: phase0.BeaconBlocksByRootRequest,
   chain: IBeaconChain,
   db: IBeaconDb
-): AsyncIterable<EncodedPayloadBytes> {
+): AsyncIterable<ResponseOutgoing> {
   for (const blockRoot of requestBody) {
     const root = blockRoot;
     const summary = chain.forkChoice.getBlock(root);
@@ -40,15 +39,8 @@ export async function* onBeaconBlocksByRoot(
       }
 
       yield {
-        type: EncodedPayloadType.bytes,
-        bytes: blockBytes,
-        contextBytes:
-          protocol.contextBytes.type === ContextBytesType.ForkDigest
-            ? {
-                type: ContextBytesType.ForkDigest,
-                fork: chain.config.getForkName(slot),
-              }
-            : {type: ContextBytesType.Empty},
+        data: blockBytes,
+        fork: chain.config.getForkName(slot),
       };
     }
   }

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

@@ -1,18 +1,17 @@
 import {deneb} from "@lodestar/types";
-import {EncodedPayloadBytes, ProtocolDescriptor} from "@lodestar/reqresp";
+import {ResponseOutgoing} from "@lodestar/reqresp";
 import {IBeaconChain} from "../../../chain/index.js";
 import {IBeaconDb} from "../../../db/index.js";
 import {onBlocksOrBlobsSidecarsByRange} from "./beaconBlocksByRange.js";
 
 // TODO DENEB: Unit test
 
 export function onBlobsSidecarsByRange(
-  protocol: ProtocolDescriptor<deneb.BlobsSidecarsByRangeRequest, deneb.BlobsSidecar>,
   request: deneb.BlobsSidecarsByRangeRequest,
   chain: IBeaconChain,
   db: IBeaconDb
-): AsyncIterable<EncodedPayloadBytes> {
-  return onBlocksOrBlobsSidecarsByRange(protocol, request, chain, {
+): AsyncIterable<ResponseOutgoing> {
+  return onBlocksOrBlobsSidecarsByRange(request, chain, {
     finalized: db.blobsSidecarArchive,
     unfinalized: db.blobsSidecar,
   });

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

@@ -1,5 +1,6 @@
-import {HandlerTypeFromMessage} from "@lodestar/reqresp";
-import * as protocols from "@lodestar/reqresp/protocols";
+import {PeerId} from "@libp2p/interface-peer-id";
+import {phase0, ssz} from "@lodestar/types";
+import {ProtocolHandler, ResponseOutgoing} from "@lodestar/reqresp";
 import {IBeaconChain} from "../../../chain/index.js";
 import {IBeaconDb} from "../../../db/index.js";
 import {onBeaconBlocksByRange} from "./beaconBlocksByRange.js";
@@ -13,48 +14,54 @@ import {onLightClientUpdatesByRange} from "./lightClientUpdatesByRange.js";
 import {onStatus} from "./status.js";
 
 export interface ReqRespHandlers {
-  onStatus: HandlerTypeFromMessage<typeof protocols.Status>;
-  onBeaconBlocksByRange: HandlerTypeFromMessage<typeof protocols.BeaconBlocksByRange>;
-  onBeaconBlocksByRoot: HandlerTypeFromMessage<typeof protocols.BeaconBlocksByRoot>;
-  onBeaconBlockAndBlobsSidecarByRoot: HandlerTypeFromMessage<typeof protocols.BeaconBlockAndBlobsSidecarByRoot>;
-  onBlobsSidecarsByRange: HandlerTypeFromMessage<typeof protocols.BlobsSidecarsByRange>;
-  onLightClientBootstrap: HandlerTypeFromMessage<typeof protocols.LightClientBootstrap>;
-  onLightClientUpdatesByRange: HandlerTypeFromMessage<typeof protocols.LightClientUpdatesByRange>;
-  onLightClientFinalityUpdate: HandlerTypeFromMessage<typeof protocols.LightClientFinalityUpdate>;
-  onLightClientOptimisticUpdate: HandlerTypeFromMessage<typeof protocols.LightClientOptimisticUpdate>;
+  onStatus: (req: phase0.Status, peerId: PeerId) => AsyncIterable<ResponseOutgoing>;
+  onBeaconBlocksByRange: ProtocolHandler;
+  onBeaconBlocksByRoot: ProtocolHandler;
+  onBeaconBlockAndBlobsSidecarByRoot: ProtocolHandler;
+  onBlobsSidecarsByRange: ProtocolHandler;
+  onLightClientBootstrap: ProtocolHandler;
+  onLightClientUpdatesByRange: ProtocolHandler;
+  onLightClientFinalityUpdate: ProtocolHandler;
+  onLightClientOptimisticUpdate: ProtocolHandler;
 }
 /**
  * The ReqRespHandler module handles app-level requests / responses from other peers,
  * fetching state from the chain and database as needed.
  */
 export function getReqRespHandlers({db, chain}: {db: IBeaconDb; chain: IBeaconChain}): ReqRespHandlers {
   return {
-    async *onStatus(protocol) {
-      yield* onStatus(protocol, chain);
+    async *onStatus() {
+      yield* onStatus(chain);
     },
-    async *onBeaconBlocksByRange(protocol, req) {
-      yield* onBeaconBlocksByRange(protocol, req, chain, db);
+    async *onBeaconBlocksByRange(req) {
+      const body = ssz.phase0.BeaconBlocksByRangeRequest.deserialize(req.data);
+      yield* onBeaconBlocksByRange(body, chain, db);
     },
-    async *onBeaconBlocksByRoot(protocol, req) {
-      yield* onBeaconBlocksByRoot(protocol, req, chain, db);
+    async *onBeaconBlocksByRoot(req) {
+      const body = ssz.phase0.BeaconBlocksByRootRequest.deserialize(req.data);
+      yield* onBeaconBlocksByRoot(body, chain, db);
     },
-    async *onBeaconBlockAndBlobsSidecarByRoot(protocol, req) {
-      yield* onBeaconBlockAndBlobsSidecarByRoot(protocol, req, chain, db);
+    async *onBeaconBlockAndBlobsSidecarByRoot(req) {
+      const body = ssz.deneb.BeaconBlockAndBlobsSidecarByRootRequest.deserialize(req.data);
+      yield* onBeaconBlockAndBlobsSidecarByRoot(body, chain, db);
     },
-    async *onBlobsSidecarsByRange(protocol, req) {
-      yield* onBlobsSidecarsByRange(protocol, req, chain, db);
+    async *onBlobsSidecarsByRange(req) {
+      const body = ssz.deneb.BlobsSidecarsByRangeRequest.deserialize(req.data);
+      yield* onBlobsSidecarsByRange(body, chain, db);
     },
-    async *onLightClientBootstrap(protocol, req) {
-      yield* onLightClientBootstrap(protocol, req, chain);
+    async *onLightClientBootstrap(req) {
+      const body = ssz.Root.deserialize(req.data);
+      yield* onLightClientBootstrap(body, chain);
     },
-    async *onLightClientUpdatesByRange(protocol, req) {
-      yield* onLightClientUpdatesByRange(protocol, req, chain);
+    async *onLightClientUpdatesByRange(req) {
+      const body = ssz.altair.LightClientUpdatesByRange.deserialize(req.data);
+      yield* onLightClientUpdatesByRange(body, chain);
     },
-    async *onLightClientFinalityUpdate(protocol) {
-      yield* onLightClientFinalityUpdate(protocol, chain);
+    async *onLightClientFinalityUpdate() {
+      yield* onLightClientFinalityUpdate(chain);
     },
-    async *onLightClientOptimisticUpdate(protocol) {
-      yield* onLightClientOptimisticUpdate(protocol, chain);
+    async *onLightClientOptimisticUpdate() {
+      yield* onLightClientOptimisticUpdate(chain);
     },
   };
 }

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

@@ -1,31 +1,22 @@
 import {
-  EncodedPayloadType,
   RespStatus,
   ResponseError,
   LightClientServerError,
   LightClientServerErrorCode,
-  EncodedPayloadBytes,
-  ContextBytesType,
-  ProtocolDescriptor,
+  ResponseOutgoing,
 } from "@lodestar/reqresp";
-import {Root, allForks} from "@lodestar/types";
+import {Root} from "@lodestar/types";
 import {IBeaconChain} from "../../../chain/index.js";
+import {ReqRespMethod, responseSszTypeByMethod} from "../types.js";
 
-export async function* onLightClientBootstrap(
-  protocol: ProtocolDescriptor<Root, allForks.LightClientBootstrap>,
-  requestBody: Root,
-  chain: IBeaconChain
-): AsyncIterable<EncodedPayloadBytes> {
+export async function* onLightClientBootstrap(requestBody: Root, chain: IBeaconChain): AsyncIterable<ResponseOutgoing> {
   try {
+    const bootstrap = await chain.lightClientServer.getBootstrap(requestBody);
+    const fork = chain.config.getForkName(bootstrap.header.beacon.slot);
+    const type = responseSszTypeByMethod[ReqRespMethod.LightClientBootstrap](fork, 0);
     yield {
-      type: EncodedPayloadType.bytes,
-      bytes: protocol
-        .responseType(chain.config.getForkName(chain.clock.currentSlot))
-        .serialize(await chain.lightClientServer.getBootstrap(requestBody)),
-      contextBytes: {
-        type: ContextBytesType.ForkDigest,
-        fork: chain.config.getForkName(chain.clock.currentSlot),
-      },
+      data: type.serialize(bootstrap),
+      fork,
     };
   } catch (e) {
     if ((e as LightClientServerError).type?.code === LightClientServerErrorCode.RESOURCE_UNAVAILABLE) {

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

@@ -1,29 +1,17 @@
-import {
-  ContextBytesType,
-  EncodedPayloadBytes,
-  EncodedPayloadType,
-  ProtocolDescriptor,
-  RespStatus,
-  ResponseError,
-} from "@lodestar/reqresp";
-import {allForks} from "@lodestar/types";
+import {ResponseOutgoing, RespStatus, ResponseError} from "@lodestar/reqresp";
 import {IBeaconChain} from "../../../chain/index.js";
+import {ReqRespMethod, responseSszTypeByMethod} from "../types.js";
 
-export async function* onLightClientFinalityUpdate(
-  protocol: ProtocolDescriptor<null, allForks.LightClientFinalityUpdate>,
-  chain: IBeaconChain
-): AsyncIterable<EncodedPayloadBytes> {
-  const finalityUpdate = chain.lightClientServer.getFinalityUpdate();
-  if (finalityUpdate === null) {
+export async function* onLightClientFinalityUpdate(chain: IBeaconChain): AsyncIterable<ResponseOutgoing> {
+  const update = chain.lightClientServer.getFinalityUpdate();
+  if (update === null) {
     throw new ResponseError(RespStatus.RESOURCE_UNAVAILABLE, "No latest finality update available");
   } else {
+    const fork = chain.config.getForkName(update.signatureSlot);
+    const type = responseSszTypeByMethod[ReqRespMethod.LightClientFinalityUpdate](fork, 0);
     yield {
-      type: EncodedPayloadType.bytes,
-      bytes: protocol.responseType(chain.config.getForkName(chain.clock.currentSlot)).serialize(finalityUpdate),
-      contextBytes: {
-        type: ContextBytesType.ForkDigest,
-        fork: chain.config.getForkName(chain.clock.currentSlot),
-      },
+      data: type.serialize(update),
+      fork,
     };
   }
 }