perf(pulse-core): benchmark suite with 20% regression gating (#924)

.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

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.

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
+    }
+  }
+}

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<string, unknown> {
+  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<string, string> = {};
+  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;
+}

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;
+}