support archive body and header caches

Author: mfornos

bench/network/archive.bench.ts

@@ -0,0 +1,59 @@
+import { printSummary, runBench, type Script } from "./base";
+
+const PROVIDERS = [
+	"ws://localhost:8181",
+	"wss://polkadot-public-rpc.blockops.network/ws",
+	"wss://polkadot.public.curie.radiumblock.co/ws",
+	"wss://rockx-dot.w3node.com/polka-public-dot/ws",
+	"wss://rpc-polkadot.luckyfriday.io",
+	"wss://polkadot.api.onfinality.io/public-ws",
+	"wss://rpc.ibp.network/polkadot",
+];
+
+export const simpleArchive = (): Script => {
+	let finalizedHeight: null | number = null;
+	let hashByHeight: null | string = null;
+	return {
+		onOpen: (send) => {
+			send("archive_v1_finalizedHeight");
+		},
+
+		onResponse: (res, send) => {
+			if (res != null) {
+				if (finalizedHeight === null) {
+					finalizedHeight = Number(res);
+					send("archive_v1_hashByHeight", [finalizedHeight]);
+				} else if (hashByHeight === null) {
+					hashByHeight = res[0];
+					send("archive_v1_header", [hashByHeight]);
+					send("archive_v1_body", [hashByHeight]);
+				}
+			} else {
+				console.error(`Error: null response`);
+			}
+		},
+
+		onError: (error) => {
+			console.error(`Error: ${error.message}`);
+		},
+	};
+};
+
+(async () => {
+	const provider = PROVIDERS[0]!;
+	const opts = {
+		iterations: 100,
+		warmup: 0,
+	};
+	const durationMs = 5_000;
+	console.log(
+		`Benchmarking ${provider} (iters=${opts.iterations}, duration=${durationMs / 1_000}s)`,
+	);
+
+	const stats = runBench(provider, opts, simpleArchive);
+
+	setTimeout(() => {
+		printSummary(stats);
+		process.exit(0);
+	}, durationMs);
+})();

bench/network/chain-head.ts bench/network/base.tsbench/network/chain-head.ts renamed to bench/network/base.ts

@@ -1,8 +1,6 @@
 import { isSuccess } from "@/json-rpc/util";
 
