From 60d71da90e58215a6f56d14c5ebd0d2e559b7f9c Mon Sep 17 00:00:00 2001 From: Eric Allam Date: Fri, 21 Aug 2026 11:53:16 +0100 Subject: [PATCH] perf(webapp,run-engine): cut CPU on the engine-facing worker-action routes (#4746) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Cuts CPU on the `engine/v1/worker-actions/*` routes a managed supervisor calls, and adds the benchmark harness the numbers come from. Measured on a local stack: **on-CPU per completed run 9.07ms → 6.59ms (−27%)**, busy fraction 45.6% → 33.8%, with every worker-action p50 down 23–27%. Load was 5,000 runs / 24 virtual supervisors / 90s window / 30,120 requests / 0 errors. Query-count work from the same investigation is deliberately **not** here — it will follow as a separate PR. ## The three changes **1. Split the event-loop monitor in two (~14% of on-CPU, plus ~5pp of GC).** `eventLoopMonitor.server.ts` installs a global `async_hooks` hook: `init` writes a `Map` entry for *every* async resource the process creates, `before` calls `process.hrtime()` and `context.active()` on every one. Enabling any async hook also puts V8 on the slow path for promise instrumentation process-wide. `EVENT_LOOP_MONITOR_ENABLED` defaulted to `"1"`, so this was the shipping configuration. The blocked-loop detector is now opt-in (`EVENT_LOOP_MONITOR_ENABLED`, default `0`). The event-loop *utilization* gauge — a single interval timer with no per-request cost — moves to its own flag (`EVENT_LOOP_UTILIZATION_MONITOR_ENABLED`, default `1`) and stays on, so the useful half survives without the expensive half. A/B under identical load: | | monitor on | monitor off | change | |---|---|---|---| | on-CPU per run | 9.08ms | 7.25ms | −20% | | GC self time | 9.80% | 5.05% | −4.75pp | | dequeue p50 | 76.6ms | 62.8ms | −18% | | attempts/start p50 | 56.3ms | 43.5ms | −23% | **2. Bucket route matching by first static path segment (10.4% → 3.9% of on-CPU).** `patches/@remix-run__router@1.23.3.patch` already memoized flattened branches and compiled path regexes. What remained was the linear scan: `matchRouteBranch` walked the ranked branch list calling `matchPath` per branch across 521 route files, so every worker-action request paid a scan proportional to the whole route table. Branches are now 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. Route-matching self time dropped 64% (3.6s → 1.3s over a 90s window). 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) with zero mismatches. `apps/webapp/test/routeMatchingPatch.test.ts` pins the matching semantics rather than the optimisation, so it still passes without the patch. **3. Demote 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. Synchronous `console` writes can block the loop when stdout backs up, which costs more than the ~1.3% CPU share suggests. ## The harness Two benchmarks, neither in the default suite (they run for minutes, attach the V8 profiler, and report numbers rather than assert on them). See `apps/webapp/test/bench/README.md`. - `apps/webapp/test/bench/engineHttp.bench.test.ts` — spawns a real webapp against throwaway Postgres/Redis containers, seeds a production environment with a promoted managed deployment, and drives a closed-loop supervisor pool through the full lifecycle. Profiling runs over CDP rather than `--cpu-prof` so it covers only the measured window instead of being swamped by boot, and `performance.eventLoopUtilization()` is sampled *inside* the webapp process. - `internal-packages/run-engine/src/engine/bench/runEngineLifecycle.bench.test.ts` — drives `RunEngine` directly, profiling enqueue and lifecycle separately so engine cost isn't mixed with request-stack overhead. - `apps/webapp/test/bench/analyzeProfile.ts` — dependency-free `.cpuprofile` analyzer that symbolicates through the build's source maps and ranks CPU by package, self time and total time. Percentages are shares of on-CPU time (V8's `(idle)`/`(program)` excluded). `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. The local OTel collector gains a traces pipeline. It only defined a metrics pipeline, so pointing `INTERNAL_OTEL_TRACE_EXPORTER_URL` at it locally failed and the webapp silently fell back to the console span logger. ## Configuration For operators upgrading: - `EVENT_LOOP_MONITOR_ENABLED` (now defaults to `0`) — the per-async-resource blocked-loop detector. Set to `1` to restore the previous behaviour and keep emitting `event-loop-blocked` spans. - `EVENT_LOOP_UTILIZATION_MONITOR_ENABLED` (new, defaults to `1`) — the `nodejs.event_loop.utilization` gauge. Unchanged in behaviour; it just has its own flag now so it survives turning the detector off. ## Notes for review - `pnpm-lock.yaml` changes only because the router patch content changed, which changes its patch hash. - One thing the profile ruled out: with a real OTLP collector receiving spans, tracing costs ~1.7% of on-CPU at 100% sampling and ~0.8% at the production rate. Span shipping is not a hidden cost, so nothing here touches it. - Caveats on the numbers: a laptop, not production hardware, so DB and Redis *latency* are unrepresentative (client-side CPU is what's ranked); single webapp process; throughput varies ~5% run to run, which is why the claims rest on on-CPU per run rather than req/s. ## Verification - 20,050-pathname router equivalence check vs the unpatched matcher, zero mismatches - `apps/webapp/test/routeMatchingPatch.test.ts` (12 cases) passes - webapp e2e smoke suite (68 tests) passes through the patched router - run-engine suites covering the snapshot/attempt paths pass - `typecheck`, `format`, `lint`, `knip` clean --- .gitignore | 3 + .../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 | 36 +- apps/webapp/package.json | 3 +- apps/webapp/test/bench/README.md | 91 +++++ apps/webapp/test/bench/analyzeProfile.ts | 66 ++++ .../test/bench/engineHttp.bench.test.ts | 288 ++++++++++++++++ apps/webapp/test/bench/lib/cdp.ts | 252 ++++++++++++++ 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/test/routeMatchingPatch.test.ts | 119 +++++++ apps/webapp/vitest.bench.config.ts | 22 ++ apps/webapp/vitest.config.ts | 7 +- docker/config/otel-collector-config.yaml | 21 +- internal-packages/run-engine/package.json | 3 +- .../src/engine/bench/inspectorProfiler.ts | 133 ++++++++ .../bench/runEngineLifecycle.bench.test.ts | 313 +++++++++++++++++ .../src/engine/systems/dequeueSystem.ts | 2 +- .../engine/systems/executionSnapshotSystem.ts | 2 +- .../run-engine/vitest.bench.config.ts | 17 + internal-packages/run-engine/vitest.config.ts | 1 + .../testcontainers/src/webapp.ts | 12 + knip.json | 1 + patches/@remix-run__router@1.23.3.patch | 98 +++++- patches/README.md | 28 +- pnpm-lock.yaml | 24 +- 30 files changed, 2346 insertions(+), 37 deletions(-) create mode 100644 .server-changes/reduce-webapp-cpu-on-worker-routes.md 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/test/routeMatchingPatch.test.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 b11dded2b..3f49db35f 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/.server-changes/reduce-webapp-cpu-on-worker-routes.md b/.server-changes/reduce-webapp-cpu-on-worker-routes.md new file mode 100644 index 000000000..95a2ce05c --- /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 traces are no longer recorded by default, because producing them was itself a large part of that cost. diff --git a/apps/webapp/app/entry.server.tsx b/apps/webapp/app/entry.server.tsx index 5335aac90..074bc39d7 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 0b86f83b4..c91793061 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 2e45676e3..5eb5ec82c 100644 --- a/apps/webapp/app/eventLoopMonitor.server.ts +++ b/apps/webapp/app/eventLoopMonitor.server.ts @@ -89,25 +89,51 @@ 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(); + }, + }; +}); - stopEventLoopUtilizationMonitoring?.(); +/** + * 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; + + return { + enable: () => { + if (stop) { + return; + } + + console.log("🥸 Initializing event loop utilization monitor"); + + stop = startEventLoopUtilizationMonitoring(); + }, + disable: () => { + stop?.(); + stop = undefined; }, }; }); diff --git a/apps/webapp/package.json b/apps/webapp/package.json index cabd55d1d..788342f23 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 000000000..ecd2231bf --- /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 000000000..42eeb9637 --- /dev/null +++ b/apps/webapp/test/bench/analyzeProfile.ts @@ -0,0 +1,66 @@ +#!/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") { + 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; + } + + 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 000000000..0b8594253 --- /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); + await 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 000000000..b5c23cfe4 --- /dev/null +++ b/apps/webapp/test/bench/lib/cdp.ts @@ -0,0 +1,252 @@ +/** + * 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 }>(); + + /** + * 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) => { + 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; + this.pending.delete(msg.id); + 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")); + } + + /** + * 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 { + 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 }); + }); + 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 }; + } + + /** + * 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. + */ + async startEluSampling(intervalMs = 250): Promise { + this.eluSamples = []; + this.eluStartedAt = Date.now(); + + await this.evaluateElu(); + + 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; + } + + /** + * 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; + } + } + + stopEluSampling(): { stats: EluStats; samples: EluSample[] } { + if (this.eluTimer) { + clearInterval(this.eluTimer); + this.eluTimer = null; + } + + const samples = this.eluSamples; + 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 000000000..442cff6eb --- /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 000000000..eef6ccc8b --- /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 000000000..6412a935a --- /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 000000000..3a35d0c26 --- /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/test/routeMatchingPatch.test.ts b/apps/webapp/test/routeMatchingPatch.test.ts new file mode 100644 index 000000000..772bcd8a9 --- /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/apps/webapp/vitest.bench.config.ts b/apps/webapp/vitest.bench.config.ts new file mode 100644 index 000000000..a80ab5a31 --- /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 995c17428..1981ce26a 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/docker/config/otel-collector-config.yaml b/docker/config/otel-collector-config.yaml index eab689609..cbf6fe762 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] diff --git a/internal-packages/run-engine/package.json b/internal-packages/run-engine/package.json index 51bb0d6da..96ace3a0e 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 000000000..a4ff42d36 --- /dev/null +++ b/internal-packages/run-engine/src/engine/bench/inspectorProfiler.ts @@ -0,0 +1,133 @@ +/** + * 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(); + + 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 { + 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 000000000..eb6e8ace0 --- /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/src/engine/systems/dequeueSystem.ts b/internal-packages/run-engine/src/engine/systems/dequeueSystem.ts index b1584cfb6..a71d9a43e 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 48299ac22..e79383a8b 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/internal-packages/run-engine/vitest.bench.config.ts b/internal-packages/run-engine/vitest.bench.config.ts new file mode 100644 index 000000000..6210a0f0c --- /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 cb048f009..64e119d3d 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 2ff73b1b5..d8a233777 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 f992e77c8..84456756c 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", diff --git a/patches/@remix-run__router@1.23.3.patch b/patches/@remix-run__router@1.23.3.patch index 10d1fdcc7..a1ac8f793 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 5c6db4b9d..14b815ba5 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 7d74cbbe9..bf4fea179 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