From 492d2985e2e29517e739f845f32b91069b9447d2 Mon Sep 17 00:00:00 2001 From: Eric Allam Date: Thu, 20 Aug 2026 20:22:48 +0100 Subject: [PATCH 1/5] test: add CPU benchmarks for the engine-facing request and run-engine paths Two on-demand benchmarks, neither in the default suite: - apps/webapp: spawns a real webapp with --inspect against throwaway containers, seeds a production environment with a promoted managed deployment, and drives a closed-loop supervisor pool through the worker-action lifecycle. Profiles over CDP so the profile covers only the measured window, and samples event-loop utilization inside the webapp. - internal-packages/run-engine: drives RunEngine directly, profiling the enqueue and lifecycle phases separately so engine cost is not mixed with request-stack overhead. Plus a dependency-free .cpuprofile analyzer that symbolicates through the build's source maps and ranks CPU by package, by self time and by total time. startWebapp gains overrideEnv, applied after the worker-disable defaults, so the HTTP bench can re-enable the run engine worker that drains the master queue into the worker queues a supervisor dequeues from. --- .gitignore | 3 + apps/webapp/package.json | 3 +- apps/webapp/test/bench/README.md | 91 +++++ apps/webapp/test/bench/analyzeProfile.ts | 59 ++++ .../test/bench/engineHttp.bench.test.ts | 288 ++++++++++++++++ apps/webapp/test/bench/lib/cdp.ts | 227 +++++++++++++ apps/webapp/test/bench/lib/engineFixtures.ts | 199 +++++++++++ apps/webapp/test/bench/lib/loadDriver.ts | 149 ++++++++ apps/webapp/test/bench/lib/profileAnalysis.ts | 319 ++++++++++++++++++ apps/webapp/test/bench/lib/sourcemap.ts | 159 +++++++++ apps/webapp/vitest.bench.config.ts | 22 ++ apps/webapp/vitest.config.ts | 7 +- internal-packages/run-engine/package.json | 3 +- .../src/engine/bench/inspectorProfiler.ts | 130 +++++++ .../bench/runEngineLifecycle.bench.test.ts | 313 +++++++++++++++++ .../run-engine/vitest.bench.config.ts | 17 + internal-packages/run-engine/vitest.config.ts | 1 + .../testcontainers/src/webapp.ts | 12 + knip.json | 1 + 19 files changed, 2000 insertions(+), 3 deletions(-) create mode 100644 apps/webapp/test/bench/README.md create mode 100644 apps/webapp/test/bench/analyzeProfile.ts create mode 100644 apps/webapp/test/bench/engineHttp.bench.test.ts create mode 100644 apps/webapp/test/bench/lib/cdp.ts create mode 100644 apps/webapp/test/bench/lib/engineFixtures.ts create mode 100644 apps/webapp/test/bench/lib/loadDriver.ts create mode 100644 apps/webapp/test/bench/lib/profileAnalysis.ts create mode 100644 apps/webapp/test/bench/lib/sourcemap.ts create mode 100644 apps/webapp/vitest.bench.config.ts create mode 100644 internal-packages/run-engine/src/engine/bench/inspectorProfiler.ts create mode 100644 internal-packages/run-engine/src/engine/bench/runEngineLifecycle.bench.test.ts create mode 100644 internal-packages/run-engine/vitest.bench.config.ts diff --git a/.gitignore b/.gitignore index b11dded2b02..3f49db35ff8 100644 --- a/.gitignore +++ b/.gitignore @@ -87,3 +87,6 @@ ailogger-output.log observability-map.json .claude/worktrees/ + +# CPU benchmark artifacts (profiles + summaries) +.bench/ diff --git a/apps/webapp/package.json b/apps/webapp/package.json index cabd55d1d7b..788342f23bb 100644 --- a/apps/webapp/package.json +++ b/apps/webapp/package.json @@ -25,7 +25,8 @@ "upload:sourcemaps": "bash ./upload-sourcemaps.sh", "test": "vitest --no-file-parallelism", "test:perf": "vitest --config ./vitest.perf.config.ts --run", - "eval:dev": "evalite watch" + "eval:dev": "evalite watch", + "test:bench": "vitest --config ./vitest.bench.config.ts --run" }, "dependencies": { "@ai-sdk/openai": "^3.0.0", diff --git a/apps/webapp/test/bench/README.md b/apps/webapp/test/bench/README.md new file mode 100644 index 00000000000..ecd2231bf85 --- /dev/null +++ b/apps/webapp/test/bench/README.md @@ -0,0 +1,91 @@ +# Engine CPU benchmarks + +Two benchmarks for the paths the production engine service spends its CPU in, plus a +`.cpuprofile` analyzer. Neither runs in CI: they take minutes, attach the V8 profiler, and +report numbers rather than assert on them. + +| bench | what it covers | where | +| --- | --- | --- | +| `engineHttp.bench.test.ts` | the full request stack for `engine/v1/worker-actions/*` | `apps/webapp` | +| `runEngineLifecycle.bench.test.ts` | run-engine and run-queue with no HTTP in the way | `internal-packages/run-engine` | + +Artifacts (profiles + JSON summaries) land in `.bench/` at the repo root, which is gitignored. + +## HTTP bench + +Measures what a managed supervisor actually does: dequeue, start attempt, heartbeat, +read latest snapshot, complete attempt. Needs a built webapp. + +```bash +pnpm run build --filter webapp +cd apps/webapp +pnpm run test:bench +``` + +It spawns a real webapp against throwaway Postgres and Redis containers, seeds a production +environment with a promoted managed deployment, fills the worker queue over the public +trigger API, then drives a closed-loop supervisor pool for the measured window. + +The webapp is spawned with `--inspect` and profiled over CDP, so the profile covers only the +measured window rather than boot. Event-loop utilization is sampled **inside** the webapp +process over the same connection. + +Knobs: + +| var | default | meaning | +| --- | --- | --- | +| `BENCH_RUNS` | 1200 | runs queued before the window opens | +| `BENCH_SUPERVISORS` | 16 | concurrent virtual supervisors | +| `BENCH_HEARTBEATS` | 2 | heartbeats per run | +| `BENCH_DURATION_MS` | 60000 | measured window | +| `BENCH_SAMPLING_INTERVAL_US` | 200 | V8 sampling interval | +| `BENCH_PROFILE_NAME` | `engine-http` | artifact basename | +| `BENCH_EXTRA_ENV` | — | JSON merged into the webapp's env | +| `BENCH_OUT_DIR` | `/.bench` | artifact directory | + +`BENCH_EXTRA_ENV` plus `BENCH_PROFILE_NAME` is how you A/B a single flag: + +```bash +BENCH_RUNS=5000 BENCH_SUPERVISORS=24 BENCH_DURATION_MS=90000 \ + BENCH_PROFILE_NAME=engine-http-no-elm \ + BENCH_EXTRA_ENV='{"EVENT_LOOP_MONITOR_ENABLED":"0"}' \ + pnpm run test:bench +``` + +Run the same size for both arms and compare `on-cpu ms per completed run` rather than +throughput: throughput on a laptop moves ~5% run to run, on-CPU per unit of work is far +steadier. + +## Run-engine bench + +No HTTP, no webapp: drives `RunEngine` directly so engine and queue costs are not mixed with +request-stack overhead. Profiles two phases separately, because blending them hides which one +owns a hot frame. + +```bash +cd internal-packages/run-engine +pnpm run test:bench +``` + +Knobs: `BENCH_RUNS`, `BENCH_CONSUMERS`, `BENCH_HEARTBEATS`, `BENCH_CONCURRENCY_LIMIT`, +`BENCH_SAMPLING_INTERVAL_US`, `BENCH_OUT_DIR`. + +The driver shares a process with the code under measurement, so its own cost is in the +profile. It is a thin await loop and appears under its own frames rather than smeared across +engine frames. + +## Analyzing a profile + +```bash +pnpm --filter webapp exec tsx test/bench/analyzeProfile.ts .bench/engine-http.cpuprofile --top 30 +``` + +Three views: CPU by bucket (which package owns the cycles), hottest frames by self time (what +to go fix), and hottest frames by total time (entry points, and a check that the load +exercised the route mix you intended). Frames are symbolicated through the build's source +maps, so bundled chunks report as the source files they came from. + +Percentages are shares of **on-CPU** time, with V8's `(idle)` and `(program)` excluded. A +share of wall clock would make everything look cheap whenever the bench was IO-bound. + +`--json ` writes the full analysis for diffing two runs. diff --git a/apps/webapp/test/bench/analyzeProfile.ts b/apps/webapp/test/bench/analyzeProfile.ts new file mode 100644 index 00000000000..8b1d0fdf428 --- /dev/null +++ b/apps/webapp/test/bench/analyzeProfile.ts @@ -0,0 +1,59 @@ +#!/usr/bin/env tsx +/** + * Ranks where a `.cpuprofile` spent its cycles. + * + * pnpm --filter webapp exec tsx test/bench/analyzeProfile.ts [--top 40] [--json out.json] + * + * `--root` overrides the repo root used to make source paths relative and to + * find the build's source maps; it defaults to the repo containing this file. + */ +import { readFileSync, writeFileSync } from "node:fs"; +import { resolve } from "node:path"; +import { analyzeProfile, formatAnalysis, type CpuProfile } from "./lib/profileAnalysis"; + +function parseArgs(argv: string[]): { + profilePath?: string; + top: number; + json?: string; + root: string; +} { + const here = typeof __dirname === "string" ? __dirname : import.meta.dirname; + + const defaults = { + top: 30, + root: resolve(here, "..", "..", "..", ".."), + }; + + let profilePath: string | undefined; + let top = defaults.top; + let json: string | undefined; + let root = defaults.root; + + for (let i = 0; i < argv.length; i++) { + const arg = argv[i]!; + if (arg === "--top") top = Number(argv[++i]); + else if (arg === "--json") json = argv[++i]; + else if (arg === "--root") root = resolve(argv[++i]!); + else if (!arg.startsWith("--")) profilePath = arg; + } + + return { profilePath, top, json, root }; +} + +const { profilePath, top, json, root } = parseArgs(process.argv.slice(2)); + +if (!profilePath) { + console.error("usage: analyzeProfile.ts [--top N] [--json out.json]"); + process.exit(1); +} + +const profile = JSON.parse(readFileSync(profilePath, "utf8")) as CpuProfile; +const analysis = analyzeProfile(profile, root); + +console.log(`\n=== ${profilePath} ===`); +console.log(formatAnalysis(analysis, top)); + +if (json) { + writeFileSync(json, JSON.stringify(analysis, null, 2)); + console.log(`\nwrote ${json}`); +} diff --git a/apps/webapp/test/bench/engineHttp.bench.test.ts b/apps/webapp/test/bench/engineHttp.bench.test.ts new file mode 100644 index 00000000000..d8f1e8deed5 --- /dev/null +++ b/apps/webapp/test/bench/engineHttp.bench.test.ts @@ -0,0 +1,288 @@ +/** + * CPU benchmark for the engine-facing HTTP surface: the + * `engine/v1/worker-actions/*` routes a managed supervisor calls. + * + * Unlike the run-engine bench, this measures the whole request stack — Remix + * routing, `createActionWorkerApiRoute`, worker-token auth, zod validation, + * JSON encode/decode — on top of the engine work, which is where a large share + * of the production engine service's event-loop time actually goes. + * + * The webapp runs as a child process with `--inspect`, and the profiler is + * driven over CDP so the profile covers only the measured window rather than + * boot. Artifacts land in `.bench/` at the repo root. Analyze one with: + * + * pnpm --filter webapp exec tsx test/bench/analyzeProfile.ts + * + * Knobs, all optional: + * BENCH_RUNS, BENCH_SUPERVISORS, BENCH_HEARTBEATS, BENCH_DURATION_MS, + * BENCH_OUT_DIR, BENCH_SAMPLING_INTERVAL_US + */ +import { startTestServer, type TestServer } from "@internal/testcontainers/webapp"; +import { createServer } from "node:net"; +import { mkdir, writeFile } from "node:fs/promises"; +import { join } from "node:path"; +import { afterAll, beforeAll, describe, expect, it, vi } from "vitest"; +import { WebappProfiler } from "./lib/cdp"; +import { seedEngineFixtures, workerHeaders, type EngineFixtures } from "./lib/engineFixtures"; +import { formatStatsTable, LatencyRecorder, runLoad } from "./lib/loadDriver"; + +vi.setConfig({ testTimeout: 900_000 }); + +const RUNS = Number(process.env.BENCH_RUNS ?? 1200); +const SUPERVISORS = Number(process.env.BENCH_SUPERVISORS ?? 16); +const HEARTBEATS_PER_RUN = Number(process.env.BENCH_HEARTBEATS ?? 2); +const DURATION_MS = Number(process.env.BENCH_DURATION_MS ?? 60_000); +const SAMPLING_INTERVAL_US = Number(process.env.BENCH_SAMPLING_INTERVAL_US ?? 200); +const OUT_DIR = process.env.BENCH_OUT_DIR ?? join(process.cwd(), "..", "..", ".bench"); +const PROFILE_NAME = process.env.BENCH_PROFILE_NAME ?? "engine-http"; + +/** + * JSON object merged into the spawned webapp's environment, for A/B runs + * against a single flag, e.g. + * `BENCH_EXTRA_ENV='{"EVENT_LOOP_MONITOR_ENABLED":"0"}'`. + */ +const EXTRA_WEBAPP_ENV: Record = process.env.BENCH_EXTRA_ENV + ? JSON.parse(process.env.BENCH_EXTRA_ENV) + : {}; + +async function findFreePort(): Promise { + return new Promise((resolve, reject) => { + const server = createServer(); + server.listen(0, () => { + const { port } = server.address() as { port: number }; + server.close((err) => (err ? reject(err) : resolve(port))); + }); + }); +} + +let server: TestServer; +let profiler: WebappProfiler; +let fixtures: EngineFixtures; +let inspectPort: number; + +beforeAll(async () => { + inspectPort = await findFreePort(); + server = await startTestServer({ + extraEnv: { + NODE_OPTIONS: `--inspect=${inspectPort}`, + API_RATE_LIMIT_MAX: "1000000", + API_RATE_LIMIT_REFILL_RATE: "1000000", + ...EXTRA_WEBAPP_ENV, + }, + overrideEnv: { RUN_ENGINE_WORKER_ENABLED: "1" }, + }); + profiler = await WebappProfiler.attach(inspectPort); +}, 300_000); + +afterAll(async () => { + profiler?.detach(); + await server?.stop(); +}, 120_000); + +/** + * Seeds the worker queue. Failures are counted by status and surfaced rather + * than swallowed: a partially-filled queue silently turns the measured window + * into mostly-empty dequeues, which reads as a fast server rather than a + * broken fixture. + */ +async function triggerRuns(count: number): Promise { + let triggered = 0; + const failures = new Map(); + + const batches = Math.ceil(count / 50); + for (let batch = 0; batch < batches; batch++) { + const size = Math.min(50, count - batch * 50); + const responses = await Promise.all( + Array.from({ length: size }, (_, i) => { + const index = batch * 50 + i; + const taskId = fixtures.taskIdentifiers[index % fixtures.taskIdentifiers.length]!; + return server.webapp.fetch(`/api/v1/tasks/${taskId}/trigger`, { + method: "POST", + headers: { + Authorization: `Bearer ${fixtures.environmentApiKey}`, + "content-type": "application/json", + }, + body: JSON.stringify({ payload: { index, message: "bench payload" } }), + }); + }) + ); + + for (const res of responses) { + if (res.ok) { + triggered += 1; + continue; + } + const existing = failures.get(res.status); + if (existing) existing.count += 1; + else failures.set(res.status, { count: 1, sample: (await res.text()).slice(0, 200) }); + } + } + + if (failures.size > 0) { + const detail = [...failures.entries()] + .map(([status, { count: n, sample }]) => `${status} x${n} (${sample})`) + .join("; "); + console.warn(`[engine-http] ${count - triggered}/${count} triggers failed: ${detail}`); + } + + if (triggered < count * 0.9) { + throw new Error(`only ${triggered}/${count} runs were queued; the measured window would idle`); + } + + return triggered; +} + +describe("engine worker-action HTTP CPU benchmark", () => { + it("profiles the supervisor request loop", async () => { + await mkdir(OUT_DIR, { recursive: true }); + + fixtures = await seedEngineFixtures(server.prisma, { taskCount: 4 }); + + const triggered = await triggerRuns(RUNS); + expect(triggered).toBeGreaterThan(0); + + const recorder = new LatencyRecorder(); + let dequeuedRuns = 0; + let completedRuns = 0; + let emptyDequeues = 0; + + await profiler.startCpuProfile(SAMPLING_INTERVAL_US); + profiler.startEluSampling(); + recorder.begin(); + + await runLoad({ + concurrency: SUPERVISORS, + durationMs: DURATION_MS, + iteration: async (workerIndex) => { + const headers = workerHeaders(fixtures, `bench-instance-${workerIndex}`); + + const dequeueResponse = await recorder.time("POST worker-actions/dequeue", async () => { + const res = await server.webapp.fetch("/engine/v1/worker-actions/dequeue", { + method: "POST", + headers, + body: JSON.stringify({}), + }); + if (!res.ok) throw new Error(`dequeue ${res.status}`); + return (await res.json()) as Array<{ + run: { friendlyId: string }; + snapshot: { friendlyId: string }; + }>; + }); + + const message = dequeueResponse?.[0]; + if (!message) { + emptyDequeues += 1; + await new Promise((r) => setTimeout(r, 25)); + return; + } + + dequeuedRuns += 1; + const runId = message.run.friendlyId; + let snapshotId = message.snapshot.friendlyId; + + const attempt = await recorder.time("POST attempts/start", async () => { + const res = await server.webapp.fetch( + `/engine/v1/worker-actions/runs/${runId}/snapshots/${snapshotId}/attempts/start`, + { method: "POST", headers, body: JSON.stringify({}) } + ); + if (!res.ok) throw new Error(`start ${res.status}`); + return (await res.json()) as { snapshot: { friendlyId: string } }; + }); + + if (!attempt) return; + snapshotId = attempt.snapshot.friendlyId; + + for (let beat = 0; beat < HEARTBEATS_PER_RUN; beat++) { + await recorder.time("POST snapshots/heartbeat", async () => { + const res = await server.webapp.fetch( + `/engine/v1/worker-actions/runs/${runId}/snapshots/${snapshotId}/heartbeat`, + { method: "POST", headers, body: JSON.stringify({}) } + ); + if (!res.ok) throw new Error(`heartbeat ${res.status}`); + return res.json(); + }); + } + + await recorder.time("GET snapshots/latest", async () => { + const res = await server.webapp.fetch( + `/engine/v1/worker-actions/runs/${runId}/snapshots/latest`, + { headers } + ); + if (!res.ok) throw new Error(`latest ${res.status}`); + return res.json(); + }); + + const completed = await recorder.time("POST attempts/complete", async () => { + const res = await server.webapp.fetch( + `/engine/v1/worker-actions/runs/${runId}/snapshots/${snapshotId}/attempts/complete`, + { + method: "POST", + headers, + body: JSON.stringify({ + completion: { + ok: true, + id: runId, + outputType: "application/json", + output: JSON.stringify({ done: true }), + }, + }), + } + ); + if (!res.ok) throw new Error(`complete ${res.status}`); + return res.json(); + }); + + if (completed) completedRuns += 1; + }, + }); + + recorder.end(); + const elu = profiler.stopEluSampling(); + const profile = await profiler.stopCpuProfile(join(OUT_DIR, `${PROFILE_NAME}.cpuprofile`)); + + const stats = recorder.stats(); + const totals = recorder.totals(); + + console.log(`\n${formatStatsTable(stats)}`); + console.log( + `\n[engine-http] ${totals.count} requests (${totals.errors} errors) at ` + + `${totals.throughputPerSecond.toFixed(1)} req/s | ` + + `${dequeuedRuns} dequeued, ${completedRuns} completed, ${emptyDequeues} empty dequeues` + ); + console.log( + `[engine-http] webapp ELU mean ${(elu.stats.mean * 100).toFixed(1)}% ` + + `p50 ${(elu.stats.p50 * 100).toFixed(1)}% ` + + `p95 ${(elu.stats.p95 * 100).toFixed(1)}% ` + + `max ${(elu.stats.max * 100).toFixed(1)}% (${elu.stats.sampleCount} samples)` + ); + console.log(`[engine-http] profile: ${profile.path} (${profile.sampleCount} samples)`); + + await writeFile( + join(OUT_DIR, `${PROFILE_NAME}-bench-summary.json`), + JSON.stringify( + { + config: { + runs: RUNS, + supervisors: SUPERVISORS, + heartbeatsPerRun: HEARTBEATS_PER_RUN, + durationMs: DURATION_MS, + samplingIntervalUs: SAMPLING_INTERVAL_US, + }, + triggered, + dequeuedRuns, + completedRuns, + emptyDequeues, + totals, + operations: stats, + elu: elu.stats, + eluSamples: elu.samples, + profilePath: profile.path, + }, + null, + 2 + ) + ); + + expect(dequeuedRuns).toBeGreaterThan(0); + }); +}); diff --git a/apps/webapp/test/bench/lib/cdp.ts b/apps/webapp/test/bench/lib/cdp.ts new file mode 100644 index 00000000000..b708d02876c --- /dev/null +++ b/apps/webapp/test/bench/lib/cdp.ts @@ -0,0 +1,227 @@ +/** + * Minimal Chrome DevTools Protocol client for benchmarking a spawned webapp. + * + * The bench spawns the webapp with `--inspect=` and drives the V8 CPU + * profiler over CDP rather than using `--cpu-prof`. Two reasons: + * + * 1. `--cpu-prof` only writes at process exit, so its profile covers boot, + * module init and shutdown as well as the load. Boot dominates a short run + * and buries the request-path frames this pass is about. + * 2. Over CDP the profiler can be started and stopped around the measured + * window only, and several separately-named profiles can be taken from a + * single webapp instance. + * + * The same connection samples `performance.eventLoopUtilization()` inside the + * target process, which is the number this pass is trying to move. Sampling it + * from the bench process would only describe the load generator. + */ +import { writeFile } from "node:fs/promises"; +import { WebSocket } from "ws"; + +type CdpMessage = { + id?: number; + result?: unknown; + error?: { code: number; message: string }; +}; + +export type EluSample = { + /** ms since the sampler started */ + atMs: number; + /** utilization over the interval since the previous sample, 0..1 */ + utilization: number; +}; + +export type EluStats = { + mean: number; + p50: number; + p95: number; + p99: number; + max: number; + sampleCount: number; +}; + +/** + * Node prints the inspector ws URL to stderr on boot, but the bench does not + * own the spawn, so discover it over the inspector's HTTP endpoint instead. + */ +async function discoverWebSocketUrl(port: number, timeoutMs = 30_000): Promise { + const deadline = Date.now() + timeoutMs; + let lastError: unknown; + + while (Date.now() < deadline) { + try { + const res = await fetch(`http://127.0.0.1:${port}/json/list`); + const targets = (await res.json()) as Array<{ webSocketDebuggerUrl?: string }>; + const url = targets.find((t) => t.webSocketDebuggerUrl)?.webSocketDebuggerUrl; + if (url) return url; + } catch (err) { + lastError = err; + } + await new Promise((r) => setTimeout(r, 200)); + } + + throw new Error(`No inspector target on port ${port} after ${timeoutMs}ms: ${lastError}`); +} + +class CdpSession { + private ws: WebSocket; + private nextId = 1; + private pending = new Map void; reject: (e: Error) => void }>(); + + private constructor(ws: WebSocket) { + this.ws = ws; + this.ws.on("message", (data) => { + const msg = JSON.parse(data.toString()) as CdpMessage; + if (msg.id === undefined) return; + const waiter = this.pending.get(msg.id); + if (!waiter) return; + this.pending.delete(msg.id); + if (msg.error) waiter.reject(new Error(`${msg.error.message} (${msg.error.code})`)); + else waiter.resolve(msg.result); + }); + } + + /** + * A full CPU profile of a busy minute is tens of MB and arrives as a single + * ws frame, so the payload cap is raised well past the 100MB default. + */ + static async connect(inspectPort: number): Promise { + const url = await discoverWebSocketUrl(inspectPort); + const ws = new WebSocket(url, { maxPayload: 512 * 1024 * 1024 }); + await new Promise((resolve, reject) => { + ws.once("open", () => resolve()); + ws.once("error", reject); + }); + return new CdpSession(ws); + } + + send(method: string, params: Record = {}): Promise { + const id = this.nextId++; + const promise = new Promise((resolve, reject) => { + this.pending.set(id, { resolve, reject }); + }); + this.ws.send(JSON.stringify({ id, method, params })); + return promise; + } + + close(): void { + this.ws.close(); + } +} + +export class WebappProfiler { + private session: CdpSession; + private eluTimer: NodeJS.Timeout | null = null; + private eluSamples: EluSample[] = []; + private eluStartedAt = 0; + + private constructor(session: CdpSession) { + this.session = session; + } + + static async attach(inspectPort: number): Promise { + const session = await CdpSession.connect(inspectPort); + await session.send("Runtime.enable"); + await session.send("Profiler.enable"); + return new WebappProfiler(session); + } + + /** + * `intervalUs` is V8's sampling interval in microseconds. The 200us default is + * 5x finer than V8's own 1ms: the engine routes are short, and at 1ms too few + * samples land inside a single request to separate the frames within it. + */ + async startCpuProfile(intervalUs = 200): Promise { + await this.session.send("Profiler.setSamplingInterval", { interval: intervalUs }); + await this.session.send("Profiler.start"); + } + + async stopCpuProfile(outPath: string): Promise<{ path: string; sampleCount: number }> { + const { profile } = await this.session.send<{ profile: { samples?: number[] } }>( + "Profiler.stop" + ); + await writeFile(outPath, JSON.stringify(profile)); + return { path: outPath, sampleCount: profile.samples?.length ?? 0 }; + } + + /** + * The first evaluate seeds a baseline reading so that the first recorded + * delta is measured from the moment sampling started rather than from + * process boot. + */ + startEluSampling(intervalMs = 250): void { + this.eluSamples = []; + this.eluStartedAt = Date.now(); + + void this.evaluateElu(); + + this.eluTimer = setInterval(() => { + void this.evaluateElu().then((utilization) => { + if (utilization !== undefined) { + this.eluSamples.push({ atMs: Date.now() - this.eluStartedAt, utilization }); + } + }); + }, intervalMs); + } + + /** + * Stashes the previous reading on globalThis inside the target so each call + * reports the delta since the last sample. A raw + * `performance.eventLoopUtilization()` is a since-boot average, which idle + * boot time drags down and which never recovers during a short run. + */ + private async evaluateElu(): Promise { + try { + const res = await this.session.send<{ result: { value?: number } }>("Runtime.evaluate", { + expression: `(() => { + const now = performance.eventLoopUtilization(); + const prev = globalThis.__benchLastElu; + globalThis.__benchLastElu = now; + if (!prev) return 0; + const diff = performance.eventLoopUtilization(now, prev); + return Number.isFinite(diff.utilization) ? diff.utilization : 0; + })()`, + returnByValue: true, + }); + return res.result?.value; + } catch { + return undefined; + } + } + + /** + * Drops the seeded first sample, which spans the gap between attach and the + * start of load and therefore reads artificially low. + */ + stopEluSampling(): { stats: EluStats; samples: EluSample[] } { + if (this.eluTimer) { + clearInterval(this.eluTimer); + this.eluTimer = null; + } + + const samples = this.eluSamples.slice(1); + if (samples.length === 0) { + return { stats: { mean: 0, p50: 0, p95: 0, p99: 0, max: 0, sampleCount: 0 }, samples }; + } + + const sorted = samples.map((s) => s.utilization).sort((a, b) => a - b); + const at = (q: number) => sorted[Math.min(sorted.length - 1, Math.floor(sorted.length * q))]!; + + return { + stats: { + mean: sorted.reduce((a, b) => a + b, 0) / sorted.length, + p50: at(0.5), + p95: at(0.95), + p99: at(0.99), + max: sorted[sorted.length - 1]!, + sampleCount: sorted.length, + }, + samples, + }; + } + + detach(): void { + this.stopEluSampling(); + this.session.close(); + } +} diff --git a/apps/webapp/test/bench/lib/engineFixtures.ts b/apps/webapp/test/bench/lib/engineFixtures.ts new file mode 100644 index 00000000000..442cff6eb0f --- /dev/null +++ b/apps/webapp/test/bench/lib/engineFixtures.ts @@ -0,0 +1,199 @@ +/** + * Seeds the minimum a supervisor needs to exist against: a production + * environment with a promoted managed deployment, and a worker group whose + * token authenticates the `engine/v1/worker-actions/*` routes. + * + * Rows are written straight through prisma rather than through the deploy + * services. The bench is measuring the worker-action request path, and going + * through the real deploy flow would add a lot of setup surface without + * changing a single byte of what those routes read. + */ +import { CURRENT_DEPLOYMENT_LABEL, generateFriendlyId } from "@trigger.dev/core/v3/isomorphic"; +import type { PrismaClient } from "@trigger.dev/database"; +import { createHash, randomBytes } from "node:crypto"; + +const MANAGED_WORKER_SECRET = "test-managed-worker-secret-for-e2e-tests"; + +export type EngineFixtures = { + organizationId: string; + projectId: string; + projectRef: string; + environmentId: string; + environmentApiKey: string; + workerGroupId: string; + workerGroupToken: string; + masterQueue: string; + taskIdentifiers: string[]; + deploymentId: string; +}; + +function randomHex(length = 12): string { + return randomBytes(Math.ceil(length / 2)) + .toString("hex") + .slice(0, length); +} + +export async function seedEngineFixtures( + prisma: PrismaClient, + options: { taskCount?: number; concurrencyLimit?: number } = {} +): Promise { + const taskCount = options.taskCount ?? 4; + const concurrencyLimit = options.concurrencyLimit ?? 500; + const suffix = randomHex(8); + + const masterQueue = `bench-${suffix}`; + const plaintextToken = `tr_wgt_${randomHex(40)}`; + const tokenHash = createHash("sha256").update(plaintextToken).digest("hex"); + + const organization = await prisma.organization.create({ + data: { title: `bench-org-${suffix}`, slug: `bench-org-${suffix}`, isActivated: true }, + }); + + const workerGroup = await prisma.workerInstanceGroup.create({ + data: { + name: `bench-group-${suffix}`, + masterQueue, + type: "MANAGED", + token: { create: { tokenHash } }, + }, + }); + + const project = await prisma.project.create({ + data: { + name: `bench-project-${suffix}`, + slug: `bench-proj-${suffix}`, + externalRef: `proj_${suffix}`, + organizationId: organization.id, + engine: "V2", + defaultWorkerGroupId: workerGroup.id, + }, + }); + + const environmentApiKey = `tr_prod_${randomHex(24)}`; + + const environment = await prisma.runtimeEnvironment.create({ + data: { + slug: "prod", + type: "PRODUCTION", + apiKey: environmentApiKey, + pkApiKey: `pk_prod_${randomHex(24)}`, + shortcode: suffix.slice(0, 6), + projectId: project.id, + organizationId: organization.id, + maximumConcurrencyLimit: concurrencyLimit, + }, + }); + + const version = "20260101.1"; + + const worker = await prisma.backgroundWorker.create({ + data: { + friendlyId: generateFriendlyId("worker"), + contentHash: `hash_${suffix}`, + projectId: project.id, + runtimeEnvironmentId: environment.id, + version, + metadata: {}, + engine: "V2", + }, + }); + + const taskIdentifiers = Array.from({ length: taskCount }, (_, i) => `bench-task-${i}`); + + for (const identifier of taskIdentifiers) { + const task = await prisma.backgroundWorkerTask.create({ + data: { + friendlyId: generateFriendlyId("task"), + slug: identifier, + filePath: `/trigger/${identifier}.ts`, + exportName: identifier, + workerId: worker.id, + runtimeEnvironmentId: environment.id, + projectId: project.id, + retryConfig: { + maxAttempts: 1, + factor: 1, + minTimeoutInMs: 100, + maxTimeoutInMs: 100, + randomize: false, + }, + }, + }); + + await prisma.taskQueue.upsert({ + where: { + runtimeEnvironmentId_name: { + name: `task/${identifier}`, + runtimeEnvironmentId: environment.id, + }, + }, + create: { + friendlyId: generateFriendlyId("queue"), + name: `task/${identifier}`, + concurrencyLimit, + runtimeEnvironmentId: environment.id, + projectId: project.id, + type: "VIRTUAL", + workers: { connect: { id: worker.id } }, + tasks: { connect: { id: task.id } }, + }, + update: { + concurrencyLimit, + workers: { connect: { id: worker.id } }, + tasks: { connect: { id: task.id } }, + }, + }); + } + + const deployment = await prisma.workerDeployment.create({ + data: { + friendlyId: generateFriendlyId("deployment"), + contentHash: worker.contentHash, + version, + shortCode: `short_${suffix}`, + imageReference: `bench/${project.externalRef}:${version}`, + status: "DEPLOYED", + projectId: project.id, + environmentId: environment.id, + workerId: worker.id, + type: "MANAGED", + }, + }); + + await prisma.workerDeploymentPromotion.upsert({ + where: { + environmentId_label: { environmentId: environment.id, label: CURRENT_DEPLOYMENT_LABEL }, + }, + create: { + deploymentId: deployment.id, + environmentId: environment.id, + label: CURRENT_DEPLOYMENT_LABEL, + }, + update: { deploymentId: deployment.id }, + }); + + return { + organizationId: organization.id, + projectId: project.id, + projectRef: project.externalRef, + environmentId: environment.id, + environmentApiKey, + workerGroupId: workerGroup.id, + workerGroupToken: plaintextToken, + masterQueue, + taskIdentifiers, + deploymentId: deployment.id, + }; +} + +/** + * Headers a managed supervisor sends on every worker-action request. + */ +export function workerHeaders(fixtures: EngineFixtures, instanceName: string): HeadersInit { + return { + Authorization: `Bearer ${fixtures.workerGroupToken}`, + "x-trigger-worker-instance-name": instanceName, + "x-trigger-worker-managed-secret": MANAGED_WORKER_SECRET, + "content-type": "application/json", + }; +} diff --git a/apps/webapp/test/bench/lib/loadDriver.ts b/apps/webapp/test/bench/lib/loadDriver.ts new file mode 100644 index 00000000000..eef6ccc8b79 --- /dev/null +++ b/apps/webapp/test/bench/lib/loadDriver.ts @@ -0,0 +1,149 @@ +/** + * Closed-loop load driver. + * + * A fixed number of virtual workers each run an async task in a loop for the + * duration of the run. Closed-loop, rather than a fixed arrival rate, is the + * right model here because it is what the supervisor actually does: it holds a + * bounded pool of consumers and each one issues its next request only after the + * previous one returns. An open-loop generator would queue work the real caller + * would never have sent, and the latency tail would then describe the + * generator's own backlog instead of the server. + */ +export type OperationStats = { + name: string; + count: number; + errors: number; + throughputPerSecond: number; + meanMs: number; + p50Ms: number; + p95Ms: number; + p99Ms: number; + maxMs: number; +}; + +export class LatencyRecorder { + private byName = new Map(); + private startedAt = 0; + private endedAt = 0; + + begin(): void { + this.startedAt = performance.now(); + } + + end(): void { + this.endedAt = performance.now(); + } + + record(name: string, durationMs: number, ok: boolean): void { + let entry = this.byName.get(name); + if (!entry) { + entry = { durations: [], errors: 0 }; + this.byName.set(name, entry); + } + entry.durations.push(durationMs); + if (!ok) entry.errors += 1; + } + + /** Times `fn` under `name`. A rejection is recorded as an error, never rethrown. */ + async time(name: string, fn: () => Promise): Promise { + const start = performance.now(); + try { + const result = await fn(); + this.record(name, performance.now() - start, true); + return result; + } catch { + this.record(name, performance.now() - start, false); + return undefined; + } + } + + private elapsedSeconds(): number { + const end = this.endedAt || performance.now(); + return Math.max((end - this.startedAt) / 1000, 1e-9); + } + + stats(): OperationStats[] { + const seconds = this.elapsedSeconds(); + + return [...this.byName.entries()] + .map(([name, { durations, errors }]) => { + const sorted = [...durations].sort((a, b) => a - b); + const at = (q: number) => + sorted[Math.min(sorted.length - 1, Math.floor(sorted.length * q))] ?? 0; + + return { + name, + count: sorted.length, + errors, + throughputPerSecond: sorted.length / seconds, + meanMs: sorted.reduce((a, b) => a + b, 0) / (sorted.length || 1), + p50Ms: at(0.5), + p95Ms: at(0.95), + p99Ms: at(0.99), + maxMs: sorted[sorted.length - 1] ?? 0, + }; + }) + .sort((a, b) => b.count - a.count); + } + + totals(): { count: number; errors: number; throughputPerSecond: number } { + const all = this.stats(); + const count = all.reduce((sum, s) => sum + s.count, 0); + const errors = all.reduce((sum, s) => sum + s.errors, 0); + return { count, errors, throughputPerSecond: count / this.elapsedSeconds() }; + } +} + +export type LoadRunOptions = { + /** Number of concurrent virtual workers. */ + concurrency: number; + /** How long to keep issuing work. */ + durationMs: number; + /** One iteration of a single virtual worker. */ + iteration: (workerIndex: number, iterationIndex: number) => Promise; + /** Runs once per worker before the measured window. Not recorded. */ + warmup?: (workerIndex: number) => Promise; +}; + +export async function runLoad(options: LoadRunOptions): Promise { + const { concurrency, durationMs, iteration, warmup } = options; + + if (warmup) { + await Promise.all(Array.from({ length: concurrency }, (_, i) => warmup(i))); + } + + const deadline = Date.now() + durationMs; + + await Promise.all( + Array.from({ length: concurrency }, async (_, workerIndex) => { + let iterationIndex = 0; + while (Date.now() < deadline) { + await iteration(workerIndex, iterationIndex++); + } + }) + ); +} + +export function formatStatsTable(stats: OperationStats[]): string { + const header = ["operation", "count", "err", "req/s", "mean", "p50", "p95", "p99", "max"]; + const rows = stats.map((s) => [ + s.name, + String(s.count), + String(s.errors), + s.throughputPerSecond.toFixed(1), + `${s.meanMs.toFixed(2)}ms`, + `${s.p50Ms.toFixed(2)}ms`, + `${s.p95Ms.toFixed(2)}ms`, + `${s.p99Ms.toFixed(2)}ms`, + `${s.maxMs.toFixed(2)}ms`, + ]); + + const widths = header.map((h, i) => Math.max(h.length, ...rows.map((r) => r[i]!.length))); + const line = (cells: string[]) => + cells + .map((c, i) => c.padEnd(widths[i]!)) + .join(" ") + .trimEnd(); + + return [line(header), line(widths.map((w) => "-".repeat(w))), ...rows.map(line)].join("\n"); +} diff --git a/apps/webapp/test/bench/lib/profileAnalysis.ts b/apps/webapp/test/bench/lib/profileAnalysis.ts new file mode 100644 index 00000000000..6412a935ade --- /dev/null +++ b/apps/webapp/test/bench/lib/profileAnalysis.ts @@ -0,0 +1,319 @@ +/** + * Turns a V8 `.cpuprofile` into ranked CPU attribution. + * + * Three views, because they answer different questions: + * + * - by bucket: which package or area of the codebase owns the cycles. This is + * the one that says "zod costs more than the database driver". + * - by function (self time): the individual frames to go and fix. + * - by function (total time): entry points, to sanity-check that the load + * actually exercised the route mix that was intended. + * + * Frames are symbolicated through the build's source maps first, so a bundled + * chunk name is reported as the source file it came from. + */ +import { SourceMapResolver } from "./sourcemap"; + +type CallFrame = { + functionName: string; + url: string; + lineNumber: number; + columnNumber: number; +}; + +type ProfileNode = { + id: number; + callFrame: CallFrame; + hitCount?: number; + children?: number[]; +}; + +export type CpuProfile = { + nodes: ProfileNode[]; + startTime: number; + endTime: number; + samples: number[]; + timeDeltas: number[]; +}; + +type FrameStat = { + key: string; + functionName: string; + location: string; + selfMs: number; + selfPercent: number; + totalMs: number; + totalPercent: number; +}; + +type BucketStat = { + bucket: string; + selfMs: number; + selfPercent: number; +}; + +export type ProfileAnalysis = { + wallClockMs: number; + sampledMs: number; + activeMs: number; + idleMs: number; + sampleCount: number; + buckets: BucketStat[]; + bySelfTime: FrameStat[]; + byTotalTime: FrameStat[]; +}; + +const IDLE_BUCKETS = new Set(["(idle)", "(program)"]); + +const FILE_URL_PREFIX = /^file:[/][/]/; +const PNPM_PACKAGE = /node_modules[/]\.pnpm[/]([^/]+)[/]node_modules[/](.+)$/; +const PLAIN_PACKAGE = /node_modules[/](.+)$/; +const WORKSPACE_PACKAGE = /^(?:\.\.[/])*((?:internal-)?packages)[/]([^/]+)[/]/; +const WEBAPP_AREA = /^(?:\.\.[/])*(?:apps[/]webapp[/])?app[/]([^/]+)[/]([^/]+)/; + +function packageNameFrom(specifier: string): string { + const parts = specifier.split("/"); + return parts[0]!.startsWith("@") ? `${parts[0]}/${parts[1]}` : parts[0]!; +} + +/** + * Collapses a source path to the unit a fix would be scoped to: a third-party + * package, a workspace package, or an area of the webapp. + */ +function bucketForSource(source: string, functionName: string): string { + if (source === "(gc)" || functionName === "(garbage collector)") return "(gc)"; + if (source === "(program)" || functionName === "(program)") return "(program)"; + if (source === "(idle)" || functionName === "(idle)") return "(idle)"; + if (source.startsWith("node:")) return source; + + const pnpmMatch = source.match(PNPM_PACKAGE); + if (pnpmMatch) return `npm:${packageNameFrom(pnpmMatch[2]!)}`; + + const plainMatch = source.match(PLAIN_PACKAGE); + if (plainMatch) return `npm:${packageNameFrom(plainMatch[1]!)}`; + + const workspaceMatch = source.match(WORKSPACE_PACKAGE); + if (workspaceMatch) return `${workspaceMatch[1]}/${workspaceMatch[2]}`; + + const webappMatch = source.match(WEBAPP_AREA); + if (webappMatch) { + const [, top, second] = webappMatch; + return top === "routes" ? "webapp/routes" : `webapp/app/${top}/${second}`; + } + + if (source.includes("build/server") || source.includes("build/")) return "webapp/(unmapped)"; + + return source.split("/").slice(0, 3).join("/") || "(unknown)"; +} + +function resolveFrame( + frame: CallFrame, + resolver: SourceMapResolver, + repoRoot: string +): { source: string; location: string } { + const name = frame.functionName || "(anonymous)"; + + if (!frame.url) { + const synthetic = name.startsWith("(") ? name : "(native)"; + return { source: synthetic, location: synthetic }; + } + + if (frame.url.startsWith("node:")) { + return { source: frame.url, location: frame.url }; + } + + const filePath = frame.url.replace(FILE_URL_PREFIX, ""); + const mapped = resolver.resolve(filePath, frame.lineNumber, frame.columnNumber); + + if (mapped) { + return { source: mapped.source, location: `${mapped.source}:${mapped.line}` }; + } + + const relativePath = filePath.startsWith(repoRoot) + ? filePath.slice(repoRoot.length + 1) + : filePath; + return { source: relativePath, location: `${relativePath}:${frame.lineNumber + 1}` }; +} + +export function analyzeProfile(profile: CpuProfile, repoRoot: string): ProfileAnalysis { + const resolver = new SourceMapResolver(repoRoot); + + const parentOf = new Map(); + for (const node of profile.nodes) { + for (const childId of node.children ?? []) parentOf.set(childId, node.id); + } + + const selfMicrosByNode = new Map(); + let sampledMicros = 0; + + for (let i = 0; i < profile.samples.length; i++) { + const nodeId = profile.samples[i]!; + const delta = profile.timeDeltas[i] ?? 0; + if (delta <= 0) continue; + selfMicrosByNode.set(nodeId, (selfMicrosByNode.get(nodeId) ?? 0) + delta); + sampledMicros += delta; + } + + const resolvedByNode = new Map(); + for (const node of profile.nodes) { + const { source, location } = resolveFrame(node.callFrame, resolver, repoRoot); + resolvedByNode.set(node.id, { + source, + location, + name: node.callFrame.functionName || "(anonymous)", + }); + } + + const bucketMicros = new Map(); + const selfMicrosByFrame = new Map< + string, + { functionName: string; location: string; micros: number } + >(); + + for (const [nodeId, micros] of selfMicrosByNode) { + const resolved = resolvedByNode.get(nodeId); + if (!resolved) continue; + + const bucket = bucketForSource(resolved.source, resolved.name); + bucketMicros.set(bucket, (bucketMicros.get(bucket) ?? 0) + micros); + + const key = `${resolved.name}@${resolved.location}`; + const existing = selfMicrosByFrame.get(key); + if (existing) { + existing.micros += micros; + } else { + selfMicrosByFrame.set(key, { + micros, + functionName: resolved.name, + location: resolved.location, + }); + } + } + + const totalMicrosByFrame = new Map(); + for (const [nodeId, micros] of selfMicrosByNode) { + const seen = new Set(); + let current: number | undefined = nodeId; + + while (current !== undefined) { + const resolved = resolvedByNode.get(current); + if (resolved) { + const key = `${resolved.name}@${resolved.location}`; + if (!seen.has(key)) { + seen.add(key); + totalMicrosByFrame.set(key, (totalMicrosByFrame.get(key) ?? 0) + micros); + } + } + current = parentOf.get(current); + } + } + + let idleMicros = 0; + for (const [bucket, micros] of bucketMicros) { + if (IDLE_BUCKETS.has(bucket)) idleMicros += micros; + } + + const denominator = sampledMicros - idleMicros || 1; + const toMs = (micros: number) => micros / 1000; + const toPercent = (micros: number) => (micros / denominator) * 100; + + const bySelfTime: FrameStat[] = [...selfMicrosByFrame.entries()] + .filter(([, { functionName }]) => !IDLE_BUCKETS.has(functionName)) + .map(([key, { functionName, location, micros }]) => ({ + key, + functionName, + location, + selfMs: toMs(micros), + selfPercent: toPercent(micros), + totalMs: toMs(totalMicrosByFrame.get(key) ?? micros), + totalPercent: toPercent(totalMicrosByFrame.get(key) ?? micros), + })) + .sort((a, b) => b.selfMs - a.selfMs); + + const byTotalTime = [...bySelfTime].sort((a, b) => b.totalMs - a.totalMs); + + const buckets: BucketStat[] = [...bucketMicros.entries()] + .filter(([bucket]) => !IDLE_BUCKETS.has(bucket)) + .map(([bucket, micros]) => ({ + bucket, + selfMs: toMs(micros), + selfPercent: toPercent(micros), + })) + .sort((a, b) => b.selfMs - a.selfMs); + + return { + wallClockMs: (profile.endTime - profile.startTime) / 1000, + sampledMs: toMs(sampledMicros), + activeMs: toMs(sampledMicros - idleMicros), + idleMs: toMs(idleMicros), + sampleCount: profile.samples.length, + buckets, + bySelfTime, + byTotalTime, + }; +} + +function table(header: string[], rows: string[][]): string { + const widths = header.map((h, i) => Math.max(h.length, ...rows.map((r) => r[i]!.length))); + const line = (cells: string[]) => + cells + .map((c, i) => (i === 0 ? c.padEnd(widths[i]!) : c.padStart(widths[i]!))) + .join(" ") + .trimEnd(); + return [line(header), line(widths.map((w) => "-".repeat(w))), ...rows.map(line)].join("\n"); +} + +export function formatAnalysis(analysis: ProfileAnalysis, topN = 30): string { + const sections: string[] = []; + + const busy = analysis.sampledMs > 0 ? (analysis.activeMs / analysis.sampledMs) * 100 : 0; + + sections.push( + `wall clock ${analysis.wallClockMs.toFixed(0)}ms | on-cpu ${analysis.activeMs.toFixed(0)}ms ` + + `(${busy.toFixed(1)}% busy, ${analysis.idleMs.toFixed(0)}ms idle) | ${analysis.sampleCount} samples\n` + + `percentages below are shares of on-cpu time, not of wall clock` + ); + + sections.push( + "\nCPU by bucket (self time)\n" + + table( + ["bucket", "self ms", "self %"], + analysis.buckets + .slice(0, topN) + .map((b) => [b.bucket, b.selfMs.toFixed(1), `${b.selfPercent.toFixed(2)}%`]) + ) + ); + + sections.push( + "\nHottest frames (self time)\n" + + table( + ["function", "location", "self ms", "self %", "total %"], + analysis.bySelfTime + .slice(0, topN) + .map((f) => [ + f.functionName, + f.location, + f.selfMs.toFixed(1), + `${f.selfPercent.toFixed(2)}%`, + `${f.totalPercent.toFixed(2)}%`, + ]) + ) + ); + + sections.push( + "\nHottest frames (total time, inclusive)\n" + + table( + ["function", "location", "total ms", "total %"], + analysis.byTotalTime + .slice(0, topN) + .map((f) => [ + f.functionName, + f.location, + f.totalMs.toFixed(1), + `${f.totalPercent.toFixed(2)}%`, + ]) + ) + ); + + return sections.join("\n"); +} diff --git a/apps/webapp/test/bench/lib/sourcemap.ts b/apps/webapp/test/bench/lib/sourcemap.ts new file mode 100644 index 00000000000..3a35d0c26f1 --- /dev/null +++ b/apps/webapp/test/bench/lib/sourcemap.ts @@ -0,0 +1,159 @@ +/** + * Just enough source-map support to turn a bundled `.cpuprofile` frame back + * into a repo-relative source path. + * + * Deliberately dependency-free: the only thing needed is generated + * (line, column) to original source path, and adding a resolver package to the + * webapp for a bench-only tool is not worth the lockfile churn. + */ +import { readFileSync } from "node:fs"; +import { dirname, relative, resolve } from "node:path"; + +const BASE64 = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/"; + +const BASE64_INDEX = new Map([...BASE64].map((char, index) => [char, index])); + +const FILE_URL_PREFIX = /^file:[/][/]/; + +function decodeVlq(segment: string, state: { pos: number }): number { + let result = 0; + let shift = 0; + let continuation = true; + + while (continuation) { + const digit = BASE64_INDEX.get(segment[state.pos]!); + if (digit === undefined) { + throw new Error(`Invalid VLQ character at ${state.pos} in "${segment}"`); + } + state.pos += 1; + continuation = (digit & 32) !== 0; + result += (digit & 31) << shift; + shift += 5; + } + + const negative = (result & 1) === 1; + result >>>= 1; + return negative ? -result : result; +} + +type Mapping = { + generatedColumn: number; + sourceIndex: number; + originalLine: number; +}; + +type RawSourceMap = { + sources: (string | null)[]; + sourceRoot?: string; + mappings: string; +}; + +type ParsedSourceMap = { sources: string[]; lines: Mapping[][] }; + +export class SourceMapResolver { + private byGeneratedFile = new Map(); + private repoRoot: string; + + constructor(repoRoot: string) { + this.repoRoot = repoRoot; + } + + private load(generatedPath: string): ParsedSourceMap | null { + if (this.byGeneratedFile.has(generatedPath)) { + return this.byGeneratedFile.get(generatedPath)!; + } + + let parsed: ParsedSourceMap | null = null; + + try { + const raw = JSON.parse(readFileSync(`${generatedPath}.map`, "utf8")) as RawSourceMap; + const mapDir = dirname(generatedPath); + const sourceRoot = raw.sourceRoot ?? ""; + + const sources = raw.sources.map((source) => { + if (!source) return "(unknown)"; + const absolute = resolve(mapDir, sourceRoot, source.replace(FILE_URL_PREFIX, "")); + return relative(this.repoRoot, absolute); + }); + + parsed = { sources, lines: decodeMappings(raw.mappings) }; + } catch { + parsed = null; + } + + this.byGeneratedFile.set(generatedPath, parsed); + return parsed; + } + + /** + * Returns the repo-relative original source and 1-based line for a generated + * position, or undefined when there is no map or no mapping at or before + * that column. + */ + resolve( + generatedPath: string, + line: number, + column: number + ): { source: string; line: number } | undefined { + const map = this.load(generatedPath); + if (!map) return undefined; + + const mappings = map.lines[line]; + if (!mappings || mappings.length === 0) return undefined; + + let low = 0; + let high = mappings.length - 1; + let found: Mapping | undefined; + + while (low <= high) { + const mid = (low + high) >> 1; + const candidate = mappings[mid]!; + if (candidate.generatedColumn <= column) { + found = candidate; + low = mid + 1; + } else { + high = mid - 1; + } + } + + const mapping = found ?? mappings[0]!; + const source = map.sources[mapping.sourceIndex]; + if (source === undefined) return undefined; + return { source, line: mapping.originalLine + 1 }; + } +} + +function decodeMappings(mappings: string): Mapping[][] { + const lines: Mapping[][] = []; + + let sourceIndex = 0; + let originalLine = 0; + + for (const lineSegments of mappings.split(";")) { + const decoded: Mapping[] = []; + let generatedColumn = 0; + + if (lineSegments.length > 0) { + for (const segment of lineSegments.split(",")) { + if (segment.length === 0) continue; + + const state = { pos: 0 }; + generatedColumn += decodeVlq(segment, state); + + if (state.pos < segment.length) { + sourceIndex += decodeVlq(segment, state); + originalLine += decodeVlq(segment, state); + decodeVlq(segment, state); + if (state.pos < segment.length) decodeVlq(segment, state); + + decoded.push({ generatedColumn, sourceIndex, originalLine }); + } + } + } + + decoded.sort((a, b) => a.generatedColumn - b.generatedColumn); + lines.push(decoded); + } + + return lines; +} diff --git a/apps/webapp/vitest.bench.config.ts b/apps/webapp/vitest.bench.config.ts new file mode 100644 index 00000000000..a80ab5a31fe --- /dev/null +++ b/apps/webapp/vitest.bench.config.ts @@ -0,0 +1,22 @@ +import { defineConfig } from "vitest/config"; +import tsconfigPaths from "vite-tsconfig-paths"; + +/** + * CPU benchmarks. Kept out of the default suite because they spawn a webapp, + * run for minutes, attach the V8 profiler, and report numbers rather than + * assert on them: on a shared runner the timings swing far more than any + * threshold worth gating on. Needs a built webapp (`pnpm run build --filter + * webapp`). Run on demand with `pnpm run test:bench`. + */ +export default defineConfig({ + test: { + include: ["test/bench/**/*.bench.test.ts"], + globals: true, + pool: "forks", + fileParallelism: false, + testTimeout: 900_000, + setupFiles: ["./test/setup.ts"], + }, + // @ts-ignore + plugins: [tsconfigPaths({ projects: ["./tsconfig.json"] })], +}); diff --git a/apps/webapp/vitest.config.ts b/apps/webapp/vitest.config.ts index 995c1742896..1981ce26a06 100644 --- a/apps/webapp/vitest.config.ts +++ b/apps/webapp/vitest.config.ts @@ -27,7 +27,12 @@ export default defineConfig({ // *.e2e.test.ts: smoke matrix, run via vitest.e2e.config.ts. // *.e2e.full.test.ts: full auth suite, runs via vitest.e2e.full.config.ts // (needs a globalSetup-spawned webapp + Postgres container). - exclude: ["test/**/*.e2e.test.ts", "test/**/*.e2e.full.test.ts", "test/**/*.perf.test.ts"], + exclude: [ + "test/**/*.e2e.test.ts", + "test/**/*.e2e.full.test.ts", + "test/**/*.perf.test.ts", + "test/bench/**", + ], globals: true, pool: "forks", setupFiles: ["./test/setup.ts"], // load apps/webapp/.env diff --git a/internal-packages/run-engine/package.json b/internal-packages/run-engine/package.json index 51bb0d6da8e..96ace3a0e43 100644 --- a/internal-packages/run-engine/package.json +++ b/internal-packages/run-engine/package.json @@ -47,6 +47,7 @@ "test": "vitest --sequence.concurrent=false --no-file-parallelism", "test:coverage": "vitest --sequence.concurrent=false --no-file-parallelism --coverage.enabled", "build": "pnpm run clean && tsc -p tsconfig.build.json", - "dev": "tsc --watch -p tsconfig.build.json" + "dev": "tsc --watch -p tsconfig.build.json", + "test:bench": "vitest --config ./vitest.bench.config.ts --run" } } diff --git a/internal-packages/run-engine/src/engine/bench/inspectorProfiler.ts b/internal-packages/run-engine/src/engine/bench/inspectorProfiler.ts new file mode 100644 index 00000000000..250598b3dcd --- /dev/null +++ b/internal-packages/run-engine/src/engine/bench/inspectorProfiler.ts @@ -0,0 +1,130 @@ +/** + * In-process CPU profiler and event-loop-utilization sampler. + * + * The webapp bench profiles a spawned child over CDP, but this bench drives the + * engine directly, so the code under measurement and the driver share a + * process. `node:inspector`'s Session profiles that process in place. + * + * The driver's own cost lands in the profile too. That is acceptable and + * visible: the driver is a thin await loop, and it shows up under its own + * frames in the ranked output rather than being smeared across engine frames. + */ +import { mkdir, writeFile } from "node:fs/promises"; +import { Session } from "node:inspector/promises"; +import { dirname } from "node:path"; +import { performance } from "node:perf_hooks"; + +const IDLE_FRAMES = new Set(["(idle)", "(program)"]); + +type ProfileShape = { + nodes: Array<{ id: number; callFrame: { functionName: string } }>; + samples?: number[]; + timeDeltas?: number[]; +}; + +function onCpuMicros(profile: ProfileShape): number { + const idleNodeIds = new Set( + profile.nodes + .filter((node) => IDLE_FRAMES.has(node.callFrame.functionName)) + .map((node) => node.id) + ); + + const samples = profile.samples ?? []; + const timeDeltas = profile.timeDeltas ?? []; + + let micros = 0; + for (let i = 0; i < samples.length; i++) { + const delta = timeDeltas[i] ?? 0; + if (delta > 0 && !idleNodeIds.has(samples[i]!)) micros += delta; + } + return micros; +} + +export type EluStats = { + mean: number; + p50: number; + p95: number; + p99: number; + max: number; + sampleCount: number; +}; + +export class InProcessProfiler { + private session: Session | null = null; + private eluTimer: NodeJS.Timeout | null = null; + private eluSamples: number[] = []; + private lastElu: ReturnType | null = null; + + /** + * `intervalUs` is V8's sampling interval in microseconds, 5x finer than + * V8's own 1ms default so short engine operations land enough samples to + * separate the frames inside them. + */ + async startCpuProfile(intervalUs = 200): Promise { + const session = new Session(); + session.connect(); + await session.post("Profiler.enable"); + await session.post("Profiler.setSamplingInterval", { interval: intervalUs }); + await session.post("Profiler.start"); + this.session = session; + } + + /** + * `onCpuMs` excludes V8's `(idle)` and `(program)` frames, so it is the time + * the phase actually held the event loop rather than the wall clock it + * spanned. That is the number a change has to move. + */ + async stopCpuProfile( + outPath: string + ): Promise<{ path: string; sampleCount: number; onCpuMs: number }> { + if (!this.session) throw new Error("startCpuProfile was never called"); + + const { profile } = await this.session.post("Profiler.stop"); + this.session.disconnect(); + this.session = null; + + await mkdir(dirname(outPath), { recursive: true }); + await writeFile(outPath, JSON.stringify(profile)); + + return { + path: outPath, + sampleCount: profile.samples?.length ?? 0, + onCpuMs: onCpuMicros(profile) / 1000, + }; + } + + startEluSampling(intervalMs = 250): void { + this.eluSamples = []; + this.lastElu = performance.eventLoopUtilization(); + + this.eluTimer = setInterval(() => { + const current = performance.eventLoopUtilization(); + const diff = performance.eventLoopUtilization(current, this.lastElu!); + this.lastElu = current; + this.eluSamples.push(Number.isFinite(diff.utilization) ? diff.utilization : 0); + }, intervalMs); + } + + stopEluSampling(): EluStats { + if (this.eluTimer) { + clearInterval(this.eluTimer); + this.eluTimer = null; + } + + if (this.eluSamples.length === 0) { + return { mean: 0, p50: 0, p95: 0, p99: 0, max: 0, sampleCount: 0 }; + } + + const sorted = [...this.eluSamples].sort((a, b) => a - b); + const at = (q: number) => sorted[Math.min(sorted.length - 1, Math.floor(sorted.length * q))]!; + + return { + mean: sorted.reduce((a, b) => a + b, 0) / sorted.length, + p50: at(0.5), + p95: at(0.95), + p99: at(0.99), + max: sorted[sorted.length - 1]!, + sampleCount: sorted.length, + }; + } +} diff --git a/internal-packages/run-engine/src/engine/bench/runEngineLifecycle.bench.test.ts b/internal-packages/run-engine/src/engine/bench/runEngineLifecycle.bench.test.ts new file mode 100644 index 00000000000..eb6e8ace00e --- /dev/null +++ b/internal-packages/run-engine/src/engine/bench/runEngineLifecycle.bench.test.ts @@ -0,0 +1,313 @@ +/** + * CPU benchmark for the run-engine and run-queue paths the production engine + * service spends its time in. + * + * Two measured phases, profiled separately because blending them hides which + * one owns a hot frame: + * + * - enqueue: `engine.trigger()` at volume, the write path. + * - lifecycle: dequeue, start attempt, heartbeats, complete attempt, which is + * the loop a supervisor actually runs. + * + * Artifacts land in `.bench/` at the repo root: a `.cpuprofile` per phase plus + * a JSON summary. Analyze a profile with: + * + * pnpm --filter webapp exec tsx test/bench/analyzeProfile.ts + * + * Knobs, all optional: + * BENCH_RUNS, BENCH_CONSUMERS, BENCH_HEARTBEATS, BENCH_CONCURRENCY_LIMIT, + * BENCH_OUT_DIR, BENCH_SAMPLING_INTERVAL_US + */ +import { containerTest } from "@internal/testcontainers"; +import { trace } from "@internal/tracing"; +import { generateFriendlyId } from "@trigger.dev/core/v3/isomorphic"; +import { mkdir, writeFile } from "node:fs/promises"; +import { join } from "node:path"; +import { setTimeout } from "node:timers/promises"; +import { RunEngine } from "../index.js"; +import { setupAuthenticatedEnvironment, setupBackgroundWorker } from "../tests/setup.js"; +import { InProcessProfiler } from "./inspectorProfiler.js"; + +vi.setConfig({ testTimeout: 900_000 }); + +const RUNS = Number(process.env.BENCH_RUNS ?? 1500); +const CONSUMERS = Number(process.env.BENCH_CONSUMERS ?? 8); +const HEARTBEATS_PER_RUN = Number(process.env.BENCH_HEARTBEATS ?? 2); +const CONCURRENCY_LIMIT = Number(process.env.BENCH_CONCURRENCY_LIMIT ?? 200); +const SAMPLING_INTERVAL_US = Number(process.env.BENCH_SAMPLING_INTERVAL_US ?? 200); +const OUT_DIR = process.env.BENCH_OUT_DIR ?? join(process.cwd(), "..", "..", ".bench"); + +const TASK_IDENTIFIER = "bench-task"; +const WORKER_QUEUE = "main"; + +type PhaseResult = { + phase: string; + operations: number; + durationMs: number; + opsPerSecond: number; + onCpuMs: number; + cpuPerOperationMs: number; + elu: ReturnType; + profilePath: string; + sampleCount: number; +}; + +function percentile(sorted: number[], q: number): number { + if (sorted.length === 0) return 0; + return sorted[Math.min(sorted.length - 1, Math.floor(sorted.length * q))]!; +} + +function summarize(durations: number[]): { p50: number; p95: number; p99: number; mean: number } { + const sorted = [...durations].sort((a, b) => a - b); + return { + mean: sorted.reduce((a, b) => a + b, 0) / (sorted.length || 1), + p50: percentile(sorted, 0.5), + p95: percentile(sorted, 0.95), + p99: percentile(sorted, 0.99), + }; +} + +describe("run-engine CPU benchmark", () => { + containerTest("enqueue and lifecycle under load", async ({ prisma, redisOptions }) => { + await mkdir(OUT_DIR, { recursive: true }); + + const environment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION"); + + const engine = new RunEngine({ + prisma, + worker: { redis: redisOptions, workers: 1, tasksPerWorker: 10, pollIntervalMs: 100 }, + queue: { + redis: redisOptions, + masterQueueConsumersDisabled: true, + processWorkerQueueDebounceMs: 50, + }, + runLock: { redis: redisOptions }, + machines: { + defaultMachine: "small-1x", + machines: { + "small-1x": { name: "small-1x" as const, cpu: 0.5, memory: 0.5, centsPerMs: 0.0001 }, + }, + baseCostInCents: 0.0005, + }, + tracer: trace.getTracer("bench", "0.0.0"), + }); + + const results: PhaseResult[] = []; + + try { + await prisma.runtimeEnvironment.update({ + where: { id: environment.id }, + data: { maximumConcurrencyLimit: CONCURRENCY_LIMIT }, + }); + environment.maximumConcurrencyLimit = CONCURRENCY_LIMIT; + + await setupBackgroundWorker(engine, environment, TASK_IDENTIFIER, undefined, undefined, { + concurrencyLimit: CONCURRENCY_LIMIT, + }); + await engine.runQueue.updateEnvConcurrencyLimits(environment); + + const enqueueDurations: number[] = []; + const enqueueProfiler = new InProcessProfiler(); + await enqueueProfiler.startCpuProfile(SAMPLING_INTERVAL_US); + enqueueProfiler.startEluSampling(); + + const enqueueStart = performance.now(); + + for (let i = 0; i < RUNS; i++) { + const started = performance.now(); + await engine.trigger( + { + number: i + 1, + friendlyId: generateFriendlyId("run"), + environment, + taskIdentifier: TASK_IDENTIFIER, + payload: JSON.stringify({ index: i, message: "bench payload" }), + payloadType: "application/json", + context: {}, + traceContext: {}, + traceId: `t${i.toString().padStart(16, "0")}`, + spanId: `s${i.toString().padStart(8, "0")}`, + workerQueue: WORKER_QUEUE, + queue: `task/${TASK_IDENTIFIER}`, + isTest: false, + tags: [], + }, + prisma + ); + enqueueDurations.push(performance.now() - started); + } + + const enqueueDurationMs = performance.now() - enqueueStart; + const enqueueElu = enqueueProfiler.stopEluSampling(); + const enqueueProfile = await enqueueProfiler.stopCpuProfile( + join(OUT_DIR, "run-engine-enqueue.cpuprofile") + ); + + results.push({ + phase: "enqueue", + operations: RUNS, + durationMs: enqueueDurationMs, + opsPerSecond: RUNS / (enqueueDurationMs / 1000), + onCpuMs: enqueueProfile.onCpuMs, + cpuPerOperationMs: enqueueProfile.onCpuMs / RUNS, + elu: enqueueElu, + profilePath: enqueueProfile.path, + sampleCount: enqueueProfile.sampleCount, + }); + + console.log( + `\n[enqueue] ${RUNS} runs in ${enqueueDurationMs.toFixed(0)}ms ` + + `(${(RUNS / (enqueueDurationMs / 1000)).toFixed(1)} runs/s), ` + + `on-cpu ${enqueueProfile.onCpuMs.toFixed(0)}ms ` + + `(${(enqueueProfile.onCpuMs / RUNS).toFixed(2)}ms cpu/run), ` + + `ELU mean ${(enqueueElu.mean * 100).toFixed(1)}% p95 ${(enqueueElu.p95 * 100).toFixed(1)}%` + ); + console.log(`[enqueue] latency ${JSON.stringify(summarize(enqueueDurations))}`); + + await setTimeout(1000); + + const dequeueDurations: number[] = []; + const startDurations: number[] = []; + const heartbeatDurations: number[] = []; + const completeDurations: number[] = []; + + let processed = 0; + let emptyDequeues = 0; + + const lifecycleProfiler = new InProcessProfiler(); + await lifecycleProfiler.startCpuProfile(SAMPLING_INTERVAL_US); + lifecycleProfiler.startEluSampling(); + + const lifecycleStart = performance.now(); + + await Promise.all( + Array.from({ length: CONSUMERS }, async (_, consumerIndex) => { + const consumerId = `bench_consumer_${consumerIndex}`; + + while (processed < RUNS && emptyDequeues < CONSUMERS * 5) { + const dequeueStarted = performance.now(); + const dequeued = await engine.dequeueFromWorkerQueue({ + consumerId, + workerQueue: WORKER_QUEUE, + workerId: consumerId, + }); + dequeueDurations.push(performance.now() - dequeueStarted); + + const message = dequeued[0]; + if (!message) { + emptyDequeues += 1; + await setTimeout(25); + continue; + } + + emptyDequeues = 0; + processed += 1; + + const startStarted = performance.now(); + const attempt = await engine.startRunAttempt({ + runId: message.run.id, + snapshotId: message.snapshot.id, + workerId: consumerId, + }); + startDurations.push(performance.now() - startStarted); + + let snapshotId = attempt.snapshot.id; + + for (let beat = 0; beat < HEARTBEATS_PER_RUN; beat++) { + const heartbeatStarted = performance.now(); + const heartbeat = await engine.heartbeatRun({ + runId: message.run.id, + snapshotId, + workerId: consumerId, + }); + heartbeatDurations.push(performance.now() - heartbeatStarted); + snapshotId = heartbeat.snapshot.id; + } + + const completeStarted = performance.now(); + await engine.completeRunAttempt({ + runId: message.run.id, + snapshotId, + workerId: consumerId, + completion: { + ok: true, + id: message.run.id, + outputType: "application/json", + output: JSON.stringify({ done: true }), + }, + }); + completeDurations.push(performance.now() - completeStarted); + } + }) + ); + + const lifecycleDurationMs = performance.now() - lifecycleStart; + const lifecycleElu = lifecycleProfiler.stopEluSampling(); + const lifecycleProfile = await lifecycleProfiler.stopCpuProfile( + join(OUT_DIR, "run-engine-lifecycle.cpuprofile") + ); + + const lifecycleOperations = + dequeueDurations.length + + startDurations.length + + heartbeatDurations.length + + completeDurations.length; + + results.push({ + phase: "lifecycle", + operations: lifecycleOperations, + durationMs: lifecycleDurationMs, + opsPerSecond: lifecycleOperations / (lifecycleDurationMs / 1000), + onCpuMs: lifecycleProfile.onCpuMs, + cpuPerOperationMs: lifecycleProfile.onCpuMs / (processed || 1), + elu: lifecycleElu, + profilePath: lifecycleProfile.path, + sampleCount: lifecycleProfile.sampleCount, + }); + + console.log( + `\n[lifecycle] ${processed} runs (${lifecycleOperations} engine calls) in ` + + `${lifecycleDurationMs.toFixed(0)}ms (${(processed / (lifecycleDurationMs / 1000)).toFixed(1)} runs/s), ` + + `on-cpu ${lifecycleProfile.onCpuMs.toFixed(0)}ms ` + + `(${(lifecycleProfile.onCpuMs / (processed || 1)).toFixed(2)}ms cpu/run), ` + + `ELU mean ${(lifecycleElu.mean * 100).toFixed(1)}% p95 ${(lifecycleElu.p95 * 100).toFixed(1)}%` + ); + console.log(`[lifecycle] dequeue ${JSON.stringify(summarize(dequeueDurations))}`); + console.log(`[lifecycle] startAttempt ${JSON.stringify(summarize(startDurations))}`); + console.log(`[lifecycle] heartbeat ${JSON.stringify(summarize(heartbeatDurations))}`); + console.log(`[lifecycle] complete ${JSON.stringify(summarize(completeDurations))}`); + + const summaryPath = join(OUT_DIR, "run-engine-bench-summary.json"); + await writeFile( + summaryPath, + JSON.stringify( + { + config: { + runs: RUNS, + consumers: CONSUMERS, + heartbeatsPerRun: HEARTBEATS_PER_RUN, + concurrencyLimit: CONCURRENCY_LIMIT, + samplingIntervalUs: SAMPLING_INTERVAL_US, + }, + phases: results, + latency: { + trigger: summarize(enqueueDurations), + dequeue: summarize(dequeueDurations), + startRunAttempt: summarize(startDurations), + heartbeatRun: summarize(heartbeatDurations), + completeRunAttempt: summarize(completeDurations), + }, + processed, + }, + null, + 2 + ) + ); + + console.log(`\n[bench] artifacts written to ${OUT_DIR}`); + expect(processed).toBeGreaterThan(0); + } finally { + await engine.quit(); + } + }); +}); diff --git a/internal-packages/run-engine/vitest.bench.config.ts b/internal-packages/run-engine/vitest.bench.config.ts new file mode 100644 index 00000000000..6210a0f0cd3 --- /dev/null +++ b/internal-packages/run-engine/vitest.bench.config.ts @@ -0,0 +1,17 @@ +import { defineConfig } from "vitest/config"; + +/** + * CPU benchmarks. Kept out of the default suite because they run for minutes, + * attach the V8 profiler, and report numbers rather than assert on them: on a + * shared runner the timings swing far more than any threshold worth gating on. + * Run on demand with `pnpm run test:bench`. + */ +export default defineConfig({ + test: { + include: ["**/*.bench.test.ts"], + globals: true, + isolate: true, + fileParallelism: false, + testTimeout: 900_000, + }, +}); diff --git a/internal-packages/run-engine/vitest.config.ts b/internal-packages/run-engine/vitest.config.ts index cb048f00927..64e119d3d70 100644 --- a/internal-packages/run-engine/vitest.config.ts +++ b/internal-packages/run-engine/vitest.config.ts @@ -5,6 +5,7 @@ export default defineConfig({ test: { sequence: { sequencer: DurationShardingSequencer }, include: ["**/*.test.ts"], + exclude: ["**/node_modules/**", "**/dist/**", "**/*.bench.test.ts"], globals: true, // CI-only: absorbs timing races (real-clock waits vs worker poll interval) under shard CPU contention retry: process.env.CI ? 2 : 0, diff --git a/internal-packages/testcontainers/src/webapp.ts b/internal-packages/testcontainers/src/webapp.ts index 2ff73b1b52b..d8a23377715 100644 --- a/internal-packages/testcontainers/src/webapp.ts +++ b/internal-packages/testcontainers/src/webapp.ts @@ -74,6 +74,17 @@ export interface StartWebappOptions { * session-stream e2e). `NODE_PATH` and the worker-disable vars still win. */ extraEnv?: Record; + + /** + * Like `extraEnv`, but applied after the worker-disable defaults so it can + * turn a background worker back on. The CPU benchmarks need the run engine + * worker running (`RUN_ENGINE_WORKER_ENABLED=1`), because that is what drains + * the master queue into the worker queues a supervisor dequeues from. + * + * Use `extraEnv` for everything else: a test that silently leaves a worker + * running is a flaky test. + */ + overrideEnv?: Record; } export async function startWebapp( @@ -144,6 +155,7 @@ export async function startWebapp( // to "0" so a local apps/webapp/.env that sets it to "1" doesn't // short-circuit the loader past the REQUIRE_PLUGINS check. ...(requirePlugins ? { REQUIRE_PLUGINS: "1", RBAC_FORCE_FALLBACK: "0" } : {}), + ...(options.overrideEnv ?? {}), NODE_PATH: nodePath, }, stdio: ["ignore", "pipe", "pipe"], diff --git a/knip.json b/knip.json index f992e77c80f..84456756ca1 100644 --- a/knip.json +++ b/knip.json @@ -19,6 +19,7 @@ "prisma/populate.ts", "scripts/**/*.{js,mjs,cjs,ts,mts,cts}", "test/**/*.producer.ts", + "test/bench/analyzeProfile.ts", "test/types/**/*.types.ts", "test/setup/global-e2e-full-setup.ts", "vite/node-globals-shim.js", From 0a572be224f2872b3353db17274688fa699fe212 Mon Sep 17 00:00:00 2001 From: Eric Allam Date: Thu, 20 Aug 2026 21:31:06 +0100 Subject: [PATCH 2/5] perf(webapp,run-engine): cut CPU on the engine-facing worker-action routes Three changes from a CPU profile of the worker-action request path, together taking on-CPU time per completed run from 9.07ms to 6.59ms (-27%) at ~300 req/s, with every worker-action p50 down 23-27%. Split the event-loop monitor in two. The blocked-loop detector installs an async_hooks hook that fires for every async resource the process creates, and enabling any async hook also puts V8 on the slow path for promise instrumentation process-wide; it measured ~14% of on-CPU time plus roughly half of all GC. It is now opt-in via EVENT_LOOP_MONITOR_ENABLED (default 0). The event-loop utilization gauge is a single interval timer with no per-request cost, so it moves to EVENT_LOOP_UTILIZATION_MONITOR_ENABLED (default 1) and stays on. Bucket route matching by first static path segment. The existing router patch removed the per-request re-flatten and regex rebuild, but matching was still a linear scan over the whole 521-route table. Route-matching self time drops 64%. Ordering is preserved exactly and equivalence was verified over 20,050 pathnames; apps/webapp/test/routeMatchingPatch.test.ts pins the semantics. Demote the per-heartbeat and per-dequeue info logs to debug. These are the two highest-rate engine calls and each wrote a synchronous structured log line on every request. --- .../reduce-webapp-cpu-on-worker-routes.md | 6 + apps/webapp/app/entry.server.tsx | 6 +- apps/webapp/app/env.server.ts | 3 +- apps/webapp/app/eventLoopMonitor.server.ts | 32 ++++- apps/webapp/test/routeMatchingPatch.test.ts | 119 ++++++++++++++++++ .../src/engine/systems/dequeueSystem.ts | 2 +- .../engine/systems/executionSnapshotSystem.ts | 2 +- patches/@remix-run__router@1.23.3.patch | 98 ++++++++++++++- patches/README.md | 28 ++++- pnpm-lock.yaml | 24 ++-- 10 files changed, 291 insertions(+), 29 deletions(-) create mode 100644 .server-changes/reduce-webapp-cpu-on-worker-routes.md create mode 100644 apps/webapp/test/routeMatchingPatch.test.ts diff --git a/.server-changes/reduce-webapp-cpu-on-worker-routes.md b/.server-changes/reduce-webapp-cpu-on-worker-routes.md new file mode 100644 index 00000000000..b23b7efbd39 --- /dev/null +++ b/.server-changes/reduce-webapp-cpu-on-worker-routes.md @@ -0,0 +1,6 @@ +--- +area: webapp +type: improvement +--- + +Cut webapp CPU usage by about a quarter on the routes that workers call most, freeing headroom at the same request rate. Detailed event-loop blocking diagnostics are now off by default (set `EVENT_LOOP_MONITOR_ENABLED=1` to restore them); the event-loop utilization metric is unaffected. diff --git a/apps/webapp/app/entry.server.tsx b/apps/webapp/app/entry.server.tsx index 5335aac90bc..074bc39d760 100644 --- a/apps/webapp/app/entry.server.tsx +++ b/apps/webapp/app/entry.server.tsx @@ -18,7 +18,7 @@ import type { OperatingSystemPlatform } from "./components/primitives/OperatingS import { OperatingSystemContextProvider } from "./components/primitives/OperatingSystemProvider"; import { assertRunOpsSplitSentinel, Prisma } from "./db.server"; import { env } from "./env.server"; -import { eventLoopMonitor } from "./eventLoopMonitor.server"; +import { eventLoopMonitor, eventLoopUtilizationMonitor } from "./eventLoopMonitor.server"; import { logger } from "./services/logger.server"; import { buildImgSrcDirective, parseCspImageOrigins, withImgSrc } from "./utils/cspImageOrigins"; import { singleton } from "./utils/singleton"; @@ -360,6 +360,10 @@ if (env.EVENT_LOOP_MONITOR_ENABLED === "1") { eventLoopMonitor.enable(); } +if (env.EVENT_LOOP_UTILIZATION_MONITOR_ENABLED === "1") { + eventLoopUtilizationMonitor.enable(); +} + if (remoteBuildsEnabled()) { console.log("🏗️ Remote builds enabled"); } else { diff --git a/apps/webapp/app/env.server.ts b/apps/webapp/app/env.server.ts index 0b86f83b492..c9179306124 100644 --- a/apps/webapp/app/env.server.ts +++ b/apps/webapp/app/env.server.ts @@ -995,7 +995,8 @@ const EnvironmentSchema = z CENTS_PER_RUN: z.coerce.number().default(0), - EVENT_LOOP_MONITOR_ENABLED: z.string().default("1"), + EVENT_LOOP_MONITOR_ENABLED: z.string().default("0"), + EVENT_LOOP_UTILIZATION_MONITOR_ENABLED: z.string().default("1"), MAXIMUM_LIVE_RELOADING_EVENTS: z.coerce.number().int().default(1000), MAXIMUM_TRACE_SUMMARY_VIEW_COUNT: z.coerce.number().int().default(25_000), MAXIMUM_TRACE_DETAILED_SUMMARY_VIEW_COUNT: z.coerce.number().int().default(10_000), diff --git a/apps/webapp/app/eventLoopMonitor.server.ts b/apps/webapp/app/eventLoopMonitor.server.ts index 2e45676e38b..0dd99b5e335 100644 --- a/apps/webapp/app/eventLoopMonitor.server.ts +++ b/apps/webapp/app/eventLoopMonitor.server.ts @@ -89,25 +89,47 @@ function after(asyncId: number) { } } +/** + * Per-async-resource blocked-loop detection. This is the expensive half: the + * hook fires for every async resource the process creates, and enabling any + * async hook also puts V8 on the slow path for promise instrumentation + * process-wide. On a request-heavy instance it costs roughly a seventh of all + * on-CPU time, which is why it is opt-in rather than on by default. + */ export const eventLoopMonitor = singleton("eventLoopMonitor", () => { const hook = createHook({ init, before, after, destroy }); - let stopEventLoopUtilizationMonitoring: () => void; - return { enable: () => { console.log("🥸 Initializing event loop monitor"); hook.enable(); - - stopEventLoopUtilizationMonitoring = startEventLoopUtilizationMonitoring(); }, disable: () => { console.log("🥸 Disabling event loop monitor"); hook.disable(); + }, + }; +}); + +/** + * The cheap half: a single interval timer reading `eventLoopUtilization()`. + * It costs nothing per request, so it stays on by default and is what a + * high-traffic instance should rely on when the async hook is too expensive. + */ +export const eventLoopUtilizationMonitor = singleton("eventLoopUtilizationMonitor", () => { + let stop: (() => void) | undefined; - stopEventLoopUtilizationMonitoring?.(); + return { + enable: () => { + console.log("🥸 Initializing event loop utilization monitor"); + + stop = startEventLoopUtilizationMonitoring(); + }, + disable: () => { + stop?.(); + stop = undefined; }, }; }); diff --git a/apps/webapp/test/routeMatchingPatch.test.ts b/apps/webapp/test/routeMatchingPatch.test.ts new file mode 100644 index 00000000000..772bcd8a94b --- /dev/null +++ b/apps/webapp/test/routeMatchingPatch.test.ts @@ -0,0 +1,119 @@ +import { matchRoutes, type RouteObject } from "@remix-run/router"; +import { describe, expect, it } from "vitest"; + +/** + * Guards `patches/@remix-run__router@1.23.3.patch`. + * + * The patch buckets ranked route branches by their first static path segment so + * a request only scans branches that could match it. That is a change to the + * matcher's search order, so these cases pin the behaviour that must survive + * it: branches whose leading segment is dynamic, splat or optional have to stay + * reachable from every pathname, case-insensitive matching has to keep working + * across the bucket lookup, and a more specific static route must still beat a + * dynamic one. + * + * If the patch is ever dropped, these should still pass against the stock + * matcher — they assert matching semantics, not the optimisation. + */ +const routes: RouteObject[] = [ + { + id: "root", + path: "/", + children: [ + { id: "index", index: true }, + { id: "engine-dequeue", path: "engine/v1/worker-actions/dequeue" }, + { + id: "engine-heartbeat", + path: "engine/v1/worker-actions/runs/:runFriendlyId/snapshots/:snapshotFriendlyId/heartbeat", + }, + { id: "engine-splat", path: "engine/*" }, + { id: "api-run", path: "api/v1/runs/:runId" }, + { id: "api-summary", path: "api/v1/runs/summary" }, + { id: "case-sensitive", path: "Engine/CaseCheck", caseSensitive: true }, + { id: "optional-lang", path: ":lang?/docs" }, + { id: "org-project", path: ":org/projects/:projectId" }, + { id: "orgs-settings", path: "orgs/settings" }, + { id: "orgs-dynamic", path: "orgs/:orgSlug" }, + { id: "catch-all", path: "*" }, + ], + }, +]; + +function matchedIds(pathname: string, basename?: string): string[] | null { + const matches = matchRoutes(routes, pathname, basename); + return matches ? matches.map((match) => match.route.id!) : null; +} + +function paramsFor(pathname: string): Record { + const matches = matchRoutes(routes, pathname); + return matches ? matches[matches.length - 1]!.params : {}; +} + +describe("route matching (patched matcher)", () => { + it("matches a fully static engine route", () => { + expect(matchedIds("/engine/v1/worker-actions/dequeue")).toEqual(["root", "engine-dequeue"]); + }); + + it("matches a dynamic engine route and extracts params", () => { + const pathname = "/engine/v1/worker-actions/runs/run_abc/snapshots/snap_def/heartbeat"; + expect(matchedIds(pathname)).toEqual(["root", "engine-heartbeat"]); + expect(paramsFor(pathname)).toMatchObject({ + runFriendlyId: "run_abc", + snapshotFriendlyId: "snap_def", + }); + }); + + it("prefers a static route over a dynamic sibling at the same depth", () => { + expect(matchedIds("/api/v1/runs/summary")).toEqual(["root", "api-summary"]); + expect(matchedIds("/api/v1/runs/run_abc")).toEqual(["root", "api-run"]); + expect(matchedIds("/orgs/settings")).toEqual(["root", "orgs-settings"]); + expect(matchedIds("/orgs/acme")).toEqual(["root", "orgs-dynamic"]); + }); + + it("falls back to a splat within the same first segment", () => { + expect(matchedIds("/engine/something/unrouted")).toEqual(["root", "engine-splat"]); + }); + + it("keeps routes with a dynamic first segment reachable", () => { + expect(matchedIds("/acme/projects/proj_1")).toEqual(["root", "org-project"]); + expect(paramsFor("/acme/projects/proj_1")).toMatchObject({ + org: "acme", + projectId: "proj_1", + }); + }); + + it("keeps routes with an optional first segment reachable both ways", () => { + expect(matchedIds("/docs")).toEqual(["root", "optional-lang"]); + expect(matchedIds("/en/docs")).toEqual(["root", "optional-lang"]); + }); + + it("matches case-insensitively by default", () => { + expect(matchedIds("/ENGINE/v1/worker-actions/dequeue")).toEqual(["root", "engine-dequeue"]); + expect(matchedIds("/API/v1/runs/summary")).toEqual(["root", "api-summary"]); + }); + + it("honours caseSensitive routes", () => { + expect(matchedIds("/Engine/CaseCheck")).toEqual(["root", "case-sensitive"]); + expect(matchedIds("/engine/casecheck")).toEqual(["root", "engine-splat"]); + }); + + it("falls through to the global catch-all for an unknown first segment", () => { + expect(matchedIds("/totally/unknown/path")).toEqual(["root", "catch-all"]); + }); + + it("matches the index route at the root", () => { + expect(matchedIds("/")).toEqual(["root", "index"]); + }); + + it("still strips a basename before matching", () => { + expect(matchedIds("/base/engine/v1/worker-actions/dequeue", "/base")).toEqual([ + "root", + "engine-dequeue", + ]); + expect(matchRoutes(routes, "/elsewhere/engine", "/base")).toBeNull(); + }); + + it("matches a percent-encoded first segment", () => { + expect(matchedIds("/%65ngine/v1/worker-actions/dequeue")).toEqual(["root", "engine-dequeue"]); + }); +}); diff --git a/internal-packages/run-engine/src/engine/systems/dequeueSystem.ts b/internal-packages/run-engine/src/engine/systems/dequeueSystem.ts index b1584cfb69f..a71d9a43eda 100644 --- a/internal-packages/run-engine/src/engine/systems/dequeueSystem.ts +++ b/internal-packages/run-engine/src/engine/systems/dequeueSystem.ts @@ -163,7 +163,7 @@ export class DequeueSystem { ? Math.max(0, Date.now() - message.message.eligibleAtMs) : undefined; - this.$.logger.info("DequeueSystem.dequeueFromWorkerQueue dequeued message", { + this.$.logger.debug("DequeueSystem.dequeueFromWorkerQueue dequeued message", { runId, orgId, environmentId: message.message.environmentId, diff --git a/internal-packages/run-engine/src/engine/systems/executionSnapshotSystem.ts b/internal-packages/run-engine/src/engine/systems/executionSnapshotSystem.ts index 48299ac220c..e79383a8bb6 100644 --- a/internal-packages/run-engine/src/engine/systems/executionSnapshotSystem.ts +++ b/internal-packages/run-engine/src/engine/systems/executionSnapshotSystem.ts @@ -567,7 +567,7 @@ export class ExecutionSnapshotSystem { }); } - this.$.logger.info("heartbeatRun snapshot heartbeat updated", { + this.$.logger.debug("heartbeatRun snapshot heartbeat updated", { id: latestSnapshot.id, runId: latestSnapshot.runId, lastHeartbeatAt: new Date(), diff --git a/patches/@remix-run__router@1.23.3.patch b/patches/@remix-run__router@1.23.3.patch index 10d1fdcc798..a1ac8f793d7 100644 --- a/patches/@remix-run__router@1.23.3.patch +++ b/patches/@remix-run__router@1.23.3.patch @@ -1,8 +1,8 @@ diff --git a/dist/router.cjs.js b/dist/router.cjs.js -index 6aa7db6fb5a7182afcdf17b16a3356abfa1e7945..95edbde228beff8dbd13fb2800302e31a932ef25 100644 +index 6aa7db6fb5a7182afcdf17b16a3356abfa1e7945..5ae0b0e632a356493c3a8b0c88ebd396e8f5305b 100644 --- a/dist/router.cjs.js +++ b/dist/router.cjs.js -@@ -783,6 +783,11 @@ function convertRoutesToDataRoutes(routes, mapRouteProperties, parentPath, manif +@@ -783,6 +783,51 @@ function convertRoutesToDataRoutes(routes, mapRouteProperties, parentPath, manif * * @see https://reactrouter.com/v6/utils/match-routes */ @@ -11,10 +11,50 @@ index 6aa7db6fb5a7182afcdf17b16a3356abfa1e7945..95edbde228beff8dbd13fb2800302e31 +// fix #14967; maintainer suggested patch-package until the Remix 3 route-pattern rewrite). +let __branchCache = new WeakMap(); +let __compileCache = new Map(); ++/** ++ * trigger.dev perf patch 2 — bucket ranked branches by their first static path ++ * segment so a request scans only branches that could match it, rather than the ++ * whole 500+ route table. See patches/README.md. ++ */ ++let __bucketCache = new WeakMap(); ++/** ++ * Returns the lowercased leading segment when it is static, or null when the ++ * branch can match any first segment (dynamic, splat or optional leading ++ * segment, or a root/pathless path) and so must always be considered. ++ */ ++function __firstStaticSegment(path) { ++ if (!path || path === "/") return null; ++ let start = path.charCodeAt(0) === 47 ? 1 : 0; ++ let end = path.indexOf("/", start); ++ let seg = end === -1 ? path.slice(start) : path.slice(start, end); ++ if (seg === "") return null; ++ if (seg.indexOf(":") !== -1 || seg.indexOf("*") !== -1 || seg.indexOf("(") !== -1 || seg.indexOf("?") !== -1) { ++ return null; ++ } ++ return seg.toLowerCase(); ++} ++function __buildBuckets(branches) { ++ let byFirstSegment = new Map(); ++ let always = []; ++ for (let i = 0; i < branches.length; ++i) { ++ let seg = __firstStaticSegment(branches[i].path); ++ if (seg === null) { ++ always.push(i); ++ continue; ++ } ++ let list = byFirstSegment.get(seg); ++ if (!list) { ++ list = []; ++ byFirstSegment.set(seg, list); ++ } ++ list.push(i); ++ } ++ return { byFirstSegment, always }; ++} function matchRoutes(routes, locationArg, basename) { if (basename === void 0) { basename = "/"; -@@ -795,8 +800,13 @@ function matchRoutesImpl(routes, locationArg, basename, allowPartial) { +@@ -795,18 +840,51 @@ function matchRoutesImpl(routes, locationArg, basename, allowPartial) { if (pathname == null) { return null; } @@ -29,8 +69,54 @@ index 6aa7db6fb5a7182afcdf17b16a3356abfa1e7945..95edbde228beff8dbd13fb2800302e31 + } let matches = null; let decoded = decodePath(pathname); - for (let i = 0; matches == null && i < branches.length; ++i) { -@@ -1115,6 +1125,12 @@ function compilePath(path, caseSensitive, end) { +- for (let i = 0; matches == null && i < branches.length; ++i) { +- // Incoming pathnames are generally encoded from either window.location +- // or from router.navigate, but we want to match against the unencoded +- // paths in the route definitions. Memory router locations won't be +- // encoded here but there also shouldn't be anything to decode so this +- // should be a safe operation. This avoids needing matchRoutes to be +- // history-aware. +- matches = matchRouteBranch(branches[i], decoded, allowPartial); ++ // Incoming pathnames are generally encoded from either window.location ++ // or from router.navigate, but we want to match against the unencoded ++ // paths in the route definitions. Memory router locations won't be ++ // encoded here but there also shouldn't be anything to decode so this ++ // should be a safe operation. This avoids needing matchRoutes to be ++ // history-aware. ++ let buckets = __bucketCache.get(branches); ++ if (!buckets) { ++ buckets = __buildBuckets(branches); ++ __bucketCache.set(branches, buckets); ++ } ++ let requestSegment = __firstStaticSegment(decoded); ++ if (requestSegment === null) { ++ for (let i = 0; matches == null && i < branches.length; ++i) { ++ matches = matchRouteBranch(branches[i], decoded, allowPartial); ++ } ++ return matches; ++ } ++ /** ++ * Both lists hold indexes into the already rank-sorted `branches`, so walking ++ * them in ascending-index order preserves the exact evaluation order the ++ * unbucketed scan would have used. ++ */ ++ let scoped = buckets.byFirstSegment.get(requestSegment); ++ let always = buckets.always; ++ let si = 0; ++ let ai = 0; ++ let scopedLength = scoped === undefined ? 0 : scoped.length; ++ while (matches == null && (si < scopedLength || ai < always.length)) { ++ let index; ++ if (si < scopedLength && (ai >= always.length || scoped[si] < always[ai])) { ++ index = scoped[si++]; ++ } else { ++ index = always[ai++]; ++ } ++ matches = matchRouteBranch(branches[index], decoded, allowPartial); + } + return matches; + } +@@ -1115,6 +1193,12 @@ function compilePath(path, caseSensitive, end) { if (end === void 0) { end = true; } @@ -43,7 +129,7 @@ index 6aa7db6fb5a7182afcdf17b16a3356abfa1e7945..95edbde228beff8dbd13fb2800302e31 warning(path === "*" || !path.endsWith("*") || path.endsWith("/*"), "Route path \"" + path + "\" will be treated as if it were " + ("\"" + path.replace(/\*$/, "/*") + "\" because the `*` character must ") + "always follow a `/` in the pattern. To get rid of this warning, " + ("please change the route path to \"" + path.replace(/\*$/, "/*") + "\".")); let params = []; let regexpSource = "^" + path.replace(/\/*\*?$/, "") // Ignore trailing / and /*, we'll handle it below -@@ -1147,7 +1163,11 @@ function compilePath(path, caseSensitive, end) { +@@ -1147,7 +1231,11 @@ function compilePath(path, caseSensitive, end) { regexpSource += "(?:(?=\\/|$))"; } else ; let matcher = new RegExp(regexpSource, caseSensitive ? undefined : "i"); diff --git a/patches/README.md b/patches/README.md index 5c6db4b9d62..14b815ba522 100644 --- a/patches/README.md +++ b/patches/README.md @@ -13,8 +13,8 @@ are documented below. ### What it does -Three changes to `matchRoutesImpl` / `compilePath`, all pure memoization of work that -depends only on the **static** route manifest: +Four changes to `matchRoutesImpl` / `compilePath`, all derived from work that depends only +on the **static** route manifest: 1. **Cache flattened + ranked branches per route-tree** (`WeakMap` keyed by the `routes` ref). `flattenRoutes()` + `rankRouteBranches()` were recomputed on *every* `matchRoutes` @@ -23,6 +23,24 @@ depends only on the **static** route manifest: was recomputed once per branch. 3. **Memoize `compilePath` compiled regexes** by `path|caseSensitive|end` (bounded `Map`, cap 2000). The matcher RegExp was rebuilt on every `matchPath` call. +4. **Bucket ranked branches by first static path segment** (`WeakMap` keyed by the branch + array). Even with (1)–(3), matching was still a linear scan calling `matchPath` on every + branch until one matched — O(route table) per request, now across 521 route files. + Branches are indexed by their lowercased leading segment, with one always-considered list + for branches whose leading segment is dynamic, splat or optional (and for root/pathless + paths). A request walks only its own bucket merged with that list. + + Ordering is preserved exactly: both lists hold indexes into the already rank-sorted + branch array and are walked in ascending-index order, so the first match found is the + same branch the full scan would have found. Bucketing lowercases on both sides, so + case-insensitive matching still resolves and `caseSensitive: true` routes are still + rejected by `matchPath` itself. A pathname whose own leading segment can't be bucketed + falls back to the full scan. + + Verified equivalent to the unpatched matcher over 20,050 pathnames (literal, dynamic, + splat, optional, case variants, basenames, percent-encoded). The semantics it depends on + are pinned by `apps/webapp/test/routeMatchingPatch.test.ts`, which asserts matching + behaviour rather than the optimisation, so it still passes without the patch. ### Why @@ -46,6 +64,12 @@ Measured on a single instance, same load, before vs after this patch: The realtime machinery itself (router/hydrate/serialize/diff) was ~0% — the bottleneck was entirely generic Remix request overhead. +Change (4) was added later, from a CPU audit of the engine-facing worker-action routes +(`apps/webapp/test/bench`). With (1)–(3) already in place, route matching was still 10.4% +of on-CPU time on that path — the residual linear scan rather than any recompilation. +Bucketing took route-matching self-time from 3.6s to 1.3s (**−64%**) over a 90s window at +~300 req/s. + ### Upstream status (why we patch instead of upgrade) This is a known, acknowledged inefficiency, and it is **only partially fixed in React diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 7d74cbbe905..bf4fea17902 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -75,7 +75,7 @@ patchedDependencies: hash: ba1a06f46256cdb8d6faf7167246692c0de2e7cd846a9dc0f13be0137e1c3745 path: patches/@kubernetes__client-node@1.0.0.patch '@remix-run/router@1.23.3': - hash: 33e2966e9ef36aa09955cec9922c5063227b83429157a71ba226369a08627ac9 + hash: 5fc6f6f85bbc0dc992a1c5e2052cb971e5ee1626eec16ff1370b235c45dcd627 path: patches/@remix-run__router@1.23.3.patch '@sentry/remix@9.46.0': hash: 146126b032581925294aaed63ab53ce3f5e0356a755f1763d7a9a76b9846943b @@ -446,7 +446,7 @@ importers: version: 2.17.5(react-dom@18.3.1(react@18.3.1))(react@18.3.1)(typescript@7.0.2) '@remix-run/router': specifier: ^1.23.3 - version: 1.23.3(patch_hash=33e2966e9ef36aa09955cec9922c5063227b83429157a71ba226369a08627ac9) + version: 1.23.3(patch_hash=5fc6f6f85bbc0dc992a1c5e2052cb971e5ee1626eec16ff1370b235c45dcd627) '@remix-run/server-runtime': specifier: 2.17.5 version: 2.17.5(typescript@7.0.2) @@ -725,7 +725,7 @@ importers: version: 0.3.1(@remix-run/react@2.17.5(react-dom@18.3.1(react@18.3.1))(react@18.3.1)(typescript@7.0.2))(@remix-run/server-runtime@2.17.5(typescript@7.0.2))(react@18.3.1) remix-utils: specifier: ^7.7.0 - version: 7.7.0(@remix-run/node@2.17.5(typescript@7.0.2))(@remix-run/react@2.17.5(react-dom@18.3.1(react@18.3.1))(react@18.3.1)(typescript@7.0.2))(@remix-run/router@1.23.3(patch_hash=33e2966e9ef36aa09955cec9922c5063227b83429157a71ba226369a08627ac9))(crypto-js@4.2.0)(intl-parse-accept-language@1.0.0)(react@18.3.1)(zod@3.25.76) + version: 7.7.0(@remix-run/node@2.17.5(typescript@7.0.2))(@remix-run/react@2.17.5(react-dom@18.3.1(react@18.3.1))(react@18.3.1)(typescript@7.0.2))(@remix-run/router@1.23.3(patch_hash=5fc6f6f85bbc0dc992a1c5e2052cb971e5ee1626eec16ff1370b235c45dcd627))(crypto-js@4.2.0)(intl-parse-accept-language@1.0.0)(react@18.3.1)(zod@3.25.76) semver: specifier: ^7.5.0 version: 7.8.1 @@ -20380,7 +20380,7 @@ snapshots: '@npmcli/package-json': 4.0.1 '@remix-run/node': 2.17.5(typescript@7.0.2) '@remix-run/react': 2.17.5(react-dom@18.3.1(react@18.3.1))(react@18.3.1)(typescript@7.0.2) - '@remix-run/router': 1.23.3(patch_hash=33e2966e9ef36aa09955cec9922c5063227b83429157a71ba226369a08627ac9) + '@remix-run/router': 1.23.3(patch_hash=5fc6f6f85bbc0dc992a1c5e2052cb971e5ee1626eec16ff1370b235c45dcd627) '@remix-run/server-runtime': 2.17.5(typescript@7.0.2) '@types/mdx': 2.0.5 '@vanilla-extract/integration': 6.2.1(@types/node@24.13.3)(lightningcss@1.32.0)(terser@5.46.1) @@ -20468,7 +20468,7 @@ snapshots: '@remix-run/react@2.17.5(react-dom@18.3.1(react@18.3.1))(react@18.3.1)(typescript@7.0.2)': dependencies: - '@remix-run/router': 1.23.3(patch_hash=33e2966e9ef36aa09955cec9922c5063227b83429157a71ba226369a08627ac9) + '@remix-run/router': 1.23.3(patch_hash=5fc6f6f85bbc0dc992a1c5e2052cb971e5ee1626eec16ff1370b235c45dcd627) '@remix-run/server-runtime': 2.17.5(typescript@7.0.2) react: 18.3.1 react-dom: 18.3.1(react@18.3.1) @@ -20478,7 +20478,7 @@ snapshots: optionalDependencies: typescript: 7.0.2 - '@remix-run/router@1.23.3(patch_hash=33e2966e9ef36aa09955cec9922c5063227b83429157a71ba226369a08627ac9)': {} + '@remix-run/router@1.23.3(patch_hash=5fc6f6f85bbc0dc992a1c5e2052cb971e5ee1626eec16ff1370b235c45dcd627)': {} '@remix-run/serve@2.17.5(typescript@7.0.2)': dependencies: @@ -20497,7 +20497,7 @@ snapshots: '@remix-run/server-runtime@2.17.5(typescript@7.0.2)': dependencies: - '@remix-run/router': 1.23.3(patch_hash=33e2966e9ef36aa09955cec9922c5063227b83429157a71ba226369a08627ac9) + '@remix-run/router': 1.23.3(patch_hash=5fc6f6f85bbc0dc992a1c5e2052cb971e5ee1626eec16ff1370b235c45dcd627) '@types/cookie': 0.6.0 '@web3-storage/multipart-parser': 1.0.0 cookie: 0.7.2 @@ -20826,7 +20826,7 @@ snapshots: '@opentelemetry/semantic-conventions': 1.41.1 '@remix-run/node': 2.17.5(typescript@7.0.2) '@remix-run/react': 2.17.5(react-dom@18.3.1(react@18.3.1))(react@18.3.1)(typescript@7.0.2) - '@remix-run/router': 1.23.3(patch_hash=33e2966e9ef36aa09955cec9922c5063227b83429157a71ba226369a08627ac9) + '@remix-run/router': 1.23.3(patch_hash=5fc6f6f85bbc0dc992a1c5e2052cb971e5ee1626eec16ff1370b235c45dcd627) '@remix-run/server-runtime': 2.17.5(typescript@7.0.2) '@sentry/cli': 2.50.2(encoding@0.1.13) '@sentry/core': 9.46.0 @@ -28150,14 +28150,14 @@ snapshots: react-router-dom@6.30.4(react-dom@18.3.1(react@18.3.1))(react@18.3.1): dependencies: - '@remix-run/router': 1.23.3(patch_hash=33e2966e9ef36aa09955cec9922c5063227b83429157a71ba226369a08627ac9) + '@remix-run/router': 1.23.3(patch_hash=5fc6f6f85bbc0dc992a1c5e2052cb971e5ee1626eec16ff1370b235c45dcd627) react: 18.3.1 react-dom: 18.3.1(react@18.3.1) react-router: 6.30.4(react@18.3.1) react-router@6.30.4(react@18.3.1): dependencies: - '@remix-run/router': 1.23.3(patch_hash=33e2966e9ef36aa09955cec9922c5063227b83429157a71ba226369a08627ac9) + '@remix-run/router': 1.23.3(patch_hash=5fc6f6f85bbc0dc992a1c5e2052cb971e5ee1626eec16ff1370b235c45dcd627) react: 18.3.1 react-smooth@4.0.4(react-dom@18.3.1(react@18.3.1))(react@18.3.1): @@ -28483,13 +28483,13 @@ snapshots: '@remix-run/server-runtime': 2.17.5(typescript@7.0.2) react: 18.3.1 - remix-utils@7.7.0(@remix-run/node@2.17.5(typescript@7.0.2))(@remix-run/react@2.17.5(react-dom@18.3.1(react@18.3.1))(react@18.3.1)(typescript@7.0.2))(@remix-run/router@1.23.3(patch_hash=33e2966e9ef36aa09955cec9922c5063227b83429157a71ba226369a08627ac9))(crypto-js@4.2.0)(intl-parse-accept-language@1.0.0)(react@18.3.1)(zod@3.25.76): + remix-utils@7.7.0(@remix-run/node@2.17.5(typescript@7.0.2))(@remix-run/react@2.17.5(react-dom@18.3.1(react@18.3.1))(react@18.3.1)(typescript@7.0.2))(@remix-run/router@1.23.3(patch_hash=5fc6f6f85bbc0dc992a1c5e2052cb971e5ee1626eec16ff1370b235c45dcd627))(crypto-js@4.2.0)(intl-parse-accept-language@1.0.0)(react@18.3.1)(zod@3.25.76): dependencies: type-fest: 4.33.0 optionalDependencies: '@remix-run/node': 2.17.5(typescript@7.0.2) '@remix-run/react': 2.17.5(react-dom@18.3.1(react@18.3.1))(react@18.3.1)(typescript@7.0.2) - '@remix-run/router': 1.23.3(patch_hash=33e2966e9ef36aa09955cec9922c5063227b83429157a71ba226369a08627ac9) + '@remix-run/router': 1.23.3(patch_hash=5fc6f6f85bbc0dc992a1c5e2052cb971e5ee1626eec16ff1370b235c45dcd627) crypto-js: 4.2.0 intl-parse-accept-language: 1.0.0 react: 18.3.1 From ffc997db31e0f57e09a5f7e379c31ed7b21546db Mon Sep 17 00:00:00 2001 From: Eric Allam Date: Thu, 20 Aug 2026 21:31:14 +0100 Subject: [PATCH 3/5] chore(docker): accept traces in the local otel collector The collector only defined a metrics pipeline, so pointing INTERNAL_OTEL_TRACE_EXPORTER_URL at it locally failed and the webapp fell back to the console span logger. Adds a traces pipeline with the debug exporter, which accepts and summarises spans - enough to exercise the whole client-side export path when profiling. --- docker/config/otel-collector-config.yaml | 21 ++++++++++++++++----- 1 file changed, 16 insertions(+), 5 deletions(-) diff --git a/docker/config/otel-collector-config.yaml b/docker/config/otel-collector-config.yaml index eab68960989..cbf6fe762b9 100644 --- a/docker/config/otel-collector-config.yaml +++ b/docker/config/otel-collector-config.yaml @@ -1,5 +1,8 @@ # OpenTelemetry Collector configuration for local development -# Receives OTLP metrics from the webapp and exposes them in Prometheus format +# Receives OTLP metrics and traces from the webapp. Metrics are exposed in +# Prometheus format; traces are counted by the debug exporter so you can point +# INTERNAL_OTEL_TRACE_EXPORTER_URL at a real local endpoint instead of falling +# back to the console span logger. receivers: otlp: @@ -23,9 +26,14 @@ exporters: resource_to_telemetry_conversion: enabled: true - # Debug exporter for troubleshooting (optional, uncomment to enable) - # debug: - # verbosity: detailed + # Traces are accepted and summarised rather than stored. That is enough to + # exercise the whole client-side export path (serialize -> HTTP -> accept), + # which is what a local run needs to measure. Raise to `detailed` to inspect + # individual spans. + debug: + verbosity: basic + sampling_initial: 5 + sampling_thereafter: 200 service: pipelines: @@ -33,4 +41,7 @@ service: receivers: [otlp] processors: [batch] exporters: [prometheus] - + traces: + receivers: [otlp] + processors: [batch] + exporters: [debug] From 733d0f7dd2196bc60216cbf400fac7aa4ee8dd9e Mon Sep 17 00:00:00 2001 From: Eric Allam Date: Fri, 21 Aug 2026 10:24:05 +0100 Subject: [PATCH 4/5] fix: address review feedback on the CPU benchmark harness Settle in-flight CDP requests when the inspector socket closes or errors, and guard the message parse. A webapp that exited mid-run previously left every send() pending, so the bench hung until the suite timeout with nothing explaining why; a parse throw inside the listener surfaced as an uncaughtException and tore down the runner. send() now also rejects instead of hanging when the socket is already closed. Unref both event-loop-utilization interval timers so a throw between start and stop cannot keep a vitest worker alive. Make eventLoopUtilizationMonitor.enable() idempotent: a second call previously started another interval and overwrote the only stored cleanup callback, leaking the first. Validate --top in the profile analyzer. A non-numeric value produced NaN, and slice(0, NaN) silently printed empty tables that looked like an empty profile. Keep the release note behavioural, without the env var name. --- .../reduce-webapp-cpu-on-worker-routes.md | 2 +- apps/webapp/app/eventLoopMonitor.server.ts | 4 +++ apps/webapp/test/bench/analyzeProfile.ts | 11 +++++-- apps/webapp/test/bench/lib/cdp.ts | 31 +++++++++++++++++-- .../src/engine/bench/inspectorProfiler.ts | 5 ++- 5 files changed, 47 insertions(+), 6 deletions(-) diff --git a/.server-changes/reduce-webapp-cpu-on-worker-routes.md b/.server-changes/reduce-webapp-cpu-on-worker-routes.md index b23b7efbd39..95a2ce05c30 100644 --- a/.server-changes/reduce-webapp-cpu-on-worker-routes.md +++ b/.server-changes/reduce-webapp-cpu-on-worker-routes.md @@ -3,4 +3,4 @@ area: webapp type: improvement --- -Cut webapp CPU usage by about a quarter on the routes that workers call most, freeing headroom at the same request rate. Detailed event-loop blocking diagnostics are now off by default (set `EVENT_LOOP_MONITOR_ENABLED=1` to restore them); the event-loop utilization metric is unaffected. +Cut webapp CPU usage by about a quarter on the routes that workers call most, freeing headroom at the same request rate. Detailed event-loop blocking traces are no longer recorded by default, because producing them was itself a large part of that cost. diff --git a/apps/webapp/app/eventLoopMonitor.server.ts b/apps/webapp/app/eventLoopMonitor.server.ts index 0dd99b5e335..5eb5ec82ceb 100644 --- a/apps/webapp/app/eventLoopMonitor.server.ts +++ b/apps/webapp/app/eventLoopMonitor.server.ts @@ -123,6 +123,10 @@ export const eventLoopUtilizationMonitor = singleton("eventLoopUtilizationMonito return { enable: () => { + if (stop) { + return; + } + console.log("🥸 Initializing event loop utilization monitor"); stop = startEventLoopUtilizationMonitoring(); diff --git a/apps/webapp/test/bench/analyzeProfile.ts b/apps/webapp/test/bench/analyzeProfile.ts index 8b1d0fdf428..42eeb9637f8 100644 --- a/apps/webapp/test/bench/analyzeProfile.ts +++ b/apps/webapp/test/bench/analyzeProfile.ts @@ -31,8 +31,15 @@ function parseArgs(argv: string[]): { for (let i = 0; i < argv.length; i++) { const arg = argv[i]!; - if (arg === "--top") top = Number(argv[++i]); - else if (arg === "--json") json = argv[++i]; + if (arg === "--top") { + const raw = argv[++i]; + const parsed = Number(raw); + if (!Number.isFinite(parsed) || parsed <= 0) { + console.error(`--top expects a positive number, got "${raw ?? ""}"`); + process.exit(1); + } + top = parsed; + } else if (arg === "--json") json = argv[++i]; else if (arg === "--root") root = resolve(argv[++i]!); else if (!arg.startsWith("--")) profilePath = arg; } diff --git a/apps/webapp/test/bench/lib/cdp.ts b/apps/webapp/test/bench/lib/cdp.ts index b708d02876c..5dd7523a7aa 100644 --- a/apps/webapp/test/bench/lib/cdp.ts +++ b/apps/webapp/test/bench/lib/cdp.ts @@ -68,10 +68,20 @@ class CdpSession { private nextId = 1; private pending = new Map void; reject: (e: Error) => void }>(); + /** + * Anything that ends the socket has to settle the in-flight requests. If the + * profiled webapp exits mid-run, an unsettled `send()` would otherwise hang + * until the suite-level timeout with nothing explaining why. + */ private constructor(ws: WebSocket) { this.ws = ws; this.ws.on("message", (data) => { - const msg = JSON.parse(data.toString()) as CdpMessage; + let msg: CdpMessage; + try { + msg = JSON.parse(data.toString()) as CdpMessage; + } catch { + return; + } if (msg.id === undefined) return; const waiter = this.pending.get(msg.id); if (!waiter) return; @@ -79,6 +89,16 @@ class CdpSession { if (msg.error) waiter.reject(new Error(`${msg.error.message} (${msg.error.code})`)); else waiter.resolve(msg.result); }); + + const rejectAll = (reason: string) => { + for (const waiter of this.pending.values()) { + waiter.reject(new Error(reason)); + } + this.pending.clear(); + }; + + this.ws.on("error", (err: Error) => rejectAll(`CDP socket error: ${err.message}`)); + this.ws.on("close", () => rejectAll("CDP socket closed before the response arrived")); } /** @@ -96,6 +116,10 @@ class CdpSession { } send(method: string, params: Record = {}): Promise { + if (this.ws.readyState !== WebSocket.OPEN) { + return Promise.reject(new Error(`CDP socket is not open, cannot send ${method}`)); + } + const id = this.nextId++; const promise = new Promise((resolve, reject) => { this.pending.set(id, { resolve, reject }); @@ -155,13 +179,16 @@ export class WebappProfiler { void this.evaluateElu(); - this.eluTimer = setInterval(() => { + const timer = setInterval(() => { void this.evaluateElu().then((utilization) => { if (utilization !== undefined) { this.eluSamples.push({ atMs: Date.now() - this.eluStartedAt, utilization }); } }); }, intervalMs); + + timer.unref(); + this.eluTimer = timer; } /** diff --git a/internal-packages/run-engine/src/engine/bench/inspectorProfiler.ts b/internal-packages/run-engine/src/engine/bench/inspectorProfiler.ts index 250598b3dcd..a4ff42d36f3 100644 --- a/internal-packages/run-engine/src/engine/bench/inspectorProfiler.ts +++ b/internal-packages/run-engine/src/engine/bench/inspectorProfiler.ts @@ -97,12 +97,15 @@ export class InProcessProfiler { this.eluSamples = []; this.lastElu = performance.eventLoopUtilization(); - this.eluTimer = setInterval(() => { + const timer = setInterval(() => { const current = performance.eventLoopUtilization(); const diff = performance.eventLoopUtilization(current, this.lastElu!); this.lastElu = current; this.eluSamples.push(Number.isFinite(diff.utilization) ? diff.utilization : 0); }, intervalMs); + + timer.unref(); + this.eluTimer = timer; } stopEluSampling(): EluStats { From aced94786b44ef98bf0f6b9774fee1671e7cfa69 Mon Sep 17 00:00:00 2001 From: Eric Allam Date: Fri, 21 Aug 2026 10:48:36 +0100 Subject: [PATCH 5/5] fix: keep the first measured ELU interval in the webapp bench The baseline reading was fired without awaiting it, so a sampling tick that landed before that round trip resolved saw no previous reading and recorded a zero delta. stopEluSampling() compensated by dropping the first sample unconditionally, which also discarded a genuine measured interval and could leave a short run with no samples at all. The baseline is now awaited before the interval starts, so every recorded sample is a real delta and the drop is gone. --- .../webapp/test/bench/engineHttp.bench.test.ts | 2 +- apps/webapp/test/bench/lib/cdp.ts | 18 ++++++++---------- 2 files changed, 9 insertions(+), 11 deletions(-) diff --git a/apps/webapp/test/bench/engineHttp.bench.test.ts b/apps/webapp/test/bench/engineHttp.bench.test.ts index d8f1e8deed5..0b85942539b 100644 --- a/apps/webapp/test/bench/engineHttp.bench.test.ts +++ b/apps/webapp/test/bench/engineHttp.bench.test.ts @@ -147,7 +147,7 @@ describe("engine worker-action HTTP CPU benchmark", () => { let emptyDequeues = 0; await profiler.startCpuProfile(SAMPLING_INTERVAL_US); - profiler.startEluSampling(); + await profiler.startEluSampling(); recorder.begin(); await runLoad({ diff --git a/apps/webapp/test/bench/lib/cdp.ts b/apps/webapp/test/bench/lib/cdp.ts index 5dd7523a7aa..b5c23cfe40a 100644 --- a/apps/webapp/test/bench/lib/cdp.ts +++ b/apps/webapp/test/bench/lib/cdp.ts @@ -169,15 +169,17 @@ export class WebappProfiler { } /** - * The first evaluate seeds a baseline reading so that the first recorded - * delta is measured from the moment sampling started rather than from - * process boot. + * Awaits a baseline reading before the interval starts, so the first recorded + * delta is measured from the moment sampling started rather than from process + * boot. Awaiting matters: the baseline is a round trip to the target, and a + * tick that landed before it resolved would report a zero delta and drag the + * average down. */ - startEluSampling(intervalMs = 250): void { + async startEluSampling(intervalMs = 250): Promise { this.eluSamples = []; this.eluStartedAt = Date.now(); - void this.evaluateElu(); + await this.evaluateElu(); const timer = setInterval(() => { void this.evaluateElu().then((utilization) => { @@ -216,17 +218,13 @@ export class WebappProfiler { } } - /** - * Drops the seeded first sample, which spans the gap between attach and the - * start of load and therefore reads artificially low. - */ stopEluSampling(): { stats: EluStats; samples: EluSample[] } { if (this.eluTimer) { clearInterval(this.eluTimer); this.eluTimer = null; } - const samples = this.eluSamples.slice(1); + const samples = this.eluSamples; if (samples.length === 0) { return { stats: { mean: 0, p50: 0, p95: 0, p99: 0, max: 0, sampleCount: 0 }, samples }; }