feat: RPC reliability study tooling and ingestion load test

docker-compose.yml

@@ -83,5 +83,24 @@ services:
     depends_on:
       - api
 
+  # RPC reliability study: polls public mainnet endpoints through @heightproof/web3 with
+  # cross-provider verification. Opt-in: docker compose --profile study up -d
+  # Report: docker compose --profile study run --rm study npm run study:report
+  study:
+    profiles: ["study"]
+    build:
+      context: .
+      dockerfile: infra/study.Dockerfile
+    restart: unless-stopped
+    environment:
+      HEIGHTPROOF_ENDPOINT: http://api:8080
+      DATABASE_URL: postgres://heightproof:heightproof@postgres:5432/heightproof
+      STUDY_STATE: /data/state.json
+    volumes:
+      - heightproof-study:/data
+    depends_on:
+      - api
+
 volumes:
   heightproof-postgres:
+  heightproof-study:

docs/performance.md

@@ -0,0 +1,32 @@
+# Performance
+
+Measured with `scripts/load/ingest.ts` against `docker compose up` on one machine (10 CPU,
+8 GB to Docker), single worker, Postgres 16 and Redis 7 from the compose file. Each run sends
+30 seconds of traffic from 8 concurrent senders in batches of 100, with a real project API key.
+
+| | REST events | EVM RPC events |
+|---|---|---|
+| Accepted by the API | 59,000 in 30 s | 59,000 in 30 s |
+| Accept latency (all responses) p50 / p95 / p99 | 2.9 / 17.5 / 37.8 ms | 3.4 / 21.5 / 55.4 ms |
+| Worker throughput while backlogged | **~1,030 events/s** | **~100 events/s** |
+| Peak backlog | 56,109 events (≈55 s) | 58,074 events (≈580 s) |
+| Drained after sending stopped | 27 s | 559 s |
+
+## What limits it
+
+1. **The per-IP rate limit, at the API.** 600 requests per minute per IP, all routes. With
+   100-event batches that is 1,000 events/s sustained from one address; the test spent its
+   minute's allowance in its first 30 s and every later batch got `429`. One SDK instance behind
+   one NAT shares this budget. Health and metrics endpoints are exempt.
+2. **The worker, for RPC traffic.** Every EVM RPC event runs about nine context queries against
+   `api_events` (stale head, disagreement, reorg, throttling, cost, baselines) before it is
+   written, one event at a time. REST events skip those and run ten times faster. This is the
+   real ceiling: ~100 RPC events/s per worker. Scaling out is supported (consumer group plus
+   `XCLAIM`), but per-event query cost is the thing to cut first: batch the context lookups per
+   endpoint per batch, and keep rolling per-endpoint state (head, baselines) instead of
+   re-deriving it from `api_events` on every event.
+3. **Not yet measured:** behaviour as `api_events` grows into millions of rows (no partitioning
+   or retention yet), and multiple workers.
+
+Accept latency percentiles include the fast `429` rejections, so they slightly flatter the
+`202` path.

infra/study.Dockerfile

@@ -0,0 +1,13 @@
+# Collector for the RPC reliability study (docker compose --profile study up -d).
+FROM node:22-alpine
+WORKDIR /app
+COPY package.json package-lock.json* tsconfig.json ./
+COPY packages ./packages
+COPY apps ./apps
+COPY examples ./examples
+COPY scripts ./scripts
+RUN npm install \
+  && npm run build -w packages/shared -w packages/sdk-node -w packages/web3 \
+  && mkdir -p /data && chown node:node /data
+USER node
+CMD ["npx", "tsx", "scripts/study/collect.ts"]

package.json

@@ -9,6 +9,8 @@
     "dev": "concurrently \"npm run dev -w apps/api\" \"npm run dev -w apps/worker\" \"npm run dev -w apps/dashboard\"",
     "seed:demo": "tsx scripts/seed-demo-events.ts",
     "demo:unreliable-provider": "tsx scripts/demo-unreliable-provider.ts",
+    "study:collect": "tsx scripts/study/collect.ts",
+    "study:report": "tsx scripts/study/report.ts",
     "lint": "eslint .",
     "test": "npm run test --workspaces --if-present",
     "test:integration": "npm run test:integration -w apps/worker",

scripts/load/ingest.ts

@@ -0,0 +1,154 @@
+/**
+ * Ingestion load test.
+ *
+ * Phase 1 drives POST /v1/events at a fixed concurrency and records accept latency, throughput and
+ * rejections. Phase 2 watches the worker drain what was accepted and records its processing rate,
+ * which is the pipeline's real throughput ceiling. Run against a stack nothing else is loading.
+ *
+ * Usage: npx tsx scripts/load/ingest.ts [--kind rest|evm_rpc] [--seconds 30] [--concurrency 8] [--batch 100]
+ */
+import { randomUUID } from "node:crypto";
+import pg from "pg";
+
+const arg = (name: string, fallback: string) => {
+  const index = process.argv.indexOf(`--${name}`);
+  return index > -1 ? process.argv[index + 1] : fallback;
+};
+const kind = arg("kind", "rest") as "rest" | "evm_rpc";
+const seconds = Number(arg("seconds", "30"));
+const concurrency = Number(arg("concurrency", "8"));
+const batchSize = Number(arg("batch", "100"));
+const api = process.env.HEIGHTPROOF_ENDPOINT ?? "http://localhost:8080";
+const databaseUrl =
+  process.env.DATABASE_URL ?? "postgres://heightproof:heightproof@localhost:5432/heightproof";
+
+// A fresh workspace per run: a real project-scoped key exercises the production auth path (a
+// key lookup in Postgres on every batch), and its events are easy to count in isolation.
+const suffix = `${kind}-${Date.now().toString(36)}`;
+const signup = await fetch(new URL("/v1/auth/signup", api), {
+  method: "POST",
+  headers: { "content-type": "application/json" },
+  body: JSON.stringify({
+    email: `loadtest-${suffix}@heightproof.local`,
+    password: `loadtest-${suffix}-password`,
+    organizationName: `Load test ${suffix}`,
+    projectName: "loadtest"
+  })
+});
+if (!signup.ok) throw new Error(`signup failed: ${signup.status} ${await signup.text()}`);
+const workspace = (await signup.json()) as {
+  projects: Array<{ id: string }>;
+  apiKey: { key: string };
+};
+const projectId = workspace.projects[0].id;
+const apiKey = workspace.apiKey.key;
+
+function event(i: number) {
+  const base = {
+    id: randomUUID(),
+    projectId,
+    serviceName: "loadtest",
+    environment: "loadtest",
+    timestamp: new Date().toISOString(),
+    request: {
+      method: "POST" as const,
+      path: kind === "rest" ? `/api/orders/${i % 500}` : "/rpc",
+      route: kind === "rest" ? "/api/orders/:id" : "/rpc",
+      ip: `198.51.100.${i % 200}`,
+      headers: {},
+      query: {},
+      auth: { present: true, failed: false }
+    },
+    response: { statusCode: 200, latencyMs: 20 + (i % 80) }
+  };
+  if (kind === "rest") return { ...base, kind: "rest" };
+  return {
+    ...base,
+    kind: "evm_rpc",
+    evmRpc: {
+      method: ["eth_blockNumber", "eth_getBalance", "eth_call"][i % 3],
+      chainId: "1",
+      provider: `provider-${i % 4}`,
+      endpointHash: `endpoint-${i % 4}`,
+      blockNumber: String(21_000_000 + Math.floor(i / 40)),
+      resultShape: "value",
+      resultHash: `hash-${i % 7}`
+    }
+  };
+}
+
+const latencies: number[] = [];
+const statuses = new Map<number, number>();
+let accepted = 0;
+let sent = 0;
+const deadline = Date.now() + seconds * 1000;
+const started = Date.now();
+
+async function sender() {
+  while (Date.now() < deadline) {
+    const events = Array.from({ length: batchSize }, () => event(sent++));
+    const t0 = performance.now();
+    const response = await fetch(new URL("/v1/events", api), {
+      method: "POST",
+      headers: { "content-type": "application/json", "x-heightproof-api-key": apiKey },
+      body: JSON.stringify({ events })
+    }).catch(() => undefined);
+    latencies.push(performance.now() - t0);
+    const status = response?.status ?? 0;
+    statuses.set(status, (statuses.get(status) ?? 0) + 1);
+    if (status === 202) accepted += events.length;
+    if (status === 429) await new Promise((r) => setTimeout(r, 100));
+  }
+}
+
+// Rows landing in Postgres, sampled every second from the start, give the worker's real
+// processing rate and how far behind it fell. (Polling /ready would share the API's per-IP rate
+// limit with the senders.)
+const pool = new pg.Pool({ connectionString: databaseUrl });
+const count = async () =>
+  Number(
+    (await pool.query(`select count(*) from api_events where project_id = $1`, [projectId])).rows[0]
+      .count
+  );
+const samples: Array<{ t: number; accepted: number; persisted: number }> = [];
+let sampling = true;
+const sampler = (async () => {
+  while (sampling) {
+    samples.push({ t: (Date.now() - started) / 1000, accepted, persisted: await count() });
+    await new Promise((r) => setTimeout(r, 1000));
+  }
+})();
+
+console.log(
+  `phase 1: ${kind} events, ${concurrency} senders × ${batchSize}/batch for ${seconds}s → project ${projectId}`
+);
+await Promise.all(Array.from({ length: concurrency }, sender));
+const sendSeconds = (Date.now() - started) / 1000;
+
+latencies.sort((a, b) => a - b);
+const p = (q: number) =>
+  latencies[Math.min(latencies.length - 1, Math.floor(q * latencies.length))].toFixed(1);
+console.log(
+  `  requests ${latencies.length} (${[...statuses].map(([s, n]) => `${s}: ${n}`).join(", ")})\n` +
+    `  accepted ${accepted} events = ${(accepted / sendSeconds).toFixed(0)} events/s averaged over the run`
+);
+console.log(
+  `  accept latency p50 ${p(0.5)} ms · p95 ${p(0.95)} ms · p99 ${p(0.99)} ms · max ${latencies.at(-1)!.toFixed(1)} ms`
+);
+
+while (samples.at(-1)!.persisted < accepted && (Date.now() - started) / 1000 < sendSeconds + 900) {
+  await new Promise((r) => setTimeout(r, 1000));
+}
+sampling = false;
+await sampler;
+
+// Worker rate: from the first persisted row to the last, the backlog was never empty.
+const first = samples.find((s) => s.persisted > 0)!;
+const done = samples.find((s) => s.persisted >= accepted) ?? samples.at(-1)!;
+const workerRate = (done.persisted - first.persisted) / Math.max(1, done.t - first.t);
+const peakBacklog = Math.max(...samples.map((s) => s.accepted - s.persisted));
+console.log(
+  `phase 2: persisted ${done.persisted}/${accepted}; worker processed ${workerRate.toFixed(0)} events/s while it had a backlog\n` +
+    `  peak backlog ${peakBacklog} events (≈${(peakBacklog / workerRate).toFixed(0)}s behind); fully drained ${Math.max(0, done.t - sendSeconds).toFixed(0)}s after sending stopped`
+);
+await pool.end();

scripts/study/collect.ts

@@ -0,0 +1,147 @@
+/**
+ * RPC reliability study: collector.
+ *
+ * Polls several Ethereum mainnet endpoints on a fixed interval through @heightproof/web3, with
+ * cross-provider verification on, so Heightproof records every answer and every detector finding.
+ * Meant to run unattended for days (see the `study` compose profile); `report.ts` turns what it
+ * collected into numbers.
+ *
+ * The first run signs up a dedicated workspace and keeps its API key in STUDY_STATE, so restarts
+ * keep writing to the same project.
+ */
+import { randomBytes } from "node:crypto";
+import { mkdir, readFile, writeFile } from "node:fs/promises";
+import { dirname } from "node:path";
+import { createHeightproofRpcClient } from "@heightproof/web3";
+
+type Endpoint = { name: string; url: string };
+type StudyState = { projectId: string; apiKey: string; startedAt: string };
+
+const DEFAULT_ENDPOINTS =
+  "publicnode=https://ethereum-rpc.publicnode.com," +
+  "drpc=https://eth.drpc.org," +
+  "blastapi=https://eth-mainnet.public.blastapi.io," +
+  "mevblocker=https://rpc.mevblocker.io," +
+  "onfinality=https://eth.api.onfinality.io/public";
+
+const api = process.env.HEIGHTPROOF_ENDPOINT ?? "http://localhost:8080";
+const statePath = process.env.STUDY_STATE ?? ".study/state.json";
+const intervalMs = Number(process.env.STUDY_INTERVAL_MS ?? 15_000);
+const sampleRate = Number(process.env.STUDY_VERIFY_SAMPLE_RATE ?? 0.2);
+const callTimeoutMs = Number(process.env.STUDY_CALL_TIMEOUT_MS ?? 10_000);
+const endpoints: Endpoint[] = (process.env.STUDY_ENDPOINTS ?? DEFAULT_ENDPOINTS)
+  .split(",")
+  .map((entry) => {
+    const [name, ...url] = entry.trim().split("=");
+    return { name, url: url.join("=") };
+  });
+
+// Deterministic reads whose answers every honest provider must agree on at a given height.
+const WALLET = "0xd8dA6BF26964aF9D7eEd9e03E53415D37aA96045";
+const USDC = "0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48";
+const CALLS: Array<{ method: string; params: unknown[] }> = [
+  { method: "eth_blockNumber", params: [] },
+  { method: "eth_getBalance", params: [WALLET, "latest"] },
+  { method: "eth_call", params: [{ to: USDC, data: "0x18160ddd" }, "latest"] }
+];
+
+const state = await loadOrCreateState();
+console.log(
+  `study project ${state.projectId} (since ${state.startedAt}); ${endpoints.length} endpoints every ${intervalMs}ms, verify sample ${sampleRate}`
+);
+
+const clients = endpoints.map((endpoint) => ({
+  endpoint,
+  client: createHeightproofRpcClient({
+    projectId: state.projectId,
+    apiKey: state.apiKey,
+    endpoint: api,
+    rpcUrl: endpoint.url,
+    provider: endpoint.name,
+    chainId: 1,
+    serviceName: "rpc-study",
+    environment: "study",
+    verify: {
+      endpoints: endpoints
+        .filter((other) => other !== endpoint)
+        .map((other) => ({ rpcUrl: other.url, provider: other.name })),
+      sampleRate,
+      maxCallsPerMinute: 20,
+      timeoutMs: callTimeoutMs
+    }
+  })
+}));
+
+let stopping = false;
+process.on("SIGTERM", () => (stopping = true));
+process.on("SIGINT", () => (stopping = true));
+
+const totals = new Map(endpoints.map((endpoint) => [endpoint.name, { ok: 0, failed: 0 }]));
+for (let tick = 1; !stopping; tick++) {
+  const started = Date.now();
+  await Promise.all(
+    clients.map(async ({ endpoint, client }) => {
+      for (const call of CALLS) {
+        const outcome = await withTimeout(client.request(call), callTimeoutMs).then(
+          () => "ok" as const,
+          () => "failed" as const
+        );
+        totals.get(endpoint.name)![outcome] += 1;
+      }
+    })
+  );
+
+  if (tick % 20 === 0) {
+    const summary = [...totals].map(([name, t]) => `${name} ${t.ok}/${t.ok + t.failed}`).join("  ");
+    console.log(`${new Date().toISOString()} tick ${tick}  ok/total: ${summary}`);
+  }
+  await sleep(Math.max(0, intervalMs - (Date.now() - started)));
+}
+// Give the SDK one flush interval to send what it has queued.
+await sleep(6_000);
+process.exit(0);
+
+async function loadOrCreateState(): Promise<StudyState> {
+  try {
+    return JSON.parse(await readFile(statePath, "utf8")) as StudyState;
+  } catch {
+    // First run: fall through and create a workspace.
+  }
+
+  const suffix = randomBytes(4).toString("hex");
+  const response = await fetch(new URL("/v1/auth/signup", api), {
+    method: "POST",
+    headers: { "content-type": "application/json" },
+    body: JSON.stringify({
+      email: `rpc-study-${suffix}@heightproof.local`,
+      password: randomBytes(18).toString("base64url"),
+      organizationName: "RPC reliability study",
+      projectName: "mainnet-study"
+    })
+  });
+  if (!response.ok) throw new Error(`signup failed: ${response.status} ${await response.text()}`);
+  const body = (await response.json()) as {
+    projects: Array<{ id: string }>;
+    apiKey: { key: string };
+  };
+
+  const created: StudyState = {
+    projectId: body.projects[0].id,
+    apiKey: body.apiKey.key,
+    startedAt: new Date().toISOString()
+  };
+  await mkdir(dirname(statePath), { recursive: true });
+  await writeFile(statePath, JSON.stringify(created, null, 2), { mode: 0o600 });
+  return created;
+}
+
+function withTimeout<T>(promise: Promise<T>, ms: number) {
+  return Promise.race([
+    promise,
+    new Promise<never>((_, reject) => setTimeout(() => reject(new Error("timeout")), ms))
+  ]);
+}
+
+function sleep(ms: number) {
+  return new Promise((resolve) => setTimeout(resolve, ms));
+}