-type Method = "follow" | "header" | "body";
-
-type LatencyMap = Record<Method, { n: number; total: number; data: number[] }>;
+type LatencyMap = Record<string, { n: number; total: number; data: number[] }>;
 
 type Stats = {
 	host: string;
@@ -15,7 +13,7 @@ type Stats = {
 	latency: LatencyMap;
 };
 
-type Pending = { method: Method; sent: number };
+type Pending = { method: string; sent: number };
 
 export type Script = {
 	onOpen?: (send: Send) => void;
@@ -24,18 +22,7 @@ export type Script = {
 	onEvent?: (subId: string | null, ev: any, send: Send) => void;
 };
 
-type Send = (
-	method: Method,
-	rpcMethod: string,
-	params: any[],
-	inFollow?: boolean,
-) => void;
-
-const emptyLatency = (): LatencyMap => ({
-	follow: { n: 0, total: 0, data: [] },
-	header: { n: 0, total: 0, data: [] },
-	body: { n: 0, total: 0, data: [] },
-});
+type Send = (rpcMethod: string, params?: any[]) => number;
 
 const createStats = (host: string): Stats => ({
 	host,
@@ -45,10 +32,10 @@ const createStats = (host: string): Stats => ({
 	exfiltrated: 0,
 	msgs: 0,
 	events: 0,
-	latency: emptyLatency(),
+	latency: {},
 });
 
-export function runChainHeadBench(
+export function runBench(
 	endpoint: string,
 	{
 		iterations,
@@ -70,12 +57,13 @@ export function runChainHeadBench(
 		let nextId = 1;
 		let subId: string | null = null;
 
-		const send: Send = (method, rpcMethod, params) => {
+		const send: Send = (rpcMethod, params) => {
 			const id = nextId++;
-			pendings.set(id, { method, sent: performance.now() });
+			pendings.set(id, { method: rpcMethod, sent: performance.now() });
 			ws.send(
 				JSON.stringify({ jsonrpc: "2.0", id, method: rpcMethod, params }),
 			);
+			return id;
 		};
 
 		ws.onopen = () => {
@@ -96,7 +84,12 @@ export function runChainHeadBench(
 					const ok = isSuccess(msg);
 					if (!isWarm) {
 						const dt = now - p.sent;
-						const lat = stats.latency[p.method];
+
+						if (stats.latency[p.method] === undefined) {
+							stats.latency[p.method] = { n: 0, total: 0, data: [] };
+						}
+						const lat = stats.latency[p.method]!;
+
 						lat.n++;
 						lat.total += dt;
 						lat.data.push(dt);
@@ -109,7 +102,7 @@ export function runChainHeadBench(
 					} else {
 						script.onError?.(msg.error, send);
 					}
-					if (p.method === "follow") subId = msg.result;
+					if (p.method === "chainHead_v1_follow") subId = msg.result;
 				} else {
 					stats.exfiltrated++;
 				}
@@ -203,7 +196,9 @@ export const printSummary = (s: Stats) => {
 	);
 	console.log("Latency (ms):");
 
-	for (const method of Object.keys(s.latency) as Method[]) {
+	for (const method of Object.keys(s.latency) as string[]) {
+		if (s.latency[method] === undefined) continue;
+
 		const xs = s.latency[method].data;
 		const r = summarize(xs);
 

bench/network/follow.bench.ts

@@ -1,9 +1,8 @@
-import { printSummary, runChainHeadBench, type Script } from "./chain-head";
+import { printSummary, runBench, type Script } from "./base";
 
 const PROVIDERS = [
 	"ws://localhost:8181",
 	"wss://polkadot-public-rpc.blockops.network/ws",
-	"wss://polkadot-rpc.dwellir.com",
 	"wss://polkadot.public.curie.radiumblock.co/ws",
 	"wss://rockx-dot.w3node.com/polka-public-dot/ws",
 	"wss://rpc-polkadot.luckyfriday.io",
@@ -21,7 +20,7 @@ export const simpleFollow = (): Script => {
 	const ops = new Set<string>();
 	return {
 		onOpen: (send) => {
-			send("follow", "chainHead_v1_follow", [true]);
+			send("chainHead_v1_follow", [true]);
 		},
 
 		onResponse: (res) => {
@@ -39,8 +38,8 @@ export const simpleFollow = (): Script => {
 
 		onEvent: (subId, ev, send) => {
 			if (ev.event === "newBlock") {
-				send("body", "chainHead_v1_body", [subId, ev.blockHash]);
-				send("header", "chainHead_v1_header", [subId, ev.blockHash]);
+				send("chainHead_v1_body", [subId, ev.blockHash]);
+				send("chainHead_v1_header", [subId, ev.blockHash]);
 			}
 
 			if (ev.operationId != null) {
@@ -66,15 +65,15 @@ export const simpleFollow = (): Script => {
 (async () => {
 	const provider = PROVIDERS[0]!;
 	const opts = {
-		iterations: 1_000,
+		iterations: 100,
 		warmup: 0,
 	};
 	const durationMs = 10_000;
 	console.log(
 		`Benchmarking ${provider} (iters=${opts.iterations}, duration=${durationMs / 1_000}s)`,
 	);
 
-	const stats = runChainHeadBench(provider, opts, simpleFollow);
+	const stats = runBench(provider, opts, simpleFollow);
 
 	setTimeout(() => {
 		printSummary(stats);

src/json-rpc/polkadot/archive/body.ts

@@ -0,0 +1,10 @@
+import type { CacheConfig } from "@/config";
+import { forwardRequestWithCache } from "../forward";
+
+export const archive_v1_body = (config: CacheConfig) => {
+	return forwardRequestWithCache(
+		"archive_v1_body",
+		config,
+		(req) => req.params[0],
+	);
+};

src/json-rpc/polkadot/archive/header.ts

@@ -0,0 +1,10 @@
+import type { CacheConfig } from "@/config";
+import { forwardRequestWithCache } from "../forward";
+
+export const archive_v1_header = (config: CacheConfig) => {
+	return forwardRequestWithCache(
+		"archive_v1_header",
+		config,
+		(req) => req.params[0],
+	);
+};

src/json-rpc/polkadot/forward.ts

@@ -1,8 +1,45 @@
+import type { CacheConfig } from "@/config";
+import { createLRUCache } from "@/util/cache";
 import type { JSONRPCMethodHandler } from "../methods";
+import type { JSONRPCRequest, JSONRPCResponse } from "../types";
+import { isSuccess } from "../util";
+import { isCacheEnabledFor } from "./chain-head/ops/util";
 
 export const forwardRequest: JSONRPCMethodHandler = {
 	handleRequest: async (upstream, downstream, req) => {
 		const response = await upstream.request(req);
 		downstream.send(response);
 	},
 };
+
+export const forwardRequestWithCache = (
+	method: string,
+	config: CacheConfig,
+	keyOf: (req: JSONRPCRequest) => string,
+): JSONRPCMethodHandler => {
+	if (isCacheEnabledFor(config, method)) {
+		const cache = createLRUCache<JSONRPCResponse>(
+			config.methods?.[method]?.maxSize ?? 100,
+		);
+		return {
+			handleRequest: async (upstream, downstream, req) => {
+				const key = keyOf(req);
+				const cached = cache.get(key);
+				if (cached) {
+					downstream.send({
+						jsonrpc: "2.0",
+						id: req.id ?? null,
+						result: cached,
+					});
+				} else {
+					const response = await upstream.request(req);
+					if (isSuccess(response) && response.result != null) {
+						cache.set(key, response.result);
+					}
+					downstream.send(response);
+				}
+			},
+		};
+	}
+	return forwardRequest;
+};

src/json-rpc/polkadot/index.ts

@@ -1,5 +1,7 @@
 import type { CacheConfig } from "@/config";
 import type { JSONRPCMethodHandler } from "../methods";
+import { archive_v1_body } from "./archive/body";
+import { archive_v1_header } from "./archive/header";
 import { archive_v1_storage, archive_v1_storageDiff } from "./archive/storage";
 import { chainHead_v1_body } from "./chain-head/body";
 import { chainHead_v1_call } from "./chain-head/call";
@@ -40,6 +42,8 @@ export function polkadotMethods(config: CacheConfig) {
 
 		archive_v1_storage: archive_v1_storage(),
 		archive_v1_storageDiff: archive_v1_storageDiff(),
+		archive_v1_body: archive_v1_body(config),
+		archive_v1_header: archive_v1_header(config),
 
 		transactionWatch_v1_submitAndWatch: transactionWatch_v1_submitAndWatch(),
 

wsmux.config.yaml

@@ -21,20 +21,12 @@ cache:
     chainHead_v1_call:
       enabled: true
       max_entries: 25
-
-# -------------------------------------------------------------------
-# IP rate limiting
-# -------------------------------------------------------------------
-rate_limit:
-  enabled: false
-  window: 1m
-  max_requests: 100
-  # Trusted networks for forwarding/proxy headers.
-  # CIDR notation is used to specify the network ranges.
-  trusted_networks:
-    - 192.168.0.0/16
-    - 172.16.0.0/12
-    - 127.0.0.1/32
+    archive_v1_header:
+      enabled: true
+      max_entries: 100
+    archive_v1_body:
+      enabled: true
+      max_entries: 25
 
 # -------------------------------------------------------------------
 # Upstream Servers Settings
@@ -83,3 +75,16 @@ json_rpc:
   max_pending_requests_per_connection: 1500
   max_pending_requests: 5000
   max_requests_per_second: 5000
+# -------------------------------------------------------------------
+# IP rate limiting
+# -------------------------------------------------------------------
+#rate_limit:
+#  enabled: false
+#  window: 1m
+#  max_requests: 100
+#  # Trusted networks for forwarding/proxy headers.
+#  # CIDR notation is used to specify the network ranges.
+#  trusted_networks:
+#    - 192.168.0.0/16
+#    - 172.16.0.0/12
+#    - 127.0.0.1/32