Modular Reqresp Package (#4775)

* ReqResp modular for light-client

* Add package skeleton for @lodestar/reqresp

* Fix all dependencies and file reference issues in reqresp

* Extract protocol messages from implementation

* Fix reqresp usage

* Update the reqresp module handler structure

* Update reqresp handlers for beacon-node

* Review abstractions

* FIx types

* Fix metrics type

* Update status handler to use same pattern

* Fix the import syntax for default imports

* Remove the existing unit tests for reqresp temporarily

* Fix some unit tests

* Fix few performance tests

* Fix tsonfig for reqresp package

* Fix e2e tests

* Fix a type error

* Fix few e2e tests

* Fix e2e timeout tests

* Fix readme

* Rename class file to its constructor name

* Fix some linter errors

Co-authored-by: dapplion <[email protected]>

packages/beacon-node/package.json

@@ -86,7 +86,6 @@
     "test:unit": "yarn test:unit:minimal && yarn test:unit:mainnet",
     "test:e2e": "mocha 'test/e2e/**/*.test.ts'",
     "test:sim": "mocha 'test/sim/**/*.test.ts'",
-    "test:sim:multiThread": "mocha 'test/sim/multiNodeMultiThread.test.ts'",
     "test:sim:merge-interop": "mocha 'test/sim/merge-interop.test.ts'",
     "test:sim:mergemock": "mocha 'test/sim/mergemock.test.ts'",
     "download-spec-tests": "node --loader=ts-node/esm test/spec/downloadTests.ts",
@@ -127,6 +126,7 @@
     "@lodestar/types": "^1.2.1",
     "@lodestar/utils": "^1.2.1",
     "@lodestar/validator": "^1.2.1",
+    "@lodestar/reqresp": "^0.1.0",
     "@multiformats/multiaddr": "^11.0.0",
     "@types/datastore-level": "^3.0.0",
     "buffer-xor": "^2.0.2",

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

@@ -7,7 +7,6 @@ export const ZERO_HASH = Buffer.alloc(32, 0);
 export const ZERO_HASH_HEX = "0x" + "00".repeat(32);
 export const EMPTY_SIGNATURE = Buffer.alloc(96, 0);
 export const GRAFFITI_SIZE = 32;
-export const MAX_VARINT_BYTES = 10;
 
 /**
  * The maximum milliseconds of clock disparity assumed between honest nodes.

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

@@ -12,50 +12,12 @@
  */
 export const ATTESTATION_PROPAGATION_SLOT_RANGE = 32;
 
-//  Request/Response constants
-
-export enum RespStatus {
-  /**
-   * A normal response follows, with contents matching the expected message schema and encoding specified in the request
-   */
-  SUCCESS = 0,
-  /**
-   * The contents of the request are semantically invalid, or the payload is malformed,
-   * or could not be understood. The response payload adheres to the ErrorMessage schema
-   */
-  INVALID_REQUEST = 1,
-  /**
-   * The responder encountered an error while processing the request. The response payload adheres to the ErrorMessage schema
-   */
-  SERVER_ERROR = 2,
-  /**
-   * The responder does not have requested resource.  The response payload adheres to the ErrorMessage schema (described below). Note: This response code is only valid as a response to BlocksByRange
-   */
-  RESOURCE_UNAVAILABLE = 3,
-  /**
-   * Our node does not have bandwidth to serve requests due to either per-peer quota or total quota.
-   */
-  RATE_LIMITED = 139,
-}
-
-export type RpcResponseStatusError = Exclude<RespStatus, RespStatus.SUCCESS>;
-
 /** The maximum allowed size of uncompressed gossip messages. */
 export const GOSSIP_MAX_SIZE = 2 ** 20;
 export const GOSSIP_MAX_SIZE_BELLATRIX = 10 * GOSSIP_MAX_SIZE;
 /** The maximum allowed size of uncompressed req/resp chunked responses. */
 export const MAX_CHUNK_SIZE = 2 ** 20;
 export const MAX_CHUNK_SIZE_BELLATRIX = 10 * MAX_CHUNK_SIZE;
-/** The maximum time to wait for first byte of request response (time-to-first-byte). */
-export const TTFB_TIMEOUT = 5 * 1000; // 5 sec
-/** The maximum time for complete response transfer. */
-export const RESP_TIMEOUT = 10 * 1000; // 10 sec
-/** Non-spec timeout from sending request until write stream closed by responder */
-export const REQUEST_TIMEOUT = 5 * 1000; // 5 sec
-/** Non-spec timeout from dialing protocol until stream opened */
-export const DIAL_TIMEOUT = 5 * 1000; // 5 sec
-// eslint-disable-next-line @typescript-eslint/naming-convention
-export const timeoutOptions = {TTFB_TIMEOUT, RESP_TIMEOUT, REQUEST_TIMEOUT, DIAL_TIMEOUT};
 
 export enum GoodByeReasonCode {
   CLIENT_SHUTDOWN = 1,

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

@@ -96,42 +96,6 @@ export function createBeaconMetrics(register: RegistryMetricCreator) {
     }),
 
     reqResp: {
-      outgoingRequests: register.gauge<"method">({
-        name: "beacon_reqresp_outgoing_requests_total",
-        help: "Counts total requests done per method",
-        labelNames: ["method"],
-      }),
-      outgoingRequestRoundtripTime: register.histogram<"method">({
-        name: "beacon_reqresp_outgoing_request_roundtrip_time_seconds",
-        help: "Histogram of outgoing requests round-trip time",
-        labelNames: ["method"],
-        buckets: [0.1, 0.2, 0.5, 1, 5, 15, 60],
-      }),
-      outgoingErrors: register.gauge<"method">({
-        name: "beacon_reqresp_outgoing_requests_error_total",
-        help: "Counts total failed requests done per method",
-        labelNames: ["method"],
-      }),
-      incomingRequests: register.gauge<"method">({
-        name: "beacon_reqresp_incoming_requests_total",
-        help: "Counts total responses handled per method",
-        labelNames: ["method"],
-      }),
-      incomingRequestHandlerTime: register.histogram<"method">({
-        name: "beacon_reqresp_incoming_request_handler_time_seconds",
-        help: "Histogram of incoming requests internal handling time",
-        labelNames: ["method"],
-        buckets: [0.1, 0.2, 0.5, 1, 5],
-      }),
-      incomingErrors: register.gauge<"method">({
-        name: "beacon_reqresp_incoming_requests_error_total",
-        help: "Counts total failed responses handled per method",
-        labelNames: ["method"],
-      }),
-      dialErrors: register.gauge({
-        name: "beacon_reqresp_dial_errors_total",
-        help: "Count total dial errors",
-      }),
       rateLimitErrors: register.gauge<"tracker">({
         name: "beacon_reqresp_rate_limiter_errors_total",
         help: "Count rate limiter errors",

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

@@ -2,7 +2,7 @@ import {EventEmitter} from "events";
 import {PeerId} from "@libp2p/interface-peer-id";
 import StrictEventEmitter from "strict-event-emitter-types";
 import {allForks, phase0} from "@lodestar/types";
-import {RequestTypedContainer} from "./reqresp/index.js";
+import {RequestTypedContainer} from "./reqresp/ReqRespBeaconNode.js";
 
 export enum NetworkEvent {
   /** A relevant peer has connected or has been re-STATUS'd */

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

@@ -3,6 +3,6 @@ export * from "./interface.js";
 export * from "./network.js";
 export * from "./nodejs/index.js";
 export * from "./gossip/index.js";
-export * from "./reqresp/index.js";
+export * from "./reqresp/ReqRespBeaconNode.js";
 export * from "./util.js";
 export * from "./peers/index.js";

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

@@ -6,7 +6,7 @@ import {INetworkEventBus} from "./events.js";
 import {Eth2Gossipsub} from "./gossip/index.js";
 import {MetadataController} from "./metadata.js";
 import {PeerAction} from "./peers/index.js";
-import {IReqResp} from "./reqresp/index.js";
+import {IReqRespBeaconNode} from "./reqresp/ReqRespBeaconNode.js";
 import {IAttnetsService, ISubnetsService, CommitteeSubscription} from "./subnets/index.js";
 
 export type PeerSearchOptions = {
@@ -16,7 +16,7 @@ export type PeerSearchOptions = {
 
 export interface INetwork {
   events: INetworkEventBus;
-  reqResp: IReqResp;
+  reqResp: IReqRespBeaconNode;
   attnetsService: IAttnetsService;
   syncnetsService: ISubnetsService;
   gossip: Eth2Gossipsub;

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

@@ -13,7 +13,7 @@ import {IMetrics} from "../metrics/index.js";
 import {ChainEvent, IBeaconChain, IBeaconClock} from "../chain/index.js";
 import {INetworkOptions} from "./options.js";
 import {INetwork} from "./interface.js";
-import {IReqResp, IReqRespOptions, ReqResp, ReqRespHandlers} from "./reqresp/index.js";
+import {IReqRespBeaconNode, ReqRespBeaconNode, ReqRespHandlers} from "./reqresp/ReqRespBeaconNode.js";
 import {Eth2Gossipsub, getGossipHandlers, GossipHandlers, GossipType} from "./gossip/index.js";
 import {MetadataController} from "./metadata.js";
 import {FORK_EPOCH_LOOKAHEAD, getActiveForks} from "./forks.js";
@@ -38,7 +38,7 @@ interface INetworkModules {
 
 export class Network implements INetwork {
   events: INetworkEventBus;
-  reqResp: IReqResp;
+  reqResp: IReqRespBeaconNode;
   attnetsService: AttnetsService;
   syncnetsService: SyncnetsService;
   gossip: Eth2Gossipsub;
@@ -56,7 +56,7 @@ export class Network implements INetwork {
 
   private subscribedForks = new Set<ForkName>();
 
-  constructor(private readonly opts: INetworkOptions & IReqRespOptions, modules: INetworkModules) {
+  constructor(private readonly opts: INetworkOptions, modules: INetworkModules) {
     const {config, libp2p, logger, metrics, chain, reqRespHandlers, gossipHandlers, signal} = modules;
     this.libp2p = libp2p;
     this.logger = logger;
@@ -71,7 +71,7 @@ export class Network implements INetwork {
     this.events = networkEventBus;
     this.metadata = metadata;
     this.peerRpcScores = peerRpcScores;
-    this.reqResp = new ReqResp(
+    this.reqResp = new ReqRespBeaconNode(
       {
         config,
         libp2p,

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

@@ -2,9 +2,9 @@ import {ENR, IDiscv5DiscoveryInputOptions} from "@chainsafe/discv5";
 import {Eth2GossipsubOpts} from "./gossip/gossipsub.js";
 import {defaultGossipHandlerOpts, GossipHandlerOpts} from "./gossip/handlers/index.js";
 import {PeerManagerOpts} from "./peers/index.js";
-import {defaultRateLimiterOpts, RateLimiterOpts} from "./reqresp/response/rateLimiter.js";
+import {ReqRespBeaconNodeOpts} from "./reqresp/ReqRespBeaconNode.js";
 
-export interface INetworkOptions extends PeerManagerOpts, RateLimiterOpts, GossipHandlerOpts, Eth2GossipsubOpts {
+export interface INetworkOptions extends PeerManagerOpts, ReqRespBeaconNodeOpts, GossipHandlerOpts, Eth2GossipsubOpts {
   localMultiaddrs: string[];
   bootMultiaddrs?: string[];
   subscribeAllSubnets?: boolean;
@@ -27,6 +27,5 @@ export const defaultNetworkOptions: INetworkOptions = {
   localMultiaddrs: ["/ip4/0.0.0.0/tcp/9000"],
   bootMultiaddrs: [],
   discv5: defaultDiscv5Options,
-  ...defaultRateLimiterOpts,
   ...defaultGossipHandlerOpts,
 };

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

@@ -11,7 +11,7 @@ import {IBeaconChain} from "../../chain/index.js";
 import {GoodByeReasonCode, GOODBYE_KNOWN_CODES, Libp2pEvent} from "../../constants/index.js";
 import {IMetrics} from "../../metrics/index.js";
 import {NetworkEvent, INetworkEventBus} from "../events.js";
-import {IReqResp, ReqRespMethod, RequestTypedContainer} from "../reqresp/index.js";
+import {IReqRespBeaconNode, ReqRespMethod, RequestTypedContainer} from "../reqresp/ReqRespBeaconNode.js";
 import {getConnection, getConnectionsMap, prettyPrintPeerId} from "../util.js";
 import {ISubnetsService} from "../subnets/index.js";
 import {SubnetType} from "../metadata.js";
@@ -76,7 +76,7 @@ export type PeerManagerModules = {
   libp2p: Libp2p;
   logger: ILogger;
   metrics: IMetrics | null;
-  reqResp: IReqResp;
+  reqResp: IReqRespBeaconNode;
   gossip: Eth2Gossipsub;
   attnetsService: ISubnetsService;
   syncnetsService: ISubnetsService;
@@ -107,7 +107,7 @@ export class PeerManager {
   private libp2p: Libp2p;
   private logger: ILogger;
   private metrics: IMetrics | null;
-  private reqResp: IReqResp;
+  private reqResp: IReqRespBeaconNode;
   private gossipsub: Eth2Gossipsub;
   private attnetsService: ISubnetsService;
   private syncnetsService: ISubnetsService;

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

@@ -1,6 +1,6 @@
 import {PeerId} from "@libp2p/interface-peer-id";
 import {altair} from "@lodestar/types";
-import {Encoding} from "../reqresp/types.js";
+import {Encoding} from "@lodestar/reqresp";
 import {ClientKind} from "./client.js";
 
 type PeerIdStr = string;