From df32ce483dbf3e82745891248bc13d6e22ad8253 Mon Sep 17 00:00:00 2001 From: lauferuqquyah Date: Thu, 6 Aug 2026 11:52:58 -0700 Subject: [PATCH] perf(pulse-core): benchmark suite with 20% regression gating (#924) --- .github/workflows/ci.yml | 24 +- docs/ARCHITECTURE.md | 44 ++++ packages/pulse-core/bench/baseline.json | 90 ++++++++ packages/pulse-core/bench/cases.ts | 111 +++++++++ packages/pulse-core/bench/fixtures.ts | 70 ++++++ packages/pulse-core/bench/gate.ts | 135 +++++++++++ packages/pulse-core/bench/harness.ts | 216 ++++++++++++++++++ packages/pulse-core/bench/index.ts | 94 ++++++++ .../pulse-core/bench/soroban-throughput.ts | 215 ----------------- packages/pulse-core/bench/throughput.ts | 157 ------------- packages/pulse-core/package.json | 6 +- 11 files changed, 787 insertions(+), 375 deletions(-) create mode 100644 packages/pulse-core/bench/baseline.json create mode 100644 packages/pulse-core/bench/cases.ts create mode 100644 packages/pulse-core/bench/fixtures.ts create mode 100644 packages/pulse-core/bench/gate.ts create mode 100644 packages/pulse-core/bench/harness.ts create mode 100644 packages/pulse-core/bench/index.ts delete mode 100644 packages/pulse-core/bench/soroban-throughput.ts delete mode 100644 packages/pulse-core/bench/throughput.ts diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index ab6a8044..c40f6255 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -93,7 +93,7 @@ jobs: run: pnpm tsc --noEmit -p packages/anchor-sdk/tsconfig.json - name: Validate ABI specs run: pnpm --filter @orbital-stellar/abi-registry validate - - name: Generate Horizon types & check drift + - name: Generate Horizon types & check drift run: | node scripts/generate-horizon-types.mjs git diff --exit-code packages/pulse-core/src/_raw-horizon.gen.ts || { @@ -291,3 +291,25 @@ jobs: path: | packages/*/coverage/ retention-days: 14 + + + benchmarks: + needs: changes + if: ${{ needs.changes.outputs.packages == 'true' }} + runs-on: ubuntu-latest + permissions: + contents: read + steps: + - uses: actions/checkout@v7 + - uses: pnpm/action-setup@v6 + - uses: actions/setup-node@v7 + with: + node-version: 20 + cache: pnpm + - run: pnpm install --frozen-lockfile + - name: Build abi-registry + run: pnpm tsc -p packages/abi-registry/tsconfig.json + - name: Build pulse-core + run: pnpm tsc -p packages/pulse-core/tsconfig.json + - name: Run benchmarks and gate against baseline + run: pnpm --filter @orbital-stellar/pulse-core bench:ci \ No newline at end of file diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 2b5e124b..0eb2d2d6 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -494,3 +494,47 @@ orbital_stellar/ [`pulse-webhooks`](../packages/pulse-webhooks/README.md), [`pulse-notify`](../packages/pulse-notify/README.md), [`abi-registry`](../packages/abi-registry/README.md) + + +## Performance benchmarks + +The hot path of every event is normalization plus, for Soroban contract events, +spec resolution and decoding. A change that halves throughput there would ship +unnoticed without a guard, so `packages/pulse-core/bench/` holds a benchmark +suite that CI runs on every packages change and gates on regressions. + +### What is measured + +The suite runs four things against the recorded CAP-67 fixture corpus in +`packages/pulse-core/test/fixtures/cap67/`, so the numbers reflect pulse-core's +own work rather than network time: + +- **normalize/raw-to-normalized** - a raw RPC event through `normalizeContractEvent`. +- **decode/cap67-transfer** - the CAP-67 unified `transfer` decoder over the + transfer fixtures (bare `i128` and `SCMap`-with-memo forms). +- **watcher/fan-out-1, -100, -1000** - `Watcher.emit` dispatch at each fan-out width. +- **cursor/memory-set** and **cursor/memory-set-many-100** - cursor write cost + for the in-memory adapter, single-key and batched. + +### Running it +`bench` exits non-zero when any case is more than 20% slower than the committed +baseline, or when a baselined case has gone missing. Improvements never fail. +The harness is dependency-free (`node:perf_hooks` only) and reports the median of +many batched samples, so a single GC pause does not trip the gate. + +### The baseline and how to update it + +`packages/pulse-core/bench/baseline.json` is the committed reference. It records +each case's throughput alongside the machine the numbers came from, because a +throughput figure is meaningless without the hardware behind it. + +Updating the baseline is deliberate. When a change legitimately moves the +numbers - a faster algorithm, or an accepted cost for new behavior - regenerate +the baseline with `bench:update` and **justify the change in the pull request +body**: what moved, why, and roughly by how much. A baseline bump without a +reason in the PR is treated as a red flag in review, because it is the one way a +real regression can be laundered into the reference. + +The current committed baseline was generated on a developer workstation and is +noisier than a dedicated runner would produce; regenerate it on stable hardware +when convenient and record that environment here. \ No newline at end of file diff --git a/packages/pulse-core/bench/baseline.json b/packages/pulse-core/bench/baseline.json new file mode 100644 index 00000000..fc9a9287 --- /dev/null +++ b/packages/pulse-core/bench/baseline.json @@ -0,0 +1,90 @@ +{ + "schemaVersion": 1, + "generatedAt": "2026-08-06T15:08:44.237Z", + "environment": { + "platform": "win32 x64", + "node": "v22.16.0", + "cpu": "Intel(R) Core(TM) i5-6200U CPU @ 2.30GHz", + "cores": 4, + "memoryGb": 8 + }, + "results": { + "normalize/raw-to-normalized": { + "name": "normalize/raw-to-normalized", + "hz": 347658.9432338144, + "medianNs": 2876.3822115384514, + "meanNs": 4154.171173878206, + "p75Ns": 5532.812499999987, + "p99Ns": 10917.57812500004, + "riqr": 1.0122014917577498, + "samples": 60, + "batchSize": 3328 + }, + "decode/cap67-transfer": { + "name": "decode/cap67-transfer", + "hz": 12578.73996403244, + "medianNs": 79499.21875000143, + "meanNs": 102292.74088541637, + "p75Ns": 101413.28125000016, + "p99Ns": 459601.5624999996, + "riqr": 0.3639874605685896, + "samples": 60, + "batchSize": 256 + }, + "watcher/fan-out-1": { + "name": "watcher/fan-out-1", + "hz": 18382910.12673317, + "medianNs": 54.398351137329435, + "meanNs": 65.37637239140467, + "p75Ns": 76.4842753507651, + "p99Ns": 144.2590415072292, + "riqr": 0.4489450111284567, + "samples": 60, + "batchSize": 602112 + }, + "watcher/fan-out-100": { + "name": "watcher/fan-out-100", + "hz": 411685.3735467131, + "medianNs": 2429.039417613733, + "meanNs": 2497.0904356060623, + "p75Ns": 2645.5965909090264, + "p99Ns": 4636.203835227287, + "riqr": 0.17215568315133162, + "samples": 60, + "batchSize": 11264 + }, + "watcher/fan-out-1000": { + "name": "watcher/fan-out-1000", + "hz": 39742.66623612245, + "medianNs": 25161.874999999156, + "meanNs": 27254.359374999964, + "p75Ns": 27585.468750000076, + "p99Ns": 55368.51562499975, + "riqr": 0.14847922203732455, + "samples": 60, + "batchSize": 1280 + }, + "cursor/memory-set": { + "name": "cursor/memory-set", + "hz": 2042467.6862916541, + "medianNs": 489.60382908951686, + "meanNs": 506.4624123906886, + "p75Ns": 668.1664737654427, + "p99Ns": 814.1625675154485, + "riqr": 0.6259785223727615, + "samples": 60, + "batchSize": 82944 + }, + "cursor/memory-set-many-100": { + "name": "cursor/memory-set-many-100", + "hz": 18270.575808615955, + "medianNs": 54732.81249999929, + "meanNs": 57850.833333332244, + "p75Ns": 59583.20312500121, + "p99Ns": 104006.24999999764, + "riqr": 0.18867224299874952, + "samples": 60, + "batchSize": 256 + } + } +} diff --git a/packages/pulse-core/bench/cases.ts b/packages/pulse-core/bench/cases.ts new file mode 100644 index 00000000..0ed26ee0 --- /dev/null +++ b/packages/pulse-core/bench/cases.ts @@ -0,0 +1,111 @@ +/** + * The benchmark cases required by issue #924. + * + * Four things sit on the hot path and are gated here: + * 1. raw event -> NormalizedEvent throughput (normalizeContractEvent) + * 2. decode-with-spec throughput (decodeUnifiedTransfer) + * 3. watcher fan-out at 1 / 100 / 1000 (Watcher.emit) + * 4. cursor write cost per adapter (CursorStore.set / setMany) + * + * Everything runs against the recorded CAP-67 corpus so the numbers reflect + * pulse-core's own work, never network time. Cases are built by `buildCases()` + * so fixtures are read once and shared across the whole run. + */ +import { normalizeContractEvent } from "../src/EventEngine.js"; +import { decodeUnifiedTransfer } from "../src/cap67/decodeTransfer.js"; +import { Watcher } from "../src/Watcher.js"; +import { MemoryCursorStore } from "../src/MemoryCursorStore.js"; +import type { NormalizedEvent } from "../src/index.js"; +import type { BenchCase } from "./harness.js"; +import { loadCap67Corpus, loadTransferFixtures, type RawSorobanFixtureEvent } from "./fixtures.js"; + +/** Fan-out widths mandated by the acceptance criteria. */ +const FANOUT_WIDTHS = [1, 100, 1000] as const; + +/** Batch size for the cursor setMany case, so the batch path is exercised, not a single write. */ +const CURSOR_BATCH = 100; + +/** + * Shapes a recorded fixture event into the argument `normalizeContractEvent` + * expects. The RPC record already carries the id, topic, and value fields, so + * this only supplies `pagingToken` (absent from the fixtures) and passes the + * rest through untouched. + */ +function toRpcEvent(event: RawSorobanFixtureEvent): Record { + return { ...event, pagingToken: event.id }; +} + +/** Builds every benchmark case. Fixtures are loaded once and closed over. */ +export function buildCases(): BenchCase[] { + const corpus = loadCap67Corpus(); + const transfers = loadTransferFixtures(); + const rpcEvents = corpus.map((f) => toRpcEvent(f.event)); + const transferInputs = transfers.map((f) => ({ topic: f.event.topic, value: f.event.value })); + + const cases: BenchCase[] = []; + + // 1. Raw event -> NormalizedEvent throughput. Cycles through the whole corpus + // so the mix of transfer/mint/burn/etc. is represented, not one shape. + let normIdx = 0; + cases.push({ + name: "normalize/raw-to-normalized", + fn: () => { + normalizeContractEvent(rpcEvents[normIdx++ % rpcEvents.length]!); + }, + }); + + // 2. Decode-with-spec throughput. Runs the CAP-67 transfer decoder over the + // transfer fixtures (bare i128 and SCMap-with-memo forms both included). + let decIdx = 0; + cases.push({ + name: "decode/cap67-transfer", + fn: () => { + decodeUnifiedTransfer(transferInputs[decIdx++ % transferInputs.length]!); + }, + }); + + // 3. Watcher fan-out at 1 / 100 / 1000 watchers. Each width is its own case so + // a regression localizes to a fan-out size. A no-op listener keeps the + // measurement on the emit/dispatch machinery, not user callback cost. + const sample = normalizeContractEvent(rpcEvents[0]!) as NormalizedEvent; + for (const width of FANOUT_WIDTHS) { + let watchers: Watcher[] = []; + cases.push({ + name: `watcher/fan-out-${width}`, + setup: () => { + watchers = Array.from({ length: width }, (_, i) => { + const w = new Watcher(`bench-${i}`); + w.on("*", () => {}); + return w; + }); + }, + fn: () => { + for (let i = 0; i < watchers.length; i++) watchers[i]!.emit("*", sample); + }, + }); + } + + // 4. Cursor write cost per adapter. MemoryCursorStore is the in-process + // adapter; single-key set and the batch setMany are measured separately + // because a store that batches I/O is judged on the batch path. Other + // adapters (File/Postgres/Redis/S3) slot in here without touching a + // network; they are left out of the default run to keep it hermetic. + // The store methods are async but resolve synchronously for the in-memory + // map, so the async harness path awaits each write honestly. + const memStore = new MemoryCursorStore(); + const batch: Record = {}; + for (let i = 0; i < CURSOR_BATCH; i++) batch[`stream-${i}`] = `cursor-${i}`; + let setIdx = 0; + cases.push({ + name: "cursor/memory-set", + async: true, + fn: () => memStore.set(`stream-${setIdx % 1000}`, `cursor-${setIdx++}`), + }); + cases.push({ + name: `cursor/memory-set-many-${CURSOR_BATCH}`, + async: true, + fn: () => memStore.setMany(batch), + }); + + return cases; +} diff --git a/packages/pulse-core/bench/fixtures.ts b/packages/pulse-core/bench/fixtures.ts new file mode 100644 index 00000000..b8994506 --- /dev/null +++ b/packages/pulse-core/bench/fixtures.ts @@ -0,0 +1,70 @@ +/** + * Loads the recorded CAP-67 fixture corpus. + * + * Each fixture is a full Soroban RPC `getEvents` JSON-RPC response with a single + * event under `result.events[0]`. Benchmarks run against these recorded records + * rather than a live network so the numbers are reproducible: a fixture-driven + * benchmark measures pulse-core, whereas one that hit testnet would measure the + * network. See the implementation note on issue #924. + */ +import { readFileSync, readdirSync } from "node:fs"; +import { fileURLToPath } from "node:url"; +import { dirname, join } from "node:path"; + +const here = dirname(fileURLToPath(import.meta.url)); + +/** Absolute path to the committed CAP-67 fixture directory. */ +export const CAP67_DIR = join(here, "..", "test", "fixtures", "cap67"); + +/** The raw event shape pulled out of a fixture's `result.events[0]`. */ +export interface RawSorobanFixtureEvent { + type: string; + ledger: number; + ledgerClosedAt: string; + contractId: string; + id: string; + txHash: string; + inSuccessfulContractCall: boolean; + topic: string[]; + value: string; + [key: string]: unknown; +} + +/** A loaded fixture: its file stem plus the unwrapped event. */ +export interface LoadedFixture { + /** File name without extension, e.g. "transfer_plain". */ + name: string; + /** The single event from `result.events[0]`. */ + event: RawSorobanFixtureEvent; +} + +/** Reads and unwraps every `*.json` fixture in the CAP-67 corpus, sorted by name. */ +export function loadCap67Corpus(): LoadedFixture[] { + const files = readdirSync(CAP67_DIR) + .filter((f) => f.endsWith(".json")) + .sort(); + + if (files.length === 0) { + throw new Error(`[bench] no CAP-67 fixtures found in ${CAP67_DIR}`); + } + + return files.map((file) => { + const raw = JSON.parse(readFileSync(join(CAP67_DIR, file), "utf8")) as { + result?: { events?: RawSorobanFixtureEvent[] }; + }; + const event = raw.result?.events?.[0]; + if (!event) { + throw new Error(`[bench] fixture ${file} has no result.events[0]`); + } + return { name: file.replace(/\.json$/, ""), event }; + }); +} + +/** The subset of the corpus that carries a CAP-67 `transfer` topic. */ +export function loadTransferFixtures(): LoadedFixture[] { + const transfers = loadCap67Corpus().filter((f) => f.name.startsWith("transfer")); + if (transfers.length === 0) { + throw new Error(`[bench] no transfer fixtures found in ${CAP67_DIR}`); + } + return transfers; +} diff --git a/packages/pulse-core/bench/gate.ts b/packages/pulse-core/bench/gate.ts new file mode 100644 index 00000000..14b833c4 --- /dev/null +++ b/packages/pulse-core/bench/gate.ts @@ -0,0 +1,135 @@ +/** + * Regression gate: compares a fresh benchmark run against the committed + * baseline and decides whether the build passes. + * + * A case fails when its throughput drops by more than {@link REGRESSION_THRESHOLD} + * relative to the baseline. Improvements never fail. Cases present in the run but + * missing from the baseline are reported as "new" (not a failure); cases in the + * baseline but missing from the run are reported as "missing" (a failure, because + * a silently dropped benchmark is how coverage rots). + */ +import type { BenchResult } from "./harness.js"; + +/** A regression worse than this fraction fails the build. 0.20 == 20%. */ +export const REGRESSION_THRESHOLD = 0.2; + +/** The on-disk shape of both the baseline file and a fresh run's output. */ +export interface BenchReport { + /** Schema version, so a future format change can be detected rather than silently misread. */ + schemaVersion: 1; + /** ISO 8601 timestamp of when the run was produced. */ + generatedAt: string; + /** The hardware and runtime the numbers were measured on. Recorded so a baseline is interpretable. */ + environment: BenchEnvironment; + /** Results keyed by case name for stable lookup regardless of run order. */ + results: Record; +} + +/** Machine context captured alongside results so a baseline means something. */ +export interface BenchEnvironment { + /** e.g. "linux x64". */ + platform: string; + /** Node version string, e.g. "v22.22.2". */ + node: string; + /** CPU model string, first core. */ + cpu: string; + /** Number of logical cores. */ + cores: number; + /** Total system memory in GB, rounded. */ + memoryGb: number; + /** Free-form note. The placeholder baseline sets this so a reader knows to regenerate it. */ + note?: string; +} + +/** One case's verdict after comparison. */ +export interface GateEntry { + name: string; + /** "ok" | "improved" | "regressed" | "new" | "missing". */ + status: "ok" | "improved" | "regressed" | "new" | "missing"; + /** Baseline ops/sec, or null when the case is new. */ + baselineHz: number | null; + /** Current ops/sec, or null when the case is missing from the run. */ + currentHz: number | null; + /** Signed relative change vs baseline (positive == faster), or null when not comparable. */ + delta: number | null; +} + +/** The full gate outcome. */ +export interface GateOutcome { + passed: boolean; + entries: GateEntry[]; +} + +/** + * Compares a run against a baseline and produces a pass/fail outcome plus a + * per-case breakdown. Pure and side-effect-free so it is trivially testable. + * + * A case is only marked "regressed" when it is slower than the baseline by more + * than `threshold`. The comparison is on throughput (`hz`), so higher is always + * better and the sign of `delta` reads naturally: negative means slower. + */ +export function gate( + current: BenchReport, + baseline: BenchReport, + threshold: number = REGRESSION_THRESHOLD, +): GateOutcome { + const entries: GateEntry[] = []; + const names = new Set([...Object.keys(baseline.results), ...Object.keys(current.results)]); + + for (const name of [...names].sort()) { + const base = baseline.results[name]; + const cur = current.results[name]; + + if (base && !cur) { + entries.push({ name, status: "missing", baselineHz: base.hz, currentHz: null, delta: null }); + continue; + } + if (!base && cur) { + entries.push({ name, status: "new", baselineHz: null, currentHz: cur.hz, delta: null }); + continue; + } + if (!base || !cur) continue; + + const delta = base.hz === 0 ? 0 : (cur.hz - base.hz) / base.hz; + + let status: GateEntry["status"]; + if (delta < -threshold) status = "regressed"; + else if (delta > threshold) status = "improved"; + else status = "ok"; + + entries.push({ name, status, baselineHz: base.hz, currentHz: cur.hz, delta }); + } + + const passed = !entries.some((e) => e.status === "regressed" || e.status === "missing"); + return { passed, entries }; +} + +/** Renders a gate outcome as an aligned text table for CI logs. */ +export function formatGate(outcome: GateOutcome, threshold: number = REGRESSION_THRESHOLD): string { + const lines: string[] = []; + const pct = (n: number | null): string => + n === null ? " -" : `${n >= 0 ? "+" : ""}${(n * 100).toFixed(1)}%`; + const hz = (n: number | null): string => (n === null ? "-" : `${(n / 1000).toFixed(1)}K`); + + const nameWidth = Math.max(4, ...outcome.entries.map((e) => e.name.length)); + const header = `${"case".padEnd(nameWidth)} ${"baseline".padStart(10)} ${"current".padStart(10)} ${"delta".padStart(8)} status`; + lines.push(header); + lines.push("-".repeat(header.length)); + + for (const e of outcome.entries) { + lines.push( + `${e.name.padEnd(nameWidth)} ${hz(e.baselineHz).padStart(10)} ${hz(e.currentHz).padStart(10)} ${pct(e.delta).padStart(8)} ${e.status}`, + ); + } + + lines.push(""); + lines.push( + outcome.passed + ? `PASS - no case regressed beyond ${(threshold * 100).toFixed(0)}%.` + : `FAIL - a case regressed beyond ${(threshold * 100).toFixed(0)}% or went missing.`, + ); + if (outcome.entries.some((e) => e.status === "new")) { + lines.push("Note: new cases are not gated until their numbers are committed to the baseline."); + } + return lines.join("\n"); +} diff --git a/packages/pulse-core/bench/harness.ts b/packages/pulse-core/bench/harness.ts new file mode 100644 index 00000000..7b12492b --- /dev/null +++ b/packages/pulse-core/bench/harness.ts @@ -0,0 +1,216 @@ +/** + * Minimal, dependency-free benchmark harness built on `node:perf_hooks`. + * + * The suite avoids a third-party benchmark runner on purpose: CI installs with + * `--frozen-lockfile`, so every case here has to run from the standard library + * alone, with no lockfile churn and no new supply-chain surface. + * + * Stability is the whole point of a gate. A benchmark that swings run-to-run + * produces false regressions, and a gate that cries wolf gets muted within a + * week. Three choices keep the noise down: + * - the headline figure is the **median** per-operation time, so a single GC + * pause or scheduler hiccup in one sample cannot drag the number; + * - each sample times a **calibrated batch** sized so the batch runs for a + * target duration, amortizing clock resolution across many calls; + * - the relative inter-quartile range travels with every result, so a caller + * (and the committed baseline) can see how trustworthy each number is. + */ +import { performance } from "node:perf_hooks"; + +/** A single thing to measure. `fn` runs once per iteration. */ +export interface BenchCase { + /** Stable identifier. This is the key the baseline is matched on, so it must not change casually. */ + name: string; + /** The operation under test. Sync by default; async cases are awaited each iteration (see `async`). */ + fn: () => void | Promise; + /** Optional one-time setup run before sampling; its cost is never counted. */ + setup?: () => void; + /** When true, each iteration is awaited. Use for cases whose real cost includes settling a promise. */ + async?: boolean; +} + +/** The measured result for one case. All times are in nanoseconds per operation. */ +export interface BenchResult { + name: string; + /** Operations per second, derived from the median. This is the headline number the gate reads. */ + hz: number; + /** Median time per operation, in nanoseconds. Robust to outlier samples. */ + medianNs: number; + /** Mean time per operation, in nanoseconds. Kept for context; not gated on. */ + meanNs: number; + /** 75th-percentile time per operation, in nanoseconds. */ + p75Ns: number; + /** 99th-percentile time per operation, in nanoseconds. */ + p99Ns: number; + /** Relative inter-quartile range ((p75-p25)/median). A spread indicator; smaller is steadier. */ + riqr: number; + /** How many timed samples were collected. */ + samples: number; + /** Iterations per timed sample actually used (after auto-calibration). */ + batchSize: number; +} + +/** Tuning knobs. Defaults keep a full run near a handful of seconds while staying steady. */ +export interface RunOptions { + /** Wall-clock budget, in ms, spent warming up before sampling. Lets the JIT settle. */ + warmupMs?: number; + /** Number of timed samples to collect. More samples tighten the median. */ + samples?: number; + /** Target time per sample, in ms. The batch is grown until one sample runs at least this long. */ + targetSampleMs?: number; + /** Minimum iterations per sample, regardless of calibration. Guards slow ops against tiny batches. */ + minBatchSize?: number; + /** Upper bound on batch auto-scaling, so a pathologically fast op can't spin forever. */ + maxBatchSize?: number; +} + +const DEFAULTS: Required = { + warmupMs: 300, + samples: 60, + targetSampleMs: 25, + minBatchSize: 256, + maxBatchSize: 5_000_000, +}; + +function quantile(sortedNs: number[], q: number): number { + if (sortedNs.length === 0) return 0; + const idx = Math.min(sortedNs.length - 1, Math.max(0, Math.floor(q * sortedNs.length))); + return sortedNs[idx]!; +} + +/** Reduces raw per-operation samples into the reported statistics. */ +function summarize(name: string, perOpNs: number[], batchSize: number): BenchResult { + const sorted = [...perOpNs].sort((a, b) => a - b); + const median = quantile(sorted, 0.5); + const p25 = quantile(sorted, 0.25); + const p75 = quantile(sorted, 0.75); + const p99 = quantile(sorted, 0.99); + const mean = perOpNs.reduce((a, b) => a + b, 0) / perOpNs.length; + const riqr = median === 0 ? 0 : (p75 - p25) / median; + return { + name, + hz: median === 0 ? 0 : 1_000_000_000 / median, + medianNs: median, + meanNs: mean, + p75Ns: p75, + p99Ns: p99, + riqr, + samples: perOpNs.length, + batchSize, + }; +} + +/** + * Grows a batch size until one batch takes at least `targetSampleMs`, so each + * timed sample spans enough work that clock resolution is negligible. Never + * returns fewer than `minBatchSize`, which keeps slow operations (tens of + * microseconds each) from calibrating down to a handful of noisy iterations. + */ +function calibrateBatch(fn: () => void | Promise, opts: Required): number { + let batch = opts.minBatchSize; + while (batch < opts.maxBatchSize) { + const start = performance.now(); + for (let i = 0; i < batch; i++) fn(); + const elapsed = performance.now() - start; + if (elapsed >= opts.targetSampleMs) break; + const factor = elapsed > 0 ? Math.max(2, Math.ceil(opts.targetSampleMs / elapsed)) : 2; + batch = Math.min(opts.maxBatchSize, batch * factor); + } + return batch; +} + +/** + * Async twin of {@link calibrateBatch}. Awaits each iteration so promise + * settling is part of the measured cost, which is what a real caller pays. + */ +async function calibrateBatchAsync( + fn: () => void | Promise, + opts: Required, +): Promise { + let batch = opts.minBatchSize; + while (batch < opts.maxBatchSize) { + const start = performance.now(); + for (let i = 0; i < batch; i++) await fn(); + const elapsed = performance.now() - start; + if (elapsed >= opts.targetSampleMs) break; + const factor = elapsed > 0 ? Math.max(2, Math.ceil(opts.targetSampleMs / elapsed)) : 2; + batch = Math.min(opts.maxBatchSize, batch * factor); + } + return batch; +} + +/** + * Runs one sync case and returns its statistics. The batch is calibrated to the + * operation's speed, the function is warmed for a wall-clock budget, then + * `samples` batches are timed. The median per-operation time becomes `hz`. + */ +export function runCase(bench: BenchCase, options: RunOptions = {}): BenchResult { + const opts = { ...DEFAULTS, ...options }; + bench.setup?.(); + + const batchSize = calibrateBatch(bench.fn, opts); + + const warmEnd = performance.now() + opts.warmupMs; + while (performance.now() < warmEnd) { + for (let i = 0; i < batchSize; i++) bench.fn(); + } + + const perOpNs: number[] = []; + for (let s = 0; s < opts.samples; s++) { + const start = performance.now(); + for (let i = 0; i < batchSize; i++) bench.fn(); + const elapsedMs = performance.now() - start; + perOpNs.push((elapsedMs * 1_000_000) / batchSize); + } + + return summarize(bench.name, perOpNs, batchSize); +} + +/** + * Runs one async case and returns its statistics. Identical in method to + * {@link runCase}, but awaits every iteration so a case backed by an async API + * (like a CursorStore) is measured settling its promise, not just scheduling it. + */ +export async function runCaseAsync( + bench: BenchCase, + options: RunOptions = {}, +): Promise { + const opts = { ...DEFAULTS, ...options }; + bench.setup?.(); + + const batchSize = await calibrateBatchAsync(bench.fn, opts); + + const warmEnd = performance.now() + opts.warmupMs; + while (performance.now() < warmEnd) { + for (let i = 0; i < batchSize; i++) await bench.fn(); + } + + const perOpNs: number[] = []; + for (let s = 0; s < opts.samples; s++) { + const start = performance.now(); + for (let i = 0; i < batchSize; i++) await bench.fn(); + const elapsedMs = performance.now() - start; + perOpNs.push((elapsedMs * 1_000_000) / batchSize); + } + + return summarize(bench.name, perOpNs, batchSize); +} + +/** Runs a list of cases in order and returns their results. */ +export async function runAll(cases: BenchCase[], options: RunOptions = {}): Promise { + const results: BenchResult[] = []; + for (const c of cases) { + process.stderr.write(` running ${c.name} ... `); + const result = c.async ? await runCaseAsync(c, options) : runCase(c, options); + process.stderr.write(`${formatHz(result.hz)} (iqr ${(result.riqr * 100).toFixed(1)}%)\n`); + results.push(result); + } + return results; +} + +/** Formats an ops/sec figure for human-readable console output. */ +export function formatHz(hz: number): string { + if (hz >= 1_000_000) return `${(hz / 1_000_000).toFixed(2)}M ops/s`; + if (hz >= 1_000) return `${(hz / 1_000).toFixed(2)}K ops/s`; + return `${hz.toFixed(0)} ops/s`; +} diff --git a/packages/pulse-core/bench/index.ts b/packages/pulse-core/bench/index.ts new file mode 100644 index 00000000..76838a56 --- /dev/null +++ b/packages/pulse-core/bench/index.ts @@ -0,0 +1,94 @@ +/** + * Benchmark entry point. + * + * tsx bench/index.ts run the suite and gate against bench/baseline.json + * tsx bench/index.ts --update run the suite and (re)write bench/baseline.json + * tsx bench/index.ts --json print the run report as JSON to stdout, no gating + * + * The gate exits non-zero when any case regresses beyond the threshold or a + * baselined case goes missing, which is what fails the CI job. Updating the + * baseline is deliberate and manual: a PR that moves it must justify the move in + * its body, per the process documented in docs/ARCHITECTURE.md. + */ +import { cpus, totalmem, platform, arch } from "node:os"; +import { readFileSync, writeFileSync } from "node:fs"; +import { fileURLToPath } from "node:url"; +import { dirname, join } from "node:path"; +import { runAll, type BenchResult } from "./harness.js"; +import { buildCases } from "./cases.js"; +import { + gate, + formatGate, + REGRESSION_THRESHOLD, + type BenchReport, + type BenchEnvironment, +} from "./gate.js"; + +const here = dirname(fileURLToPath(import.meta.url)); +const BASELINE_PATH = join(here, "baseline.json"); + +function captureEnvironment(): BenchEnvironment { + const cores = cpus(); + return { + platform: `${platform()} ${arch()}`, + node: process.version, + cpu: cores[0]?.model.trim() ?? "unknown", + cores: cores.length, + memoryGb: Math.round(totalmem() / 1024 ** 3), + }; +} + +function toReport(results: BenchResult[]): BenchReport { + return { + schemaVersion: 1, + generatedAt: new Date().toISOString(), + environment: captureEnvironment(), + results: Object.fromEntries(results.map((r) => [r.name, r])), + }; +} + +function readBaseline(): BenchReport | null { + try { + return JSON.parse(readFileSync(BASELINE_PATH, "utf8")) as BenchReport; + } catch { + return null; + } +} + +async function main(): Promise { + const args = new Set(process.argv.slice(2)); + const update = args.has("--update"); + const jsonOnly = args.has("--json"); + + process.stderr.write("pulse-core benchmarks\n"); + const results = await runAll(buildCases()); + const report = toReport(results); + + if (jsonOnly) { + process.stdout.write(JSON.stringify(report, null, 2) + "\n"); + return; + } + + if (update) { + writeFileSync(BASELINE_PATH, JSON.stringify(report, null, 2) + "\n"); + process.stderr.write(`\nBaseline written to ${BASELINE_PATH}\n`); + process.stderr.write( + "Remember: a committed baseline change must be justified in the PR body.\n", + ); + return; + } + + const baseline = readBaseline(); + if (!baseline) { + process.stderr.write( + "\nNo baseline found. Seed one with:\n pnpm --filter @orbital-stellar/pulse-core bench:update\n", + ); + process.exit(1); + } + + const outcome = gate(report, baseline, REGRESSION_THRESHOLD); + process.stderr.write("\n" + formatGate(outcome, REGRESSION_THRESHOLD) + "\n"); + process.exit(outcome.passed ? 0 : 1); +} + +main(); diff --git a/packages/pulse-core/bench/soroban-throughput.ts b/packages/pulse-core/bench/soroban-throughput.ts deleted file mode 100644 index 5b8c8a05..00000000 --- a/packages/pulse-core/bench/soroban-throughput.ts +++ /dev/null @@ -1,215 +0,0 @@ -import { EventEngine } from "../src/EventEngine.js"; - -type EngineBenchInternals = { - normalize: (record: unknown) => unknown; - route: (event: unknown) => void; -}; - -type BenchmarkResult = { - subscriptions: number; - responses: number; - eventsPerResponse: number; - rpcEvents: number; - routedEvents: number; - durationMs: number; - eventsPerSecond: number; - memory: { - baselineHeapMb: number; - subscribedHeapMb: number; - postReplayHeapMb: number; - postReplayRssMb: number; - }; -}; - -type SyntheticRpcEvent = { - type: "contract_event"; - id: string; - pagingToken: string; - contract_id: string; - topics: string[]; - data: Record; - created_at: string; -}; - -type SyntheticGetEventsResponse = { - cursor: string; - events: SyntheticRpcEvent[]; -}; - -const SUBSCRIPTION_COUNTS = [1000, 5000, 10000] as const; -const DEFAULT_RESPONSE_COUNT = 10; -const DEFAULT_EVENTS_PER_RESPONSE = 100; - -function toMb(bytes: number): number { - return Number((bytes / (1024 * 1024)).toFixed(2)); -} - -function forceGc(): void { - if (typeof global.gc === "function") { - global.gc(); - } -} - -function makeContractId(index: number): string { - // Contract IDs are routing keys in this benchmark; keep the shape deterministic. - return `C${String(index).padStart(55, "0")}`; -} - -function makeSyntheticGetEventsResponse( - responseIndex: number, - eventsPerResponse: number, - subscriptionCount: number, -): SyntheticGetEventsResponse { - const events: SyntheticRpcEvent[] = []; - const baseIndex = responseIndex * eventsPerResponse; - - for (let i = 0; i < eventsPerResponse; i += 1) { - const eventIndex = baseIndex + i; - const contractIndex = eventIndex % subscriptionCount; - const pagingToken = String(eventIndex + 1).padStart(12, "0"); - - events.push({ - type: "contract_event", - id: `evt-${pagingToken}`, - pagingToken, - contract_id: makeContractId(contractIndex), - topics: ["transfer", `account-${contractIndex}`], - data: { amount: "1.0000000" }, - created_at: "2026-01-01T00:00:00.000Z", - }); - } - - return { - cursor: events.at(-1)?.pagingToken ?? String(baseIndex).padStart(12, "0"), - events, - }; -} - -function subscribeContractWatchers(engine: EventEngine, subscriptionCount: number): void { - for (let i = 0; i < subscriptionCount; i += 1) { - const watcher = engine.subscribeContract(`contract-sub-${i}`, { - filters: [{ contractIds: [makeContractId(i)] }], - }); - watcher.on("*", () => { - // Intentionally empty: includes EventEmitter dispatch cost in throughput figures. - }); - } -} - -function runScenario( - subscriptionCount: number, - responseCount: number, - eventsPerResponse: number, -): BenchmarkResult { - forceGc(); - const baselineHeapMb = toMb(process.memoryUsage().heapUsed); - - const engine = new EventEngine({ network: "testnet" }); - const internals = engine as unknown as EngineBenchInternals; - - subscribeContractWatchers(engine, subscriptionCount); - forceGc(); - const subscribedHeapMb = toMb(process.memoryUsage().heapUsed); - - const start = process.hrtime.bigint(); - - for (let responseIndex = 0; responseIndex < responseCount; responseIndex += 1) { - const response = makeSyntheticGetEventsResponse( - responseIndex, - eventsPerResponse, - subscriptionCount, - ); - - for (const rpcEvent of response.events) { - const normalized = internals.normalize(rpcEvent); - if (!normalized) { - continue; - } - - internals.route(normalized); - } - } - - const durationMs = Number(process.hrtime.bigint() - start) / 1_000_000; - - forceGc(); - const postReplayMemory = process.memoryUsage(); - - engine.stop(); - - // Each matching contract subscription receives the typed event and '*'. - const rpcEvents = responseCount * eventsPerResponse; - const routedEvents = rpcEvents * 2; - const eventsPerSecond = Number((routedEvents / (durationMs / 1000)).toFixed(2)); - - return { - subscriptions: subscriptionCount, - responses: responseCount, - eventsPerResponse, - rpcEvents, - routedEvents, - durationMs: Number(durationMs.toFixed(2)), - eventsPerSecond, - memory: { - baselineHeapMb, - subscribedHeapMb, - postReplayHeapMb: toMb(postReplayMemory.heapUsed), - postReplayRssMb: toMb(postReplayMemory.rss), - }, - }; -} - -function getPositiveIntegerArg(name: string, defaultValue: number): number { - const arg = process.argv.find((value) => value.startsWith(`--${name}=`)); - if (!arg) return defaultValue; - - const rawValue = arg.split("=")[1]; - const parsed = Number.parseInt(rawValue, 10); - - if (!Number.isFinite(parsed) || parsed <= 0) { - throw new Error(`Invalid --${name} value: ${rawValue}`); - } - - return parsed; -} - -function main(): void { - const responses = getPositiveIntegerArg("responses", DEFAULT_RESPONSE_COUNT); - const eventsPerResponse = getPositiveIntegerArg( - "events-per-response", - DEFAULT_EVENTS_PER_RESPONSE, - ); - - console.log("pulse-core Soroban throughput benchmark"); - console.log(`node=${process.version}`); - console.log(`responses_per_scenario=${responses}`); - console.log(`events_per_response=${eventsPerResponse}`); - if (typeof global.gc !== "function") { - console.log("gc=unavailable (run with --expose-gc for tighter memory numbers)"); - } - console.log(""); - - const results = SUBSCRIPTION_COUNTS.map((subscriptions) => - runScenario(subscriptions, responses, eventsPerResponse), - ); - - console.table( - results.map((result) => ({ - subscriptions: result.subscriptions, - responses: result.responses, - rpc_events: result.rpcEvents, - routed_events: result.routedEvents, - duration_ms: result.durationMs, - events_per_sec: result.eventsPerSecond, - baseline_heap_mb: result.memory.baselineHeapMb, - subscribed_heap_mb: result.memory.subscribedHeapMb, - post_replay_heap_mb: result.memory.postReplayHeapMb, - post_replay_rss_mb: result.memory.postReplayRssMb, - })), - ); - - console.log("\nJSON results:"); - console.log(JSON.stringify(results, null, 2)); -} - -main(); diff --git a/packages/pulse-core/bench/throughput.ts b/packages/pulse-core/bench/throughput.ts deleted file mode 100644 index 5e938115..00000000 --- a/packages/pulse-core/bench/throughput.ts +++ /dev/null @@ -1,157 +0,0 @@ -import { EventEngine } from "../src/index.js"; - -type EngineBenchInternals = EventEngine & { - normalize: (record: unknown) => unknown; - route: (event: unknown) => void; -}; - -type BenchmarkResult = { - watchers: number; - records: number; - routedEvents: number; - durationMs: number; - eventsPerSecond: number; - memory: { - baselineHeapMb: number; - subscribedHeapMb: number; - postReplayHeapMb: number; - postReplayRssMb: number; - }; -}; - -const WATCHER_COUNTS = [1000, 5000, 10000] as const; -const DEFAULT_RECORD_COUNT = 100_000; - -function toMb(bytes: number): number { - return Number((bytes / (1024 * 1024)).toFixed(2)); -} - -function forceGc(): void { - if (typeof global.gc === "function") { - global.gc(); - } -} - -function makeAddress(index: number): string { - // Routing keys are string lookups; this deterministic shape keeps the benchmark stable. - return `G${String(index).padStart(55, "0")}`; -} - -function makeSyntheticPaymentRecord(index: number, watcherCount: number): Record { - const toIndex = index % watcherCount; - const fromIndex = (index + 1) % watcherCount; - - return { - type: "payment", - to: makeAddress(toIndex), - from: makeAddress(fromIndex), - amount: "1.0000000", - asset_type: "native", - created_at: "2026-01-01T00:00:00.000Z", - id: `${index}`, - }; -} - -function subscribeWatchers(engine: EventEngine, watcherCount: number): void { - for (let i = 0; i < watcherCount; i += 1) { - const watcher = engine.subscribe(makeAddress(i)); - watcher.on("*", () => { - // Intentionally empty: includes EventEmitter dispatch cost in throughput figures. - }); - } -} - -function runScenario(watcherCount: number, recordCount: number): BenchmarkResult { - forceGc(); - const baselineHeapMb = toMb(process.memoryUsage().heapUsed); - - const engine = new EventEngine({ network: "testnet" }); - const internals = engine as EngineBenchInternals; - - subscribeWatchers(engine, watcherCount); - forceGc(); - const subscribedHeapMb = toMb(process.memoryUsage().heapUsed); - - const start = process.hrtime.bigint(); - - for (let i = 0; i < recordCount; i += 1) { - const normalized = internals.normalize(makeSyntheticPaymentRecord(i, watcherCount)); - if (!normalized) { - continue; - } - - internals.route(normalized); - } - - const durationMs = Number(process.hrtime.bigint() - start) / 1_000_000; - - forceGc(); - const postReplayMemory = process.memoryUsage(); - - engine.stop(); - - // Each payment routes to both "to" and "from" watcher; each watcher gets the typed event and '*'. - const routedEvents = recordCount * 4; - const eventsPerSecond = Number((routedEvents / (durationMs / 1000)).toFixed(2)); - - return { - watchers: watcherCount, - records: recordCount, - routedEvents, - durationMs: Number(durationMs.toFixed(2)), - eventsPerSecond, - memory: { - baselineHeapMb, - subscribedHeapMb, - postReplayHeapMb: toMb(postReplayMemory.heapUsed), - postReplayRssMb: toMb(postReplayMemory.rss), - }, - }; -} - -function getRecordCountFromArgs(): number { - const arg = process.argv.find((value) => value.startsWith("--records=")); - if (!arg) return DEFAULT_RECORD_COUNT; - - const rawValue = arg.split("=")[1]; - const parsed = Number.parseInt(rawValue, 10); - - if (!Number.isFinite(parsed) || parsed <= 0) { - throw new Error(`Invalid --records value: ${rawValue}`); - } - - return parsed; -} - -function main(): void { - const records = getRecordCountFromArgs(); - - console.log("pulse-core throughput benchmark"); - console.log(`node=${process.version}`); - console.log(`records_per_scenario=${records}`); - if (typeof global.gc !== "function") { - console.log("gc=unavailable (run with --expose-gc for tighter memory numbers)"); - } - console.log(""); - - const results = WATCHER_COUNTS.map((watchers) => runScenario(watchers, records)); - - console.table( - results.map((result) => ({ - watchers: result.watchers, - records: result.records, - routed_events: result.routedEvents, - duration_ms: result.durationMs, - events_per_sec: result.eventsPerSecond, - baseline_heap_mb: result.memory.baselineHeapMb, - subscribed_heap_mb: result.memory.subscribedHeapMb, - post_replay_heap_mb: result.memory.postReplayHeapMb, - post_replay_rss_mb: result.memory.postReplayRssMb, - })), - ); - - console.log("\nJSON results:"); - console.log(JSON.stringify(results, null, 2)); -} - -main(); diff --git a/packages/pulse-core/package.json b/packages/pulse-core/package.json index 3e822cb2..25c682a5 100644 --- a/packages/pulse-core/package.json +++ b/packages/pulse-core/package.json @@ -34,8 +34,10 @@ "test:coverage": "vitest run --coverage", "test:integration": "INTEGRATION_TESTS=true vitest run test/integration", "typecheck": "tsc --noEmit -p tsconfig.json", - "bench:throughput": "tsx bench/throughput.ts", - "bench:soroban-throughput": "tsx bench/soroban-throughput.ts" + "bench": "tsx bench/index.ts", + "bench:update": "tsx bench/index.ts --update", + "bench:ci": "tsx bench/index.ts", + "bench:json": "tsx bench/index.ts --json" }, "dependencies": { "@orbital-stellar/abi-registry": "workspace:*",