Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

fix: use correct by peer rate limiting on BlobSidecarsByRoot/Range post-electra #7405

Merged
merged 4 commits into from
Jan 29, 2025
Merged
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Prev Previous commit
Next Next commit
Re-apply new rate limit when registering same protocol
  • Loading branch information
nflaig committed Jan 27, 2025
commit 47d610af550d951c3fb34764e93c8e7719d97a67
38 changes: 22 additions & 16 deletions packages/beacon-node/src/network/reqresp/ReqRespBeaconNode.ts
Original file line number Diff line number Diff line change
@@ -139,7 +139,7 @@ export class ReqRespBeaconNode extends ReqResp {

// Subscribe required protocols, prevent libp2p for throwing if already registered
for (const [protocol, handler] of mustSubscribeProtocols) {
this.registerProtocol(fork, {...protocol, handler}, {ignoreIfDuplicate: true}).catch((e) => {
this.registerProtocol({...protocol, handler}, {ignoreIfDuplicate: true}).catch((e) => {
this.logger.error("Error on ReqResp.registerProtocol", {protocolID: this.formatProtocolID(protocol)}, e);
});
}
@@ -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)]
);
}

5 changes: 3 additions & 2 deletions packages/beacon-node/src/network/reqresp/protocols.ts
Original file line number Diff line number Diff line change
@@ -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),
});
11 changes: 7 additions & 4 deletions packages/beacon-node/src/network/reqresp/rateLimit.ts
Original file line number Diff line number Diff line change
@@ -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,12 +37,12 @@ export const rateLimitQuotas: (config: BeaconConfig) => Record<ReqRespMethod, In
},
[ReqRespMethod.BlobSidecarsByRange]: {
// Rationale: MAX_REQUEST_BLOCKS_DENEB * MAX_BLOBS_PER_BLOCK
byPeer: (fork) => ({quota: config.getMaxRequestBlobSidecars(fork), quotaTimeMs: 10_000}),
byPeer: {quota: config.getMaxRequestBlobSidecars(fork), quotaTimeMs: 10_000},
nazarhussain marked this conversation as resolved.
Show resolved Hide resolved
getRequestCount: getRequestCountFn(config, ReqRespMethod.BlobSidecarsByRange, (req) => req.count),
},
[ReqRespMethod.BlobSidecarsByRoot]: {
// Rationale: quota of BeaconBlocksByRoot * MAX_BLOBS_PER_BLOCK
byPeer: (fork) => ({quota: config.getMaxRequestBlobSidecars(fork), quotaTimeMs: 10_000}),
byPeer: {quota: config.getMaxRequestBlobSidecars(fork), quotaTimeMs: 10_000},
getRequestCount: getRequestCountFn(config, ReqRespMethod.BlobSidecarsByRoot, (req) => req.length),
},
[ReqRespMethod.LightClientBootstrap]: {
13 changes: 7 additions & 6 deletions packages/reqresp/src/ReqResp.ts
Original file line number Diff line number Diff line change
@@ -91,22 +91,23 @@ export class ReqResp {
* Throws if the same protocol is registered twice.
* Can be called at any time, no concept of started / stopped
*/
async registerProtocol(fork: ForkName, protocol: Protocol, opts?: ReqRespRegisterOpts): Promise<void> {
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);
}
nazarhussain marked this conversation as resolved.
Show resolved Hide resolved

// 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(fork, protocolID, inboundRateLimits);
}

return this.libp2p.handle(protocolID, this.getRequestHandler(protocol, protocolID));
}

14 changes: 5 additions & 9 deletions packages/reqresp/src/rate_limiter/ReqRespRateLimiter.ts
Original file line number Diff line number Diff line change
@@ -1,5 +1,4 @@
import {PeerId} from "@libp2p/interface";
import {ForkName} from "@lodestar/params";
import {InboundRateLimitQuota, ReqRespRateLimiterOpts} from "../types.js";
import {RateLimiterGRCA} from "./rateLimiterGRCA.js";

@@ -29,30 +28,27 @@ export class ReqRespRateLimiter {
return this.rateLimitMultiplier > 0;
}

initRateLimits(fork: ForkName, protocolID: ProtocolID, rateLimits: InboundRateLimitQuota): void {
setRateLimits(protocolID: ProtocolID, rateLimits: InboundRateLimitQuota): void {
if (!this.enabled) {
return;
}

if (rateLimits.byPeer) {
const {quota, quotaTimeMs} =
typeof rateLimits.byPeer === "function" ? rateLimits.byPeer(fork) : rateLimits.byPeer;
this.rateLimitersPerPeer.set(
protocolID,
RateLimiterGRCA.fromQuota<string>({
quotaTimeMs: quotaTimeMs,
quota: quota * this.rateLimitMultiplier,
quotaTimeMs: rateLimits.byPeer.quotaTimeMs,
quota: rateLimits.byPeer.quota * this.rateLimitMultiplier,
})
);
}

if (rateLimits.total) {
const {quota, quotaTimeMs} = typeof rateLimits.total === "function" ? rateLimits.total(fork) : rateLimits.total;
this.rateLimitersTotal.set(
protocolID,
RateLimiterGRCA.fromQuota<null>({
quotaTimeMs: quotaTimeMs,
quota: quota * this.rateLimitMultiplier,
quotaTimeMs: rateLimits.total.quotaTimeMs,
quota: rateLimits.total.quota * this.rateLimitMultiplier,
})
);
}
4 changes: 2 additions & 2 deletions packages/reqresp/src/types.ts
Original file line number Diff line number Diff line change
@@ -45,11 +45,11 @@ export interface InboundRateLimitQuota {
/**
* Will be tracked for the protocol per peer
*/
byPeer?: ((fork: ForkName) => RateLimiterQuota) | RateLimiterQuota;
byPeer?: RateLimiterQuota;
/**
* Will be tracked regardless of the peer
*/
total?: ((fork: ForkName) => RateLimiterQuota) | RateLimiterQuota;
total?: RateLimiterQuota;
/**
* Some requests may be counted multiple e.g. getBlocksByRange
* for such implement this method else `1` will be used default
36 changes: 31 additions & 5 deletions packages/reqresp/test/unit/ReqResp.test.ts
Original file line number Diff line number Diff line change
@@ -1,10 +1,11 @@
import {getEmptyLogger} from "@lodestar/logger/empty";
import {ForkName} from "@lodestar/params";
import {Logger} from "@lodestar/utils";
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";
@@ -14,7 +15,6 @@ describe("ResResp", () => {
let reqresp: ReqResp;
let libp2p: Libp2p;
let logger: Logger;
const fork = ForkName.deneb;
const ping = pingProtocol(getEmptyHandler());

beforeEach(() => {
@@ -60,18 +60,44 @@ describe("ResResp", () => {

describe("duplex protocol", () => {
it("should register protocol and dial", async () => {
await reqresp.registerProtocol(fork, numberToStringProtocol);
await reqresp.registerProtocol(numberToStringProtocol);

expect(reqresp.getRegisteredProtocols()).toEqual(["/eth2/beacon_chain/req/number_to_string/1/ssz_snappy"]);
expect(libp2p.handle).toHaveBeenCalledOnce();
});

it("should not register handler twice for same protocol if ignoreIfDuplicate=true", async () => {
await reqresp.registerProtocol(fork, numberToStringProtocol, {ignoreIfDuplicate: true});
await reqresp.registerProtocol(numberToStringProtocol, {ignoreIfDuplicate: true});
expect(libp2p.handle).toHaveBeenCalledOnce();

await reqresp.registerProtocol(fork, numberToStringProtocol, {ignoreIfDuplicate: true});
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);
});
});
});
Loading