diff --git a/packages/p2p/source/rate-limiter.ts b/packages/p2p/source/rate-limiter.ts index 8a57d6953..16245caaa 100644 --- a/packages/p2p/source/rate-limiter.ts +++ b/packages/p2p/source/rate-limiter.ts @@ -60,10 +60,14 @@ export class RateLimiter { return false; } - public async hasExceededRateLimitNoConsume(ip: string, endpoint?: string): Promise { + // How long to wait before the next request to `ip`/`endpoint` fits the + // budget again; 0 when a request may be made right away. + public async msBeforeNext(ip: string, endpoint?: string): Promise { + let wait = 0; + const global = await this.#global.get(ip); if (global !== null && global.remainingPoints <= 0) { - return true; + wait = Math.max(wait, global.msBeforeNext); } if (endpoint && this.#endpoints.has(endpoint)) { @@ -73,11 +77,11 @@ export class RateLimiter { const endpointLimiter = await endpointLimiters.get(ip); if (endpointLimiter !== null && endpointLimiter.remainingPoints <= 0) { - return true; + wait = Math.max(wait, endpointLimiter.msBeforeNext); } } - return false; + return wait; } public getRateLimitedEndpoints(): string[] { diff --git a/packages/p2p/source/throttle.test.ts b/packages/p2p/source/throttle.test.ts new file mode 100644 index 000000000..642bd8408 --- /dev/null +++ b/packages/p2p/source/throttle.test.ts @@ -0,0 +1,133 @@ +import { Identifiers } from "@mainsail/constants"; +import { Application } from "@mainsail/kernel"; +import { describe } from "@mainsail/test-runner"; + +import { Throttle } from "./throttle"; + +const makePeer = (ip: string): any => ({ ip }); + +// A serial queue like the kernel memory queue: jobs run one after another. +const makeQueue = () => { + let chain = Promise.resolve(); + + const queue = { + isStarted: () => queue.started, + push: (job: { handle: () => Promise }) => { + queue.pushed++; + chain = chain.then(() => job.handle()); + return chain; + }, + pushed: 0, + start: async () => { + queue.started = true; + }, + started: false, + stop: async () => { + queue.started = false; + }, + }; + + return queue; +}; + +describe<{ + throttle: Throttle; + queue: ReturnType; + clk: any; +}>("Throttle", ({ it, beforeEach, afterEach, assert, clock }) => { + beforeEach(async (context) => { + // The rate limiter derives refill times from Date.now() and the backoff + // uses setTimeout; fake both so waits are deterministic. + context.clk = clock({ now: 1_000_000, toFake: ["Date", "setTimeout"] }); + + context.queue = makeQueue(); + + const app = new Application(); + app.bind(Identifiers.ServiceProvider.Configuration) + .toConstantValue({ getRequired: () => 100 }) + .whenTagged("plugin", "p2p"); + app.bind(Identifiers.Cryptography.Configuration).toConstantValue({ getMaxRoundValidators: () => 5 }); + app.bind(Identifiers.Services.Queue.Factory).toConstantValue(async () => context.queue); + app.bind(Identifiers.Services.Log.Service).toConstantValue({ debug: () => {} }); + + context.throttle = await app.resolve(Throttle).initialize(); + }); + + afterEach(({ clk }) => { + clk.restore(); + }); + + it("allows sends within the budget immediately", async ({ throttle, queue }) => { + await throttle.throttle(makePeer("1.1.1.1"), "getStatus"); + + assert.equal(queue.pushed, 1); + }); + + it("defers a send that exceeds the endpoint budget until it actually refills", async ({ throttle, queue, clk }) => { + const peer = makePeer("1.1.1.1"); + + // getStatus allows 2 per second. + await throttle.throttle(peer, "getStatus"); + await throttle.throttle(peer, "getStatus"); + + let sent = false; + void throttle.throttle(peer, "getStatus").then(() => { + sent = true; + }); + + // The retry waits for the actual refill instead of polling: no + // intermediate requeue happens before the budget is available again. + await clk.tickAsync(900); + assert.false(sent); + assert.equal(queue.pushed, 3); + + await clk.tickAsync(101); + assert.true(sent); + assert.equal(queue.pushed, 4); + }); + + it("never head-of-line blocks other peers or endpoints behind a throttled send", async ({ throttle, clk }) => { + const peerA = makePeer("1.1.1.1"); + const peerB = makePeer("2.2.2.2"); + + // Exhaust peerA's getStatus budget and queue a throttled send. + await throttle.throttle(peerA, "getStatus"); + await throttle.throttle(peerA, "getStatus"); + + let throttledSent = false; + void throttle.throttle(peerA, "getStatus").then(() => { + throttledSent = true; + }); + await clk.tickAsync(0); + + // Consensus-critical sends to the same peer and any send to other peers + // must pass right through while the throttled one backs off. + await throttle.throttle(peerA, "postMessage"); + await throttle.throttle(peerB, "getStatus"); + + assert.false(throttledSent); + + await clk.tickAsync(1001); + assert.true(throttledSent); + }); + + it("drops deferred sends once the queue has stopped", async ({ throttle, queue, clk }) => { + const peer = makePeer("1.1.1.1"); + + await throttle.throttle(peer, "getStatus"); + await throttle.throttle(peer, "getStatus"); + + let sent = false; + void throttle.throttle(peer, "getStatus").then(() => { + sent = true; + }); + await clk.tickAsync(0); + + const pushedBeforeStop = queue.pushed; + await queue.stop(); + await clk.tickAsync(5000); + + assert.false(sent); + assert.equal(queue.pushed, pushedBeforeStop); + }); +}); diff --git a/packages/p2p/source/throttle.ts b/packages/p2p/source/throttle.ts index 0cd9c414c..dfae84e7a 100644 --- a/packages/p2p/source/throttle.ts +++ b/packages/p2p/source/throttle.ts @@ -2,7 +2,6 @@ import type { Contracts } from "@mainsail/contracts"; import { Identifiers } from "@mainsail/constants"; import { inject, injectable, tagged } from "@mainsail/container"; -import delay from "delay"; import { RateLimiter } from "./rate-limiter.js"; import { buildRateLimiter } from "./utils/index.js"; @@ -55,23 +54,35 @@ export class Throttle { } async #process(peer: Contracts.P2P.Peer, event: string, resolve: () => void): Promise { - if (await this.#outgoingRateLimiter.hasExceededRateLimitNoConsume(peer.ip, event)) { + const retryAfter = await this.#outgoingRateLimiter.msBeforeNext(peer.ip, event); + + if (retryAfter > 0) { this.logger.debug( `Throttling outgoing requests to ${peer.ip}/${event} to avoid triggering their rate limit`, "p2p", ); - await delay(100); + // Wait outside the queue so a throttled peer or endpoint never + // delays sends to anybody else; requeue once the budget has + // actually refilled instead of checking again on a fixed interval. + setTimeout(() => { + // Mirror the queue's own stop() semantics: pending work is dropped on shutdown. + if (!this.#queue.isStarted()) { + return; + } + + void this.#queue.push({ + handle: async () => { + await this.#process(peer, event, resolve); + }, + }); + }, retryAfter); + + return; + } - void this.#queue.push({ - handle: async () => { - await this.#process(peer, event, resolve); - }, - }); - } else { - await this.#outgoingRateLimiter.consume(peer.ip, event); + await this.#outgoingRateLimiter.consume(peer.ip, event); - resolve(); - } + resolve(); } }