Merge ed9aa8b into ee9d4fa

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

@@ -219,42 +219,48 @@ export class ReqRespBeaconNode extends ReqResp {
    */
   private getProtocolsAtFork(fork: ForkName): [ProtocolNoHandler, ProtocolHandler][] {
     const protocolsAtFork: [ProtocolNoHandler, ProtocolHandler][] = [
-      [protocols.Ping(this.config), this.onPing.bind(this)],
-      [protocols.Status(this.config), this.onStatus.bind(this)],
-      [protocols.Goodbye(this.config), this.onGoodbye.bind(this)],
+      [protocols.Ping(fork, this.config), this.onPing.bind(this)],
+      [protocols.Status(fork, this.config), this.onStatus.bind(this)],
+      [protocols.Goodbye(fork, this.config), this.onGoodbye.bind(this)],
       // Support V2 methods as soon as implemented (for altair)
       // Ref https://github.com/ethereum/consensus-specs/blob/v1.2.0/specs/altair/p2p-interface.md#transitioning-from-v1-to-v2
-      [protocols.MetadataV2(this.config), this.onMetadata.bind(this)],
-      [protocols.BeaconBlocksByRangeV2(this.config), this.getHandler(ReqRespMethod.BeaconBlocksByRange)],
-      [protocols.BeaconBlocksByRootV2(this.config), this.getHandler(ReqRespMethod.BeaconBlocksByRoot)],
+      [protocols.MetadataV2(fork, this.config), this.onMetadata.bind(this)],
+      [protocols.BeaconBlocksByRangeV2(fork, this.config), this.getHandler(ReqRespMethod.BeaconBlocksByRange)],
+      [protocols.BeaconBlocksByRootV2(fork, this.config), this.getHandler(ReqRespMethod.BeaconBlocksByRoot)],
     ];
 
     if (ForkSeq[fork] < ForkSeq.altair) {
       // Unregister V1 topics at the fork boundary, so only declare for pre-altair
       protocolsAtFork.push(
-        [protocols.Metadata(this.config), this.onMetadata.bind(this)],
-        [protocols.BeaconBlocksByRange(this.config), this.getHandler(ReqRespMethod.BeaconBlocksByRange)],
-        [protocols.BeaconBlocksByRoot(this.config), this.getHandler(ReqRespMethod.BeaconBlocksByRoot)]
+        [protocols.Metadata(fork, this.config), this.onMetadata.bind(this)],
+        [protocols.BeaconBlocksByRange(fork, this.config), this.getHandler(ReqRespMethod.BeaconBlocksByRange)],
+        [protocols.BeaconBlocksByRoot(fork, this.config), this.getHandler(ReqRespMethod.BeaconBlocksByRoot)]
       );
     }
 
     if (ForkSeq[fork] >= ForkSeq.altair && !this.disableLightClientServer) {
       // Should be okay to enable before altair, but for consistency only enable afterwards
       protocolsAtFork.push(
-        [protocols.LightClientBootstrap(this.config), this.getHandler(ReqRespMethod.LightClientBootstrap)],
-        [protocols.LightClientFinalityUpdate(this.config), this.getHandler(ReqRespMethod.LightClientFinalityUpdate)],
+        [protocols.LightClientBootstrap(fork, this.config), this.getHandler(ReqRespMethod.LightClientBootstrap)],
         [
-          protocols.LightClientOptimisticUpdate(this.config),
+          protocols.LightClientFinalityUpdate(fork, this.config),
+          this.getHandler(ReqRespMethod.LightClientFinalityUpdate),
+        ],
+        [
+          protocols.LightClientOptimisticUpdate(fork, this.config),
           this.getHandler(ReqRespMethod.LightClientOptimisticUpdate),
         ],
-        [protocols.LightClientUpdatesByRange(this.config), this.getHandler(ReqRespMethod.LightClientUpdatesByRange)]
+        [
+          protocols.LightClientUpdatesByRange(fork, this.config),
+          this.getHandler(ReqRespMethod.LightClientUpdatesByRange),
+        ]
       );
     }
 
     if (ForkSeq[fork] >= ForkSeq.deneb) {
       protocolsAtFork.push(
-        [protocols.BlobSidecarsByRoot(this.config), this.getHandler(ReqRespMethod.BlobSidecarsByRoot)],
-        [protocols.BlobSidecarsByRange(this.config), this.getHandler(ReqRespMethod.BlobSidecarsByRange)]
+        [protocols.BlobSidecarsByRoot(fork, this.config), this.getHandler(ReqRespMethod.BlobSidecarsByRoot)],
+        [protocols.BlobSidecarsByRange(fork, this.config), this.getHandler(ReqRespMethod.BlobSidecarsByRange)]
       );
     }
 

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

@@ -1,4 +1,5 @@
 import {BeaconConfig, ForkDigestContext} from "@lodestar/config";
+import {ForkName} from "@lodestar/params";
 import {ContextBytesFactory, ContextBytesType, Encoding} from "@lodestar/reqresp";
 import {rateLimitQuotas} from "./rateLimit.js";
 import {ProtocolNoHandler, ReqRespMethod, Version, requestSszTypeByMethod, responseSszTypeByMethod} from "./types.js";
@@ -100,12 +101,12 @@ type ProtocolSummary = {
 };
 
 function toProtocol(protocol: ProtocolSummary) {
-  return (config: BeaconConfig): ProtocolNoHandler => ({
+  return (fork: ForkName, config: BeaconConfig): ProtocolNoHandler => ({
     method: protocol.method,
     version: protocol.version,
     encoding: Encoding.SSZ_SNAPPY,
     contextBytes: toContextBytes(protocol.contextBytesType, config),
-    inboundRateLimits: rateLimitQuotas(config)[protocol.method],
+    inboundRateLimits: rateLimitQuotas(fork, config)[protocol.method],
     requestSizes: requestSszTypeByMethod(config)[protocol.method],
     responseSizes: (fork) => responseSszTypeByMethod[protocol.method](fork, protocol.version),
   });

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

@@ -1,10 +1,13 @@
 import {BeaconConfig} from "@lodestar/config";
-import {MAX_REQUEST_BLOCKS, MAX_REQUEST_LIGHT_CLIENT_UPDATES} from "@lodestar/params";
+import {ForkName, MAX_REQUEST_BLOCKS, MAX_REQUEST_LIGHT_CLIENT_UPDATES} from "@lodestar/params";
 import {InboundRateLimitQuota} from "@lodestar/reqresp";
 import {ReqRespMethod, RequestBodyByMethod} from "./types.js";
 import {requestSszTypeByMethod} from "./types.js";
 
-export const rateLimitQuotas: (config: BeaconConfig) => Record<ReqRespMethod, InboundRateLimitQuota> = (config) => ({
+export const rateLimitQuotas: (fork: ForkName, config: BeaconConfig) => Record<ReqRespMethod, InboundRateLimitQuota> = (
+  fork,
+  config
+) => ({
   [ReqRespMethod.Status]: {
     // Rationale: https://github.com/sigp/lighthouse/blob/bf533c8e42cc73c35730e285c21df8add0195369/beacon_node/lighthouse_network/src/rpc/mod.rs#L118-L130
     byPeer: {quota: 5, quotaTimeMs: 15_000},
@@ -34,14 +37,12 @@ export const rateLimitQuotas: (config: BeaconConfig) => Record<ReqRespMethod, In
   },
   [ReqRespMethod.BlobSidecarsByRange]: {
     // Rationale: MAX_REQUEST_BLOCKS_DENEB * MAX_BLOBS_PER_BLOCK
-    // TODO Electra: Stays as `MAX_REQUEST_BLOB_SIDECARS` until we have fork-aware `byPeer` and set it to `MAX_REQUEST_BLOB_SIDECARS_ELECTRA`
-    byPeer: {quota: config.MAX_REQUEST_BLOB_SIDECARS, quotaTimeMs: 10_000},
+    byPeer: {quota: config.getMaxRequestBlobSidecars(fork), quotaTimeMs: 10_000},
     getRequestCount: getRequestCountFn(config, ReqRespMethod.BlobSidecarsByRange, (req) => req.count),
   },
   [ReqRespMethod.BlobSidecarsByRoot]: {
     // Rationale: quota of BeaconBlocksByRoot * MAX_BLOBS_PER_BLOCK
-    // TODO Electra: Stays as `MAX_REQUEST_BLOB_SIDECARS` until we have fork-aware `byPeer` and set it to `MAX_REQUEST_BLOB_SIDECARS_ELECTRA`
-    byPeer: {quota: config.MAX_REQUEST_BLOB_SIDECARS, quotaTimeMs: 10_000},
+    byPeer: {quota: config.getMaxRequestBlobSidecars(fork), quotaTimeMs: 10_000},
     getRequestCount: getRequestCountFn(config, ReqRespMethod.BlobSidecarsByRoot, (req) => req.length),
   },
   [ReqRespMethod.LightClientBootstrap]: {

packages/reqresp/src/ReqResp.ts

@@ -92,20 +92,21 @@ export class ReqResp {
    */
   async registerProtocol(protocol: Protocol, opts?: ReqRespRegisterOpts): Promise<void> {
     const protocolID = this.formatProtocolID(protocol);
+    const {handler: _handler, inboundRateLimits, ...rest} = protocol;
+
+    if (inboundRateLimits) {
+      // Rate limits can change across hard forks and must always be updated
+      this.rateLimiter.setRateLimits(protocolID, inboundRateLimits);
+    }
 
     // libp2p will throw if handler for protocol is already registered, allow to overwrite behavior
     if (opts?.ignoreIfDuplicate && this.registeredProtocols.has(protocolID)) {
       return;
     }
 
-    const {handler: _handler, inboundRateLimits, ...rest} = protocol;
     this.registerDialOnlyProtocol(rest);
     this.dialOnlyProtocols.set(protocolID, false);
 
-    if (inboundRateLimits) {
-      this.rateLimiter.initRateLimits(protocolID, inboundRateLimits);
-    }
-
     return this.libp2p.handle(protocolID, this.getRequestHandler(protocol, protocolID));
   }
 

packages/reqresp/src/rate_limiter/ReqRespRateLimiter.ts

@@ -28,7 +28,7 @@ export class ReqRespRateLimiter {
     return this.rateLimitMultiplier > 0;
   }
 
-  initRateLimits(protocolID: ProtocolID, rateLimits: InboundRateLimitQuota): void {
+  setRateLimits(protocolID: ProtocolID, rateLimits: InboundRateLimitQuota): void {
     if (!this.enabled) {
       return;
     }

packages/reqresp/test/unit/ReqResp.test.ts

@@ -4,6 +4,8 @@ import {Libp2p} from "libp2p";
 import {afterEach, beforeEach, describe, expect, it, vi} from "vitest";
 import {ReqResp} from "../../src/ReqResp.js";
 import {RespStatus} from "../../src/interface.js";
+import {RateLimiterQuota} from "../../src/rate_limiter/rateLimiterGRCA.js";
+import {Protocol} from "../../src/types.js";
 import {getEmptyHandler, sszSnappyPing} from "../fixtures/messages.js";
 import {numberToStringProtocol, numberToStringProtocolDialOnly, pingProtocol} from "../fixtures/protocols.js";
 import {MockLibP2pStream} from "../utils/index.js";
@@ -71,5 +73,31 @@ describe("ResResp", () => {
       await reqresp.registerProtocol(numberToStringProtocol, {ignoreIfDuplicate: true});
       expect(libp2p.handle).toHaveBeenCalledOnce();
     });
+
+    it("should apply new rate limits if same protocol is registered with different limits", async () => {
+      // Initial registration of protocol
+      const {quota, quotaTimeMs} = numberToStringProtocol.inboundRateLimits?.byPeer as RateLimiterQuota;
+      const initialMsPerToken = quotaTimeMs / quota;
+      await reqresp.registerProtocol(numberToStringProtocol, {ignoreIfDuplicate: true});
+      const initialLimit = reqresp["rateLimiter"]["rateLimitersPerPeer"].get(
+        "/eth2/beacon_chain/req/number_to_string/1/ssz_snappy"
+      );
+      // Sanity check expected value
+      expect(initialLimit?.["msPerToken"]).toBe(initialMsPerToken);
+
+      // Register same protocol with new by peer rate limits
+      const updatedQuota: RateLimiterQuota = {quota: 10, quotaTimeMs: 15_000};
+      const updatedProtocol: Protocol = {
+        ...numberToStringProtocol,
+        inboundRateLimits: {byPeer: updatedQuota},
+      };
+      const updatedMsPerToken = updatedQuota.quotaTimeMs / updatedQuota.quota;
+      await reqresp.registerProtocol(updatedProtocol, {ignoreIfDuplicate: true});
+      const updatedLimit = reqresp["rateLimiter"]["rateLimitersPerPeer"].get(
+        "/eth2/beacon_chain/req/number_to_string/1/ssz_snappy"
+      );
+      // New limits should be applied
+      expect(updatedLimit?.["msPerToken"]).toBe(updatedMsPerToken);
+    });
   });
 });