diff --git a/packages/api-common/source/rcp/processor.test.ts b/packages/api-common/source/rcp/processor.test.ts index 5a43e7ab5e..5c59429b51 100644 --- a/packages/api-common/source/rcp/processor.test.ts +++ b/packages/api-common/source/rcp/processor.test.ts @@ -109,6 +109,39 @@ describe<{ ]); }); + it("process should bound batch concurrency and preserve order", async ({ subject }) => { + let inFlight = 0; + let peak = 0; + + subject.registerAction({ + handle: async (parameters: any[]) => { + inFlight++; + peak = Math.max(peak, inFlight); + await new Promise((resolve) => setImmediate(resolve)); + inFlight--; + return `R-${parameters[0]}`; + }, + name: "m", + schema: { $id: undefined }, + } as any); + + const payload = Array.from({ length: 25 }, (_, index) => ({ + id: index, + jsonrpc: "2.0", + method: "m", + params: [index], + })); + + const result = (await subject.process({ payload } as any)) as any[]; + + // never more than BATCH_CONCURRENCY (10) simulations in flight at once + assert.true(peak <= 10); + assert.equal( + result, + payload.map((request) => ({ id: request.id, jsonrpc: "2.0", result: `R-${request.id}` })), + ); + }); + it("process should return MethodNotFound for unknown method", async ({ subject }) => { const result = await subject.process({ payload: { id: 3, jsonrpc: "2.0", method: "unknown", params: [] }, diff --git a/packages/api-common/source/rcp/processor.ts b/packages/api-common/source/rcp/processor.ts index 3f58d1e7aa..19aa2e076c 100644 --- a/packages/api-common/source/rcp/processor.ts +++ b/packages/api-common/source/rcp/processor.ts @@ -4,10 +4,12 @@ import Hapi from "@hapi/hapi"; import { Enums, Identifiers } from "@mainsail/constants"; import { inject, injectable } from "@mainsail/container"; import { RpcError } from "@mainsail/exceptions"; -import { ensureError } from "@mainsail/utils"; +import { chunk, ensureError } from "@mainsail/utils"; import { getRcpId, prepareRcpError } from "./utilities.js"; +const BATCH_CONCURRENCY = 10; + @injectable() export class Processor implements Contracts.Api.RPC.Processor { @inject(Identifiers.Cryptography.Validator) @@ -30,11 +32,16 @@ export class Processor implements Contracts.Api.RPC.Processor { } const payload = request.payload as Contracts.Api.RPC.Request<[]>; - if (Array.isArray(payload)) { - return Promise.all(payload.map(async (rcpRequest) => await this.#processSingle(rcpRequest))); - } else { + + if (!Array.isArray(payload)) { return this.#processSingle(payload); } + + const responses: (Contracts.Api.RPC.Response | Contracts.Api.RPC.Error)[] = []; + for (const group of chunk(payload, BATCH_CONCURRENCY)) { + responses.push(...(await Promise.all(group.map((rcpRequest) => this.#processSingle(rcpRequest))))); + } + return responses; } async #processSingle( diff --git a/packages/api-common/source/validation/schemas.test.ts b/packages/api-common/source/validation/schemas.test.ts new file mode 100644 index 0000000000..e0be915fd8 --- /dev/null +++ b/packages/api-common/source/validation/schemas.test.ts @@ -0,0 +1,58 @@ +import { describe } from "@mainsail/test-runner"; +import { Ajv } from "ajv"; + +import { jsonRpcPayloadSchema, jsonRpcResponseSchema } from "./schemas"; + +const makeRequest = (id: number) => ({ id, jsonrpc: "2.0", method: "eth_call", params: [] }); + +describe<{ + ajv: Ajv; +}>("JSON-RPC schemas", ({ it, beforeEach, assert }) => { + beforeEach((context) => { + context.ajv = new Ajv(); + context.ajv.addSchema(jsonRpcPayloadSchema); + context.ajv.addSchema(jsonRpcResponseSchema); + }); + + it("accepts a single request", ({ ajv }) => { + assert.true(ajv.validate("jsonRpcPayload", makeRequest(1))); + }); + + it("accepts a batch of up to 100 requests", ({ ajv }) => { + assert.true( + ajv.validate( + "jsonRpcPayload", + Array.from({ length: 100 }, (_, index) => makeRequest(index)), + ), + ); + }); + + it("rejects a batch above the amplification cap", ({ ajv }) => { + assert.false( + ajv.validate( + "jsonRpcPayload", + Array.from({ length: 101 }, (_, index) => makeRequest(index)), + ), + ); + }); + + it("rejects an empty batch", ({ ajv }) => { + assert.false(ajv.validate("jsonRpcPayload", [])); + }); + + it("rejects a batch containing a malformed request", ({ ajv }) => { + assert.false(ajv.validate("jsonRpcPayload", [makeRequest(1), { id: 2, jsonrpc: "2.0" }])); + }); + + it("accepts result and error responses", ({ ajv }) => { + assert.true(ajv.validate("jsonRpcResponse", { id: 1, jsonrpc: "2.0", result: "0x" })); + assert.true( + ajv.validate("jsonRpcResponse", { + error: { code: -32_600, message: "Invalid request" }, + id: 1, + jsonrpc: "2.0", + }), + ); + assert.false(ajv.validate("jsonRpcResponse", { id: 1, jsonrpc: "2.0" })); + }); +}); diff --git a/packages/api-common/source/validation/schemas.ts b/packages/api-common/source/validation/schemas.ts index 1751ba3e7d..8b481bd1ac 100644 --- a/packages/api-common/source/validation/schemas.ts +++ b/packages/api-common/source/validation/schemas.ts @@ -60,5 +60,5 @@ export const jsonRpcResponseSchema: SchemaObject = { export const jsonRpcPayloadSchema: SchemaObject = { $id: "jsonRpcPayload", - oneOf: [jsonRpcRequest, { items: jsonRpcRequest, type: "array" }], + oneOf: [jsonRpcRequest, { items: jsonRpcRequest, maxItems: 100, minItems: 1, type: "array" }], };