From 898a9861c321c4e36f3b5865330edecabe71c187 Mon Sep 17 00:00:00 2001 From: macjayz Date: Thu, 29 Oct 2026 18:17:44 +0100 Subject: [PATCH 1/2] feat: add a collector and report for the RPC reliability study The roadmap's Phase 3 measurement study needs real traffic, collected for days without supervision. scripts/study/collect.ts polls several Ethereum mainnet endpoints every 15s through @heightproof/web3 (eth_blockNumber, eth_getBalance, an eth_call) with cross-provider verification on, so every answer and every detector finding lands in Heightproof. It signs up a dedicated workspace on first run and keeps the key in a state file, so restarts write to the same project. scripts/study/report.ts reads the results from Postgres and prints a Markdown summary per provider: transport failures, JSON-RPC errors inside 200s, rate limiting, latency, chain-head lag against the other providers in the same tick, and disagreement at equal height from verification runs. Runs as an opt-in compose profile: docker compose --profile study up -d docker compose --profile study run --rm study npm run study:report --- docker-compose.yml | 19 ++++ infra/study.Dockerfile | 13 +++ package.json | 2 + scripts/study/collect.ts | 147 ++++++++++++++++++++++++++ scripts/study/report.ts | 220 +++++++++++++++++++++++++++++++++++++++ 5 files changed, 401 insertions(+) create mode 100644 infra/study.Dockerfile create mode 100644 scripts/study/collect.ts create mode 100644 scripts/study/report.ts diff --git a/docker-compose.yml b/docker-compose.yml index 842f208..b747961 100644 --- a/docker-compose.yml +++ b/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: diff --git a/infra/study.Dockerfile b/infra/study.Dockerfile new file mode 100644 index 0000000..7851aff --- /dev/null +++ b/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"] diff --git a/package.json b/package.json index 0231bea..ea1416a 100644 --- a/package.json +++ b/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", diff --git a/scripts/study/collect.ts b/scripts/study/collect.ts new file mode 100644 index 0000000..f657698 --- /dev/null +++ b/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 { + 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(promise: Promise, ms: number) { + return Promise.race([ + promise, + new Promise((_, reject) => setTimeout(() => reject(new Error("timeout")), ms)) + ]); +} + +function sleep(ms: number) { + return new Promise((resolve) => setTimeout(resolve, ms)); +} diff --git a/scripts/study/report.ts b/scripts/study/report.ts new file mode 100644 index 0000000..f94069b --- /dev/null +++ b/scripts/study/report.ts @@ -0,0 +1,220 @@ +/** + * RPC reliability study: report. + * + * Reads what the collector recorded and prints a Markdown summary per provider: availability, + * errors inside 200 responses, rate limiting, latency, how often each provider's chain head + * trailed the others, and how often providers disagreed at the same block height. + * + * Usage: DATABASE_URL=... npx tsx scripts/study/report.ts [--out report.md] + * The project comes from STUDY_PROJECT_ID, or from the collector's STUDY_STATE file. + */ +import { readFile, writeFile } from "node:fs/promises"; +import pg from "pg"; + +const databaseUrl = + process.env.DATABASE_URL ?? "postgres://heightproof:heightproof@localhost:5432/heightproof"; +const intervalSeconds = Number(process.env.STUDY_INTERVAL_MS ?? 15_000) / 1000; +const outIndex = process.argv.indexOf("--out"); +const outPath = outIndex > -1 ? process.argv[outIndex + 1] : undefined; + +const projectId = + process.env.STUDY_PROJECT_ID ?? + ( + JSON.parse(await readFile(process.env.STUDY_STATE ?? ".study/state.json", "utf8")) as { + projectId: string; + } + ).projectId; + +const pool = new pg.Pool({ connectionString: databaseUrl }); +const q = async (sql: string, params: unknown[] = [projectId]) => + (await pool.query(sql, params)).rows as T[]; + +// Primary calls only: verification (shadow) calls are reported separately as disagreements. +const PRIMARY = `project_id = $1 and kind = 'evm_rpc' and evm_shadow_of_trace_id is null`; + +const [window] = await q<{ first: Date; last: Date; calls: number }>( + `select min(timestamp) as first, max(timestamp) as last, count(*)::int as calls from api_events where ${PRIMARY}` +); +if (!window?.calls) throw new Error(`no study data for project ${projectId}`); +const hours = (window.last.getTime() - window.first.getTime()) / 3_600_000; + +const availability = await q<{ + provider: string; + calls: number; + transport_failed: number; + error_in_body: number; + throttled: number; + p50: number; + p95: number; +}>( + `select evm_provider as provider, count(*)::int as calls, + count(*) filter (where status_code >= 400)::int as transport_failed, + count(*) filter (where evm_result_shape = 'error_in_body')::int as error_in_body, + count(*) filter (where evm_throttled)::int as throttled, + percentile_cont(0.5) within group (order by latency_ms) filter (where status_code < 400) as p50, + percentile_cont(0.95) within group (order by latency_ms) filter (where status_code < 400) as p95 + from api_events where ${PRIMARY} + group by 1 order by 1` +); + +// Every provider is polled at the same moment each tick, so within one tick the highest head any +// provider reported is the reference. Lag of 1 block is mostly a block landing mid-tick; the +// stale-head detector's threshold is more than 3. +const headLag = await q<{ + provider: string; + samples: number; + lag_ge_2: number; + lag_gt_3: number; + max_lag: number; +}>( + `with heads as ( + select evm_provider as provider, evm_block_number::bigint as head, + floor(extract(epoch from timestamp) / $2) as tick + from api_events + where ${PRIMARY} and evm_rpc_method = 'eth_blockNumber' and evm_block_number is not null + ), lagged as ( + select provider, max(head) over (partition by tick) - head as lag from heads + ) + select provider, count(*)::int as samples, + count(*) filter (where lag >= 2)::int as lag_ge_2, + count(*) filter (where lag > 3)::int as lag_gt_3, + coalesce(max(lag), 0)::int as max_lag + from lagged group by 1 order by 1`, + [projectId, intervalSeconds] +); + +// One verification run = every endpoint answering the same call pinned to the same height. +const disagreements = await q<{ method: string; runs: number; disagreeing: number }>( + `with runs as ( + select evm_shadow_of_trace_id as run, evm_rpc_method as method, evm_block_number as block, + count(distinct evm_result_hash) filter (where evm_result_shape <> 'error_in_body') as answers + from api_events + where project_id = $1 and evm_shadow_of_trace_id is not null + group by 1, 2, 3 + ) + select method, count(*)::int as runs, count(*) filter (where answers > 1)::int as disagreeing + from runs group by 1 order by 1` +); + +const minority = await q<{ provider: string; times_in_minority: number }>( + `with answers as ( + select evm_shadow_of_trace_id as run, evm_block_number as block, evm_provider as provider, evm_result_hash as hash + from api_events + where project_id = $1 and evm_shadow_of_trace_id is not null and evm_result_shape <> 'error_in_body' + ), counted as ( + select run, block, hash, count(*) as votes from answers group by 1, 2, 3 + ), majority as ( + select distinct on (run, block) run, block, hash from counted order by run, block, votes desc + ) + select a.provider, count(*)::int as times_in_minority + from answers a join majority m using (run, block) + where a.hash is distinct from m.hash + group by 1 order by 2 desc` +); + +const signals = await q<{ signal: string; provider: string; events: number }>( + `select s->>'name' as signal, evm_provider as provider, count(*)::int as events + from api_events, jsonb_array_elements(threat_signals) s + where ${PRIMARY} + group by 1, 2 order by 1, 2` +); + +const incidents = await q<{ + title: string; + severity: string; + status: string; + request_count: number; +}>( + `select title, severity, status, request_count from incidents where project_id = $1 order by request_count desc` +); + +const pct = (part: number, whole: number) => + whole ? `${((100 * part) / whole).toFixed(2)}%` : "—"; +const ms = (value: number | null) => (value === null ? "—" : `${Math.round(value)} ms`); +const table = (head: string[], rows: Array>) => + [ + `| ${head.join(" | ")} |`, + `|${head.map(() => "---").join("|")}|`, + ...rows.map((r) => `| ${r.join(" | ")} |`) + ].join("\n"); + +const report = `# Ethereum mainnet RPC reliability + +Project \`${projectId}\` · ${window.first.toISOString()} → ${window.last.toISOString()} (${hours.toFixed(1)} h) · ${window.calls.toLocaleString()} primary calls · polled every ${intervalSeconds}s + +## Availability and answers + +${table( + [ + "Provider", + "Calls", + "Transport failures", + "JSON-RPC error inside 200", + "Rate limited", + "p50", + "p95" + ], + availability.map((r) => [ + r.provider, + r.calls, + pct(r.transport_failed, r.calls), + pct(r.error_in_body, r.calls), + pct(r.throttled, r.calls), + ms(r.p50), + ms(r.p95) + ]) +)} + +## Chain head lag (vs. the highest head any provider reported in the same tick) + +${table( + ["Provider", "Samples", "≥ 2 blocks behind", "> 3 blocks behind", "Worst"], + headLag.map((r) => [ + r.provider, + r.samples, + pct(r.lag_ge_2, r.samples), + pct(r.lag_gt_3, r.samples), + `${r.max_lag} blocks` + ]) +)} + +## Disagreement at equal height (verification runs) + +${table( + ["Method", "Runs", "Runs with differing answers"], + disagreements.map((r) => [r.method, r.runs, `${r.disagreeing} (${pct(r.disagreeing, r.runs)})`]) +)} + +${ + minority.length + ? table( + ["Provider", "Times in the minority"], + minority.map((r) => [r.provider, r.times_in_minority]) + ) + : "No provider ever disagreed with the majority." +} + +## Detector findings + +${ + signals.length + ? table( + ["Signal", "Provider", "Events"], + signals.map((r) => [r.signal, r.provider ?? "—", r.events]) + ) + : "None." +} + +${ + incidents.length + ? table( + ["Incident", "Severity", "Status", "Events"], + incidents.map((r) => [r.title, r.severity, r.status, r.request_count]) + ) + : "No incidents." +} +`; + +if (outPath) await writeFile(outPath, report); +console.log(report); +await pool.end(); From 3a37d5cc97a55f65bafcc5181604fd5469c5acd7 Mon Sep 17 00:00:00 2001 From: macjayz Date: Thu, 29 Oct 2026 18:36:37 +0100 Subject: [PATCH 2/2] perf: add an ingestion load test and record the first results scripts/load/ingest.ts drives POST /v1/events at fixed concurrency with a real project key, then samples Postgres every second to measure how fast the worker drains the backlog. docs/performance.md records the first run: the API accepts ~2,000 events/s per client IP until the 600 req/min rate limit, the worker processes ~1,030 REST events/s but only ~100 EVM RPC events/s, because each RPC event runs about nine detector context queries before it is written. --- docs/performance.md | 32 +++++++++ scripts/load/ingest.ts | 154 +++++++++++++++++++++++++++++++++++++++++ 2 files changed, 186 insertions(+) create mode 100644 docs/performance.md create mode 100644 scripts/load/ingest.ts diff --git a/docs/performance.md b/docs/performance.md new file mode 100644 index 0000000..40c12bc --- /dev/null +++ b/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. diff --git a/scripts/load/ingest.ts b/scripts/load/ingest.ts new file mode 100644 index 0000000..272fd46 --- /dev/null +++ b/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(); +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();