From 2df2377fdb0e2e66e01dd13773a864a74c011a17 Mon Sep 17 00:00:00 2001 From: Eric Allam Date: Sun, 2 Aug 2026 10:16:47 +0100 Subject: [PATCH 1/8] perf(webapp): pad list filters, decouple db metrics from the engine Database list filters (`in` / `notIn`) expand to one bind parameter per element, so every distinct list length produced a different SQL statement and pushed other entries out of the pooler's prepared-statement cache. Values are now padded up to the next power of two by repeating the last element, which leaves results unchanged and collapses hundreds of statement shapes to a handful. Set DB_PAD_IN_LISTS=0 to disable. Separately, the metrics endpoint no longer depends on query-engine internals. Engine stats are an optional section rather than a prefix on the whole scrape body, so a failure there degrades one section instead of failing the entire endpoint. Query duration and error counts are now recorded from application code, keyed by database, datasource and operation. --- .../db-metrics-engine-independent.md | 6 + .server-changes/pad-prisma-in-list-filters.md | 6 + apps/webapp/app/db.server.ts | 37 ++++-- apps/webapp/app/routes/metrics.ts | 18 ++- .../webapp/app/utils/dbMetrics.server.test.ts | 124 ++++++++++++++++++ apps/webapp/app/utils/dbMetrics.server.ts | 109 +++++++++++++++ .../app/utils/padInLists.server.test.ts | 124 ++++++++++++++++++ apps/webapp/app/utils/padInLists.server.ts | 120 +++++++++++++++++ apps/webapp/app/v3/tracer.server.ts | 14 +- 9 files changed, 545 insertions(+), 13 deletions(-) create mode 100644 .server-changes/db-metrics-engine-independent.md create mode 100644 .server-changes/pad-prisma-in-list-filters.md create mode 100644 apps/webapp/app/utils/dbMetrics.server.test.ts create mode 100644 apps/webapp/app/utils/dbMetrics.server.ts create mode 100644 apps/webapp/app/utils/padInLists.server.test.ts create mode 100644 apps/webapp/app/utils/padInLists.server.ts diff --git a/.server-changes/db-metrics-engine-independent.md b/.server-changes/db-metrics-engine-independent.md new file mode 100644 index 00000000000..258de9aaa07 --- /dev/null +++ b/.server-changes/db-metrics-engine-independent.md @@ -0,0 +1,6 @@ +--- +area: webapp +type: improvement +--- + +The metrics endpoint keeps serving when database engine metrics are unavailable, and now reports query duration and error counts per database and per operation. diff --git a/.server-changes/pad-prisma-in-list-filters.md b/.server-changes/pad-prisma-in-list-filters.md new file mode 100644 index 00000000000..684548f83ab --- /dev/null +++ b/.server-changes/pad-prisma-in-list-filters.md @@ -0,0 +1,6 @@ +--- +area: webapp +type: improvement +--- + +Database queries that filter on a list of values now reuse cached query plans instead of forcing the database to re-plan for every distinct list length. diff --git a/apps/webapp/app/db.server.ts b/apps/webapp/app/db.server.ts index 97a3ae7bb53..9dfbe8921e5 100644 --- a/apps/webapp/app/db.server.ts +++ b/apps/webapp/app/db.server.ts @@ -30,6 +30,8 @@ import { DATASOURCE_CONTEXT_KEY, startActiveSpan } from "./v3/tracer.server"; import type { Span } from "@opentelemetry/api"; import { context, trace } from "@opentelemetry/api"; import { queryPerformanceMonitor } from "./utils/queryPerformanceMonitor.server"; +import { recordOperation } from "./utils/dbMetrics.server"; +import { padInLists } from "./utils/padInLists.server"; export type { PrismaTransactionClient, @@ -123,15 +125,22 @@ async function $transactionInner( export { Prisma }; -function tagDatasource(datasource: "writer" | "replica", client: T): T { +function tagDatasource( + datasource: "writer" | "replica", + client: T, + database: string = "control-plane" +): T { return client.$extends({ name: "datasource-tagger", query: { - $allOperations: ({ query, args }) => { + $allOperations: ({ query, args, model, operation }) => { trace.getActiveSpan()?.setAttribute("db.datasource", datasource); return context.with( context.active().setValue(DATASOURCE_CONTEXT_KEY, datasource), - async () => await query(args) + async () => + await recordOperation({ datasource, client: database }, operation, model, () => + query(padInLists(args)) + ) ); }, }, @@ -142,16 +151,20 @@ function tagDatasource(datasource: "writer" | "replica", // generated package — does not extend @trigger.dev/database.PrismaClient). function tagDatasourceRunOps( datasource: "writer" | "replica", - client: RunOpsPrismaClient + client: RunOpsPrismaClient, + database: string = "run-ops" ): RunOpsPrismaClient { return client.$extends({ name: "datasource-tagger", query: { - $allOperations: ({ query, args }) => { + $allOperations: ({ query, args, model, operation }) => { trace.getActiveSpan()?.setAttribute("db.datasource", datasource); return context.with( context.active().setValue(DATASOURCE_CONTEXT_KEY, datasource), - async () => await query(args) + async () => + await recordOperation({ datasource, client: database }, operation, model, () => + query(padInLists(args)) + ) ); }, }, @@ -295,7 +308,7 @@ const runOpsTopology: RunOpsTopology = singleton("runOpsTopology", () => { controlPlane: { writer: prisma, replica: $replica }, buildNewWriter: (url, clientType) => captureInfraErrorsRunOps( - tagDatasourceRunOps("writer", buildRunOpsWriterClient({ url, clientType })) + tagDatasourceRunOps("writer", buildRunOpsWriterClient({ url, clientType }), "run-ops-new") ), // Brand the run-ops replica (only built for a real replica URL) so routed replica reads stay // off the primary. When no replica URL is set, selectRunOpsTopology reuses the writer here — @@ -303,19 +316,23 @@ const runOpsTopology: RunOpsTopology = singleton("runOpsTopology", () => { buildNewReplica: (url, clientType) => markReadReplicaClient( captureInfraErrorsRunOps( - tagDatasourceRunOps("replica", buildRunOpsReplicaClient({ url, clientType })) + tagDatasourceRunOps( + "replica", + buildRunOpsReplicaClient({ url, clientType }), + "run-ops-new" + ) ) ), // Legacy client shares the exact control-plane wrapper stack (the legacy DB carries the full // control-plane schema); markReadReplicaClient only on a real replica URL, as with the NEW replica. buildLegacyWriter: (url, clientType) => captureInfrastructureErrors( - tagDatasource("writer", buildWriterClient({ url, clientType })) + tagDatasource("writer", buildWriterClient({ url, clientType }), "run-ops-legacy") ), buildLegacyReplica: (url, clientType) => markReadReplicaClient( captureInfrastructureErrors( - tagDatasource("replica", buildReplicaClient({ url, clientType })) + tagDatasource("replica", buildReplicaClient({ url, clientType }), "run-ops-legacy") ) ), } diff --git a/apps/webapp/app/routes/metrics.ts b/apps/webapp/app/routes/metrics.ts index 62d8befe5f6..1f824b9d62b 100644 --- a/apps/webapp/app/routes/metrics.ts +++ b/apps/webapp/app/routes/metrics.ts @@ -1,6 +1,22 @@ import type { LoaderFunctionArgs } from "@remix-run/server-runtime"; import { prisma } from "~/db.server"; import { metricsRegister } from "~/metrics.server"; +import { logger } from "~/services/logger.server"; + +/** + * Engine-provided metrics are optional. `$metrics` only exists on the Rust query engine — + * the Rust-free client throws `Method not implemented`, and Prisma 7 removes the feature. + * A throw here must not take the whole scrape down with it, so the section degrades to + * empty and the rest of the registry is still served. + */ +async function readEngineMetrics(): Promise { + try { + return (await prisma.$metrics.prometheus()).replace(/^\s*[\r\n]/gm, ""); + } catch (error) { + logger.debug("Prisma engine metrics unavailable, serving core metrics only", { error }); + return ""; + } +} export async function loader({ request }: LoaderFunctionArgs) { // If the TRIGGER_METRICS_AUTH_PASSWORD is set, we need to check if the request has the correct password in auth header @@ -14,7 +30,7 @@ export async function loader({ request }: LoaderFunctionArgs) { } // We need to remove empty lines from the prisma metrics, grafana doesn't like them - const prismaMetrics = (await prisma.$metrics.prometheus()).replace(/^\s*[\r\n]/gm, ""); + const prismaMetrics = await readEngineMetrics(); const coreMetrics = await metricsRegister.metrics(); // Order matters, core metrics end with `# EOF`, prisma metrics don't diff --git a/apps/webapp/app/utils/dbMetrics.server.test.ts b/apps/webapp/app/utils/dbMetrics.server.test.ts new file mode 100644 index 00000000000..a4678e50c63 --- /dev/null +++ b/apps/webapp/app/utils/dbMetrics.server.test.ts @@ -0,0 +1,124 @@ +import { metrics } from "@opentelemetry/api"; +import { + AggregationTemporality, + InMemoryMetricExporter, + MeterProvider, + PeriodicExportingMetricReader, +} from "@opentelemetry/sdk-metrics"; +import { beforeAll, describe, expect, it } from "vitest"; +import type { recordOperation as RecordOperation } from "./dbMetrics.server"; + +const exporter = new InMemoryMetricExporter(AggregationTemporality.CUMULATIVE); +const reader = new PeriodicExportingMetricReader({ + exporter, + exportIntervalMillis: 60_000, + exportTimeoutMillis: 10_000, +}); +const meterProvider = new MeterProvider({ readers: [reader] }); + +let recordOperation: typeof RecordOperation; + +type Point = { attributes: Record; value: unknown }; + +/** + * The in-memory exporter keeps one batch per flush, so a point recorded by a later test + * only appears in a later batch. Flatten every batch and take the most recent match. + */ +async function collect(): Promise> { + await reader.forceFlush(); + + const byName = new Map(); + + for (const resourceMetric of exporter.getMetrics()) { + for (const scopeMetric of resourceMetric.scopeMetrics) { + for (const metric of scopeMetric.metrics) { + const points = byName.get(metric.descriptor.name) ?? []; + points.push(...(metric.dataPoints as unknown as Point[])); + byName.set(metric.descriptor.name, points); + } + } + } + + return byName; +} + +function pointFor(all: Map, name: string, operation: string) { + const points = (all.get(name) ?? []).filter( + (point) => point.attributes["db.operation"] === operation + ); + + return points.at(-1); +} + +describe("dbMetrics", () => { + beforeAll(async () => { + metrics.disable(); + metrics.setGlobalMeterProvider(meterProvider); + + ({ recordOperation } = await import("./dbMetrics.server")); + }); + + it("records duration for a successful operation, tagged by datasource, client and operation", async () => { + const result = await recordOperation( + { datasource: "writer", client: "control-plane" }, + "findMany", + "TaskRun", + async () => "ok" + ); + + expect(result).toBe("ok"); + + const point = pointFor(await collect(), "db.client.operation.duration", "findMany"); + + expect(point).toBeDefined(); + expect(point!.attributes["db.datasource"]).toBe("writer"); + expect(point!.attributes["db.client"]).toBe("control-plane"); + }); + + it("omits the model attribute by default so series count stays bounded", async () => { + await recordOperation( + { datasource: "replica", client: "run-ops-new" }, + "count", + "TaskRun", + async () => 1 + ); + + const point = pointFor(await collect(), "db.client.operation.duration", "count"); + + expect(point).toBeDefined(); + expect(point!.attributes["db.model"]).toBeUndefined(); + }); + + it("counts a failure by Prisma error code, times it, and rethrows", async () => { + const failure = Object.assign(new Error("no connection available"), { code: "P2024" }); + + await expect( + recordOperation({ datasource: "writer", client: "control-plane" }, "create", "TaskRun", () => + Promise.reject(failure) + ) + ).rejects.toThrow("no connection available"); + + const all = await collect(); + + const errorPoint = pointFor(all, "db.client.operation.errors", "create"); + + expect(errorPoint).toBeDefined(); + expect(errorPoint!.attributes["db.error_code"]).toBe("P2024"); + expect(errorPoint!.value).toBe(1); + + expect(pointFor(all, "db.client.operation.duration", "create")).toBeDefined(); + }); + + it("labels a non-Prisma error as unknown rather than dropping the count", async () => { + await expect( + recordOperation({ datasource: "writer", client: "control-plane" }, "update", "TaskRun", () => + Promise.reject(new Error("boom")) + ) + ).rejects.toThrow("boom"); + + const point = pointFor(await collect(), "db.client.operation.errors", "update"); + + expect(point).toBeDefined(); + expect(point!.attributes["db.error_code"]).toBe("unknown"); + }); +}); diff --git a/apps/webapp/app/utils/dbMetrics.server.ts b/apps/webapp/app/utils/dbMetrics.server.ts new file mode 100644 index 00000000000..c189f06dd0d --- /dev/null +++ b/apps/webapp/app/utils/dbMetrics.server.ts @@ -0,0 +1,109 @@ +import { metrics, type Attributes } from "@opentelemetry/api"; + +/** + * Query-level database metrics that do not depend on the Prisma query engine. + * + * The pool metrics in tracer.server.ts read `prisma.$metrics`, which only the Rust + * engine implements — it throws `Method not implemented` on the Rust-free client, and + * the preview feature is gone entirely in Prisma 7. These instruments are recorded from + * our own client extension instead, so the signal survives an engine change. + * + * They do NOT replace the pool gauges: connections open/busy/idle can only come from + * whoever owns the pool, which is still the engine. The error counter is the part that + * carries pool pressure across the swap — P2024 is "could not get a connection in + * time", and it surfaces to application code either way. + */ + +const METER_NAME = "trigger.dev/db"; + +export type DbMetricAttributes = { + /** writer | replica — mirrors the db.datasource span attribute. */ + datasource: "writer" | "replica"; + /** Which logical database: control-plane, run-ops, and so on. */ + client: string; +}; + +type Instruments = { + duration: ReturnType["createHistogram"]>; + errors: ReturnType["createCounter"]>; +}; + +let instruments: Instruments | undefined; + +/** + * Resolves the instruments on first use rather than at module load. The MeterProvider is + * registered by tracer.server.ts; a meter taken before that resolves against the no-op + * provider and stays no-op for the life of the process. + */ +function getInstruments(): Instruments { + if (instruments) { + return instruments; + } + + const meter = metrics.getMeter(METER_NAME); + + instruments = { + duration: meter.createHistogram("db.client.operation.duration", { + description: "Duration of a Prisma client operation, measured in application code", + unit: "ms", + }), + errors: meter.createCounter("db.client.operation.errors", { + description: "Prisma client operations that threw, keyed by Prisma error code", + unit: "operations", + }), + }; + + return instruments; +} + +/** + * Off by default: model names multiply the series count by the schema's table count + * (325 on the control plane). Opt in only while debugging a specific model. + */ +const includeModel = process.env.DB_METRICS_INCLUDE_MODEL === "1"; + +function errorCode(error: unknown): string { + if (typeof error === "object" && error !== null && "code" in error) { + const code = (error as { code: unknown }).code; + if (typeof code === "string") { + return code; + } + } + return "unknown"; +} + +/** + * Times one Prisma operation and records it. Call from inside an existing + * `$allOperations` extension rather than adding a second `$extends` layer — each layer + * is another proxy on a hot path. + * + * Failures are timed as well as counted: a database that is timing out would otherwise + * look like a database serving no traffic. + */ +export async function recordOperation( + attributes: DbMetricAttributes, + operation: string, + model: string | undefined, + run: () => Promise +): Promise { + const { duration, errors } = getInstruments(); + + const base: Attributes = { + "db.datasource": attributes.datasource, + "db.client": attributes.client, + "db.operation": operation, + ...(includeModel && model ? { "db.model": model } : {}), + }; + + const startedAt = performance.now(); + + try { + const result = await run(); + duration.record(performance.now() - startedAt, base); + return result; + } catch (error) { + duration.record(performance.now() - startedAt, base); + errors.add(1, { ...base, "db.error_code": errorCode(error) }); + throw error; + } +} diff --git a/apps/webapp/app/utils/padInLists.server.test.ts b/apps/webapp/app/utils/padInLists.server.test.ts new file mode 100644 index 00000000000..9639d1ba1af --- /dev/null +++ b/apps/webapp/app/utils/padInLists.server.test.ts @@ -0,0 +1,124 @@ +import { describe, expect, it } from "vitest"; +import { padInLists } from "./padInLists.server"; + +describe("padInLists", () => { + it("pads an in list up to the next power of two by repeating the last element", () => { + const args = { where: { id: { in: ["a", "b", "c"] } } }; + + expect(padInLists(args)).toEqual({ where: { id: { in: ["a", "b", "c", "c"] } } }); + }); + + it("pads notIn as well", () => { + const args = { where: { status: { notIn: ["A", "B", "C", "D", "E"] } } }; + + expect((padInLists(args) as typeof args).where.status.notIn).toHaveLength(8); + }); + + it("collapses arity 1..300 to 10 distinct lengths", () => { + const lengths = new Set(); + + for (let arity = 1; arity <= 300; arity++) { + const values = Array.from({ length: arity }, (_, i) => `id-${i}`); + const result = padInLists({ where: { id: { in: values } } }) as { + where: { id: { in: string[] } }; + }; + lengths.add(result.where.id.in.length); + } + + expect(lengths.size).toBe(10); + }); + + it("leaves the args reference untouched when nothing needs padding", () => { + const args = { where: { id: { in: ["a", "b"] } }, take: 10 }; + + expect(padInLists(args)).toBe(args); + }); + + it("does not mutate the caller's object", () => { + const values = ["a", "b", "c"]; + const args = { where: { id: { in: values } } }; + + padInLists(args); + + expect(values).toEqual(["a", "b", "c"]); + expect(args.where.id.in).toBe(values); + }); + + it("leaves empty and single-element lists alone", () => { + expect(padInLists({ where: { id: { in: [] } } })).toEqual({ where: { id: { in: [] } } }); + expect(padInLists({ where: { id: { in: ["only"] } } })).toEqual({ + where: { id: { in: ["only"] } }, + }); + }); + + it("leaves lists longer than the 1024 cap alone", () => { + const values = Array.from({ length: 1500 }, (_, i) => `id-${i}`); + const result = padInLists({ where: { id: { in: values } } }) as { + where: { id: { in: string[] } }; + }; + + expect(result.where.id.in).toHaveLength(1500); + }); + + it("pads a list that lands exactly on the cap boundary", () => { + const values = Array.from({ length: 700 }, (_, i) => `id-${i}`); + const result = padInLists({ where: { id: { in: values } } }) as { + where: { id: { in: string[] } }; + }; + + expect(result.where.id.in).toHaveLength(1024); + }); + + it("skips lists holding non-primitive values", () => { + const values = [{ id: "a" }, { id: "b" }, { id: "c" }]; + const result = padInLists({ where: { OR: { in: values } } }) as { + where: { OR: { in: unknown[] } }; + }; + + expect(result.where.OR.in).toHaveLength(3); + }); + + it("pads numbers and bigints", () => { + const numbers = padInLists({ where: { n: { in: [1, 2, 3] } } }) as { + where: { n: { in: number[] } }; + }; + expect(numbers.where.n.in).toEqual([1, 2, 3, 3]); + + const bigints = padInLists({ where: { n: { in: [1n, 2n, 3n] } } }) as { + where: { n: { in: bigint[] } }; + }; + expect(bigints.where.n.in).toHaveLength(4); + }); + + it("preserves class instances rather than rebuilding them as plain objects", () => { + const createdAt = new Date("2026-08-01T00:00:00.000Z"); + const args = { where: { createdAt: { gte: createdAt }, id: { in: ["a", "b", "c"] } } }; + + const result = padInLists(args) as typeof args; + + expect(result.where.createdAt.gte).toBe(createdAt); + expect(result.where.createdAt.gte).toBeInstanceOf(Date); + }); + + it("pads inside nested boolean filters", () => { + const args = { + where: { AND: [{ id: { in: ["a", "b", "c"] } }, { status: { in: ["X", "Y", "Z"] } }] }, + }; + + const result = padInLists(args) as { + where: { AND: Array<{ id?: { in: string[] }; status?: { in: string[] } }> }; + }; + + expect(result.where.AND[0].id!.in).toHaveLength(4); + expect(result.where.AND[1].status!.in).toHaveLength(4); + }); + + it("stops descending past the depth limit", () => { + let node: Record = { id: { in: ["a", "b", "c"] } }; + for (let i = 0; i < 12; i++) { + node = { nested: node }; + } + + expect(() => padInLists(node)).not.toThrow(); + }); +}); diff --git a/apps/webapp/app/utils/padInLists.server.ts b/apps/webapp/app/utils/padInLists.server.ts new file mode 100644 index 00000000000..37cf4e5b21c --- /dev/null +++ b/apps/webapp/app/utils/padInLists.server.ts @@ -0,0 +1,120 @@ +/** + * Pads `in:` / `notIn:` filter arrays up to the next power of two so that a query shape + * stops minting a new prepared statement for every distinct list length. + * + * Prisma expands a list filter into one bind parameter per element, so + * `id: { in: [...] }` produces a different SQL string for every arity. On the run-graph + * batch loaders, where arity is the batch size, a single call site can produce hundreds of + * distinct statements and evict everything else from the pooler's prepared-statement cache + * (PlanetScale's default budget is 200). Measured on a local rig: arity 1..300 produced 300 + * distinct statements unpadded and 10 padded. + * + * Repeating the last element is semantically free — `IN` and `NOT IN` ignore duplicates — + * so results are unchanged. + */ + +const MAX_DEPTH = 8; + +/** + * Above this length the padding is not worth it: it is at worst 2x the bind parameters, so + * a large list would pay real bandwidth and parse cost to save a cache entry. Such lists + * are better served by a rewrite to `= ANY($1)`. + */ +const MAX_PADDED_LENGTH = 1024; + +const PADDED_KEYS = new Set(["in", "notIn"]); + +function isPaddableValue(value: unknown): boolean { + const type = typeof value; + return type === "string" || type === "number" || type === "bigint"; +} + +function padded(values: unknown[]): unknown[] | undefined { + const { length } = values; + + if (length < 2 || length > MAX_PADDED_LENGTH) { + return undefined; + } + + let target = 1; + while (target < length) { + target *= 2; + } + + if (target === length || target > MAX_PADDED_LENGTH) { + return undefined; + } + + if (!values.every(isPaddableValue)) { + return undefined; + } + + const result = values.slice(); + const last = values[length - 1]; + while (result.length < target) { + result.push(last); + } + + return result; +} + +/** + * Only plain objects are rebuilt. Class instances (Decimal, Date, Buffer, Prisma.sql, …) + * are returned untouched, because reconstructing them would drop their prototype. + */ +function isPlainObject(value: object): boolean { + const prototype = Object.getPrototypeOf(value); + return prototype === Object.prototype || prototype === null; +} + +function walk(node: unknown, depth: number): unknown { + if (depth > MAX_DEPTH || node === null || typeof node !== "object") { + return node; + } + + if (Array.isArray(node)) { + let changed = false; + const next = node.map((item) => { + const walked = walk(item, depth + 1); + changed ||= walked !== item; + return walked; + }); + return changed ? next : node; + } + + if (!isPlainObject(node)) { + return node; + } + + let changed = false; + const next: Record = {}; + + for (const [key, value] of Object.entries(node)) { + if (PADDED_KEYS.has(key) && Array.isArray(value)) { + const result = padded(value); + next[key] = result ?? value; + changed ||= result !== undefined; + continue; + } + + const walked = walk(value, depth + 1); + changed ||= walked !== value; + next[key] = walked; + } + + return changed ? next : node; +} + +/** + * Kill switch for a behaviour change on the query hot path. Set `DB_PAD_IN_LISTS=0` to + * fall straight through to the original args without a deploy. + */ +const enabled = process.env.DB_PAD_IN_LISTS !== "0"; + +/** + * Returns `args` unchanged (same reference) when there is nothing to pad, so the common + * path allocates nothing. + */ +export function padInLists(args: T): T { + return enabled ? (walk(args, 0) as T) : args; +} diff --git a/apps/webapp/app/v3/tracer.server.ts b/apps/webapp/app/v3/tracer.server.ts index 87e0877091f..a87abbb8567 100644 --- a/apps/webapp/app/v3/tracer.server.ts +++ b/apps/webapp/app/v3/tracer.server.ts @@ -475,7 +475,11 @@ function configurePrismaMetrics({ meter }: { meter: Meter }) { // Single helper so we hit Prisma only once per scrape --------------------- async function readPrismaMetrics() { - const metrics = await prisma.$metrics.json(); + const metrics = await prisma.$metrics.json().catch(() => undefined); + + if (!metrics) { + return undefined; + } // Extract counter values const counters: Record = {}; @@ -519,7 +523,13 @@ function configurePrismaMetrics({ meter }: { meter: Meter }) { meter.addBatchObservableCallback( async (res) => { - const { counters, gauges, histograms } = await readPrismaMetrics(); + const prismaMetrics = await readPrismaMetrics(); + + if (!prismaMetrics) { + return; + } + + const { counters, gauges, histograms } = prismaMetrics; // Observe counters res.observe(queriesTotal, counters.queriesTotal); From cedb54cf52c313b3d379db02b2fab57fa9777b34 Mon Sep 17 00:00:00 2001 From: Eric Allam Date: Mon, 3 Aug 2026 10:05:35 +0100 Subject: [PATCH 2/8] fix(webapp): read Prisma initialization error codes from errorCode PrismaClientInitializationError carries its code on errorCode, not code, so connection failures such as P1001 were counted as unknown. Those are the failures this counter most needs to name. --- .../webapp/app/utils/dbMetrics.server.test.ts | 20 +++++++++++++++++++ apps/webapp/app/utils/dbMetrics.server.ts | 15 ++++++++++---- 2 files changed, 31 insertions(+), 4 deletions(-) diff --git a/apps/webapp/app/utils/dbMetrics.server.test.ts b/apps/webapp/app/utils/dbMetrics.server.test.ts index a4678e50c63..81c9e214a70 100644 --- a/apps/webapp/app/utils/dbMetrics.server.test.ts +++ b/apps/webapp/app/utils/dbMetrics.server.test.ts @@ -109,6 +109,26 @@ describe("dbMetrics", () => { expect(pointFor(all, "db.client.operation.duration", "create")).toBeDefined(); }); + it("reads errorCode for initialization errors, which do not carry code", async () => { + const failure = Object.assign(new Error("Can't reach database server"), { + errorCode: "P1001", + }); + + await expect( + recordOperation( + { datasource: "replica", client: "run-ops-new" }, + "findFirst", + "TaskRun", + () => Promise.reject(failure) + ) + ).rejects.toThrow("Can't reach database server"); + + const point = pointFor(await collect(), "db.client.operation.errors", "findFirst"); + + expect(point).toBeDefined(); + expect(point!.attributes["db.error_code"]).toBe("P1001"); + }); + it("labels a non-Prisma error as unknown rather than dropping the count", async () => { await expect( recordOperation({ datasource: "writer", client: "control-plane" }, "update", "TaskRun", () => diff --git a/apps/webapp/app/utils/dbMetrics.server.ts b/apps/webapp/app/utils/dbMetrics.server.ts index c189f06dd0d..7beef9bf403 100644 --- a/apps/webapp/app/utils/dbMetrics.server.ts +++ b/apps/webapp/app/utils/dbMetrics.server.ts @@ -62,11 +62,18 @@ function getInstruments(): Instruments { */ const includeModel = process.env.DB_METRICS_INCLUDE_MODEL === "1"; +/** + * Prisma splits the error code across two property names: `PrismaClientKnownRequestError` + * carries `code`, while `PrismaClientInitializationError` carries `errorCode`. Both are + * worth labelling, and the initialization codes are the connection failures (P1001 and + * friends) this counter most needs to surface, so check `code` first and fall back. + */ function errorCode(error: unknown): string { - if (typeof error === "object" && error !== null && "code" in error) { - const code = (error as { code: unknown }).code; - if (typeof code === "string") { - return code; + if (typeof error === "object" && error !== null) { + const candidate = + (error as { code?: unknown }).code ?? (error as { errorCode?: unknown }).errorCode; + if (typeof candidate === "string") { + return candidate; } } return "unknown"; From 823c46dd217e2358a0794014730ad560470ce827 Mon Sep 17 00:00:00 2001 From: Eric Allam Date: Mon, 3 Aug 2026 10:15:14 +0100 Subject: [PATCH 3/8] refactor(webapp): remove the unused query-engine metrics The metrics endpoint exposed a block of query-engine counters and gauges that nothing consumes; database observability comes from the OpenTelemetry integration. Drops the engine metrics section, the observable gauges built on top of it, and the schema preview feature that enabled them. --- .../db-metrics-engine-independent.md | 6 - .../remove-prisma-engine-metrics.md | 6 + apps/webapp/app/db.server.ts | 36 +-- apps/webapp/app/routes/metrics.ts | 24 +- .../webapp/app/utils/dbMetrics.server.test.ts | 144 ------------ apps/webapp/app/utils/dbMetrics.server.ts | 116 --------- apps/webapp/app/v3/tracer.server.ts | 221 ------------------ .../database/prisma/schema.prisma | 1 - .../run-ops-database/prisma/schema.prisma | 1 - 9 files changed, 17 insertions(+), 538 deletions(-) delete mode 100644 .server-changes/db-metrics-engine-independent.md create mode 100644 .server-changes/remove-prisma-engine-metrics.md delete mode 100644 apps/webapp/app/utils/dbMetrics.server.test.ts delete mode 100644 apps/webapp/app/utils/dbMetrics.server.ts diff --git a/.server-changes/db-metrics-engine-independent.md b/.server-changes/db-metrics-engine-independent.md deleted file mode 100644 index 258de9aaa07..00000000000 --- a/.server-changes/db-metrics-engine-independent.md +++ /dev/null @@ -1,6 +0,0 @@ ---- -area: webapp -type: improvement ---- - -The metrics endpoint keeps serving when database engine metrics are unavailable, and now reports query duration and error counts per database and per operation. diff --git a/.server-changes/remove-prisma-engine-metrics.md b/.server-changes/remove-prisma-engine-metrics.md new file mode 100644 index 00000000000..f162aa3e6d2 --- /dev/null +++ b/.server-changes/remove-prisma-engine-metrics.md @@ -0,0 +1,6 @@ +--- +area: webapp +type: improvement +--- + +Remove the unused query-engine metrics from the metrics endpoint. Database observability continues through the existing OpenTelemetry integration. diff --git a/apps/webapp/app/db.server.ts b/apps/webapp/app/db.server.ts index 9dfbe8921e5..df4ddfb4e98 100644 --- a/apps/webapp/app/db.server.ts +++ b/apps/webapp/app/db.server.ts @@ -30,7 +30,6 @@ import { DATASOURCE_CONTEXT_KEY, startActiveSpan } from "./v3/tracer.server"; import type { Span } from "@opentelemetry/api"; import { context, trace } from "@opentelemetry/api"; import { queryPerformanceMonitor } from "./utils/queryPerformanceMonitor.server"; -import { recordOperation } from "./utils/dbMetrics.server"; import { padInLists } from "./utils/padInLists.server"; export type { @@ -125,22 +124,15 @@ async function $transactionInner( export { Prisma }; -function tagDatasource( - datasource: "writer" | "replica", - client: T, - database: string = "control-plane" -): T { +function tagDatasource(datasource: "writer" | "replica", client: T): T { return client.$extends({ name: "datasource-tagger", query: { - $allOperations: ({ query, args, model, operation }) => { + $allOperations: ({ query, args }) => { trace.getActiveSpan()?.setAttribute("db.datasource", datasource); return context.with( context.active().setValue(DATASOURCE_CONTEXT_KEY, datasource), - async () => - await recordOperation({ datasource, client: database }, operation, model, () => - query(padInLists(args)) - ) + async () => await query(padInLists(args)) ); }, }, @@ -151,20 +143,16 @@ function tagDatasource( // generated package — does not extend @trigger.dev/database.PrismaClient). function tagDatasourceRunOps( datasource: "writer" | "replica", - client: RunOpsPrismaClient, - database: string = "run-ops" + client: RunOpsPrismaClient ): RunOpsPrismaClient { return client.$extends({ name: "datasource-tagger", query: { - $allOperations: ({ query, args, model, operation }) => { + $allOperations: ({ query, args }) => { trace.getActiveSpan()?.setAttribute("db.datasource", datasource); return context.with( context.active().setValue(DATASOURCE_CONTEXT_KEY, datasource), - async () => - await recordOperation({ datasource, client: database }, operation, model, () => - query(padInLists(args)) - ) + async () => await query(padInLists(args)) ); }, }, @@ -308,7 +296,7 @@ const runOpsTopology: RunOpsTopology = singleton("runOpsTopology", () => { controlPlane: { writer: prisma, replica: $replica }, buildNewWriter: (url, clientType) => captureInfraErrorsRunOps( - tagDatasourceRunOps("writer", buildRunOpsWriterClient({ url, clientType }), "run-ops-new") + tagDatasourceRunOps("writer", buildRunOpsWriterClient({ url, clientType })) ), // Brand the run-ops replica (only built for a real replica URL) so routed replica reads stay // off the primary. When no replica URL is set, selectRunOpsTopology reuses the writer here — @@ -316,23 +304,19 @@ const runOpsTopology: RunOpsTopology = singleton("runOpsTopology", () => { buildNewReplica: (url, clientType) => markReadReplicaClient( captureInfraErrorsRunOps( - tagDatasourceRunOps( - "replica", - buildRunOpsReplicaClient({ url, clientType }), - "run-ops-new" - ) + tagDatasourceRunOps("replica", buildRunOpsReplicaClient({ url, clientType })) ) ), // Legacy client shares the exact control-plane wrapper stack (the legacy DB carries the full // control-plane schema); markReadReplicaClient only on a real replica URL, as with the NEW replica. buildLegacyWriter: (url, clientType) => captureInfrastructureErrors( - tagDatasource("writer", buildWriterClient({ url, clientType }), "run-ops-legacy") + tagDatasource("writer", buildWriterClient({ url, clientType })) ), buildLegacyReplica: (url, clientType) => markReadReplicaClient( captureInfrastructureErrors( - tagDatasource("replica", buildReplicaClient({ url, clientType }), "run-ops-legacy") + tagDatasource("replica", buildReplicaClient({ url, clientType })) ) ), } diff --git a/apps/webapp/app/routes/metrics.ts b/apps/webapp/app/routes/metrics.ts index 1f824b9d62b..042b18d07bd 100644 --- a/apps/webapp/app/routes/metrics.ts +++ b/apps/webapp/app/routes/metrics.ts @@ -1,22 +1,5 @@ import type { LoaderFunctionArgs } from "@remix-run/server-runtime"; -import { prisma } from "~/db.server"; import { metricsRegister } from "~/metrics.server"; -import { logger } from "~/services/logger.server"; - -/** - * Engine-provided metrics are optional. `$metrics` only exists on the Rust query engine — - * the Rust-free client throws `Method not implemented`, and Prisma 7 removes the feature. - * A throw here must not take the whole scrape down with it, so the section degrades to - * empty and the rest of the registry is still served. - */ -async function readEngineMetrics(): Promise { - try { - return (await prisma.$metrics.prometheus()).replace(/^\s*[\r\n]/gm, ""); - } catch (error) { - logger.debug("Prisma engine metrics unavailable, serving core metrics only", { error }); - return ""; - } -} export async function loader({ request }: LoaderFunctionArgs) { // If the TRIGGER_METRICS_AUTH_PASSWORD is set, we need to check if the request has the correct password in auth header @@ -29,12 +12,7 @@ export async function loader({ request }: LoaderFunctionArgs) { } } - // We need to remove empty lines from the prisma metrics, grafana doesn't like them - const prismaMetrics = await readEngineMetrics(); - const coreMetrics = await metricsRegister.metrics(); - - // Order matters, core metrics end with `# EOF`, prisma metrics don't - const metrics = prismaMetrics + coreMetrics; + const metrics = await metricsRegister.metrics(); return new Response(metrics, { headers: { diff --git a/apps/webapp/app/utils/dbMetrics.server.test.ts b/apps/webapp/app/utils/dbMetrics.server.test.ts deleted file mode 100644 index 81c9e214a70..00000000000 --- a/apps/webapp/app/utils/dbMetrics.server.test.ts +++ /dev/null @@ -1,144 +0,0 @@ -import { metrics } from "@opentelemetry/api"; -import { - AggregationTemporality, - InMemoryMetricExporter, - MeterProvider, - PeriodicExportingMetricReader, -} from "@opentelemetry/sdk-metrics"; -import { beforeAll, describe, expect, it } from "vitest"; -import type { recordOperation as RecordOperation } from "./dbMetrics.server"; - -const exporter = new InMemoryMetricExporter(AggregationTemporality.CUMULATIVE); -const reader = new PeriodicExportingMetricReader({ - exporter, - exportIntervalMillis: 60_000, - exportTimeoutMillis: 10_000, -}); -const meterProvider = new MeterProvider({ readers: [reader] }); - -let recordOperation: typeof RecordOperation; - -type Point = { attributes: Record; value: unknown }; - -/** - * The in-memory exporter keeps one batch per flush, so a point recorded by a later test - * only appears in a later batch. Flatten every batch and take the most recent match. - */ -async function collect(): Promise> { - await reader.forceFlush(); - - const byName = new Map(); - - for (const resourceMetric of exporter.getMetrics()) { - for (const scopeMetric of resourceMetric.scopeMetrics) { - for (const metric of scopeMetric.metrics) { - const points = byName.get(metric.descriptor.name) ?? []; - points.push(...(metric.dataPoints as unknown as Point[])); - byName.set(metric.descriptor.name, points); - } - } - } - - return byName; -} - -function pointFor(all: Map, name: string, operation: string) { - const points = (all.get(name) ?? []).filter( - (point) => point.attributes["db.operation"] === operation - ); - - return points.at(-1); -} - -describe("dbMetrics", () => { - beforeAll(async () => { - metrics.disable(); - metrics.setGlobalMeterProvider(meterProvider); - - ({ recordOperation } = await import("./dbMetrics.server")); - }); - - it("records duration for a successful operation, tagged by datasource, client and operation", async () => { - const result = await recordOperation( - { datasource: "writer", client: "control-plane" }, - "findMany", - "TaskRun", - async () => "ok" - ); - - expect(result).toBe("ok"); - - const point = pointFor(await collect(), "db.client.operation.duration", "findMany"); - - expect(point).toBeDefined(); - expect(point!.attributes["db.datasource"]).toBe("writer"); - expect(point!.attributes["db.client"]).toBe("control-plane"); - }); - - it("omits the model attribute by default so series count stays bounded", async () => { - await recordOperation( - { datasource: "replica", client: "run-ops-new" }, - "count", - "TaskRun", - async () => 1 - ); - - const point = pointFor(await collect(), "db.client.operation.duration", "count"); - - expect(point).toBeDefined(); - expect(point!.attributes["db.model"]).toBeUndefined(); - }); - - it("counts a failure by Prisma error code, times it, and rethrows", async () => { - const failure = Object.assign(new Error("no connection available"), { code: "P2024" }); - - await expect( - recordOperation({ datasource: "writer", client: "control-plane" }, "create", "TaskRun", () => - Promise.reject(failure) - ) - ).rejects.toThrow("no connection available"); - - const all = await collect(); - - const errorPoint = pointFor(all, "db.client.operation.errors", "create"); - - expect(errorPoint).toBeDefined(); - expect(errorPoint!.attributes["db.error_code"]).toBe("P2024"); - expect(errorPoint!.value).toBe(1); - - expect(pointFor(all, "db.client.operation.duration", "create")).toBeDefined(); - }); - - it("reads errorCode for initialization errors, which do not carry code", async () => { - const failure = Object.assign(new Error("Can't reach database server"), { - errorCode: "P1001", - }); - - await expect( - recordOperation( - { datasource: "replica", client: "run-ops-new" }, - "findFirst", - "TaskRun", - () => Promise.reject(failure) - ) - ).rejects.toThrow("Can't reach database server"); - - const point = pointFor(await collect(), "db.client.operation.errors", "findFirst"); - - expect(point).toBeDefined(); - expect(point!.attributes["db.error_code"]).toBe("P1001"); - }); - - it("labels a non-Prisma error as unknown rather than dropping the count", async () => { - await expect( - recordOperation({ datasource: "writer", client: "control-plane" }, "update", "TaskRun", () => - Promise.reject(new Error("boom")) - ) - ).rejects.toThrow("boom"); - - const point = pointFor(await collect(), "db.client.operation.errors", "update"); - - expect(point).toBeDefined(); - expect(point!.attributes["db.error_code"]).toBe("unknown"); - }); -}); diff --git a/apps/webapp/app/utils/dbMetrics.server.ts b/apps/webapp/app/utils/dbMetrics.server.ts deleted file mode 100644 index 7beef9bf403..00000000000 --- a/apps/webapp/app/utils/dbMetrics.server.ts +++ /dev/null @@ -1,116 +0,0 @@ -import { metrics, type Attributes } from "@opentelemetry/api"; - -/** - * Query-level database metrics that do not depend on the Prisma query engine. - * - * The pool metrics in tracer.server.ts read `prisma.$metrics`, which only the Rust - * engine implements — it throws `Method not implemented` on the Rust-free client, and - * the preview feature is gone entirely in Prisma 7. These instruments are recorded from - * our own client extension instead, so the signal survives an engine change. - * - * They do NOT replace the pool gauges: connections open/busy/idle can only come from - * whoever owns the pool, which is still the engine. The error counter is the part that - * carries pool pressure across the swap — P2024 is "could not get a connection in - * time", and it surfaces to application code either way. - */ - -const METER_NAME = "trigger.dev/db"; - -export type DbMetricAttributes = { - /** writer | replica — mirrors the db.datasource span attribute. */ - datasource: "writer" | "replica"; - /** Which logical database: control-plane, run-ops, and so on. */ - client: string; -}; - -type Instruments = { - duration: ReturnType["createHistogram"]>; - errors: ReturnType["createCounter"]>; -}; - -let instruments: Instruments | undefined; - -/** - * Resolves the instruments on first use rather than at module load. The MeterProvider is - * registered by tracer.server.ts; a meter taken before that resolves against the no-op - * provider and stays no-op for the life of the process. - */ -function getInstruments(): Instruments { - if (instruments) { - return instruments; - } - - const meter = metrics.getMeter(METER_NAME); - - instruments = { - duration: meter.createHistogram("db.client.operation.duration", { - description: "Duration of a Prisma client operation, measured in application code", - unit: "ms", - }), - errors: meter.createCounter("db.client.operation.errors", { - description: "Prisma client operations that threw, keyed by Prisma error code", - unit: "operations", - }), - }; - - return instruments; -} - -/** - * Off by default: model names multiply the series count by the schema's table count - * (325 on the control plane). Opt in only while debugging a specific model. - */ -const includeModel = process.env.DB_METRICS_INCLUDE_MODEL === "1"; - -/** - * Prisma splits the error code across two property names: `PrismaClientKnownRequestError` - * carries `code`, while `PrismaClientInitializationError` carries `errorCode`. Both are - * worth labelling, and the initialization codes are the connection failures (P1001 and - * friends) this counter most needs to surface, so check `code` first and fall back. - */ -function errorCode(error: unknown): string { - if (typeof error === "object" && error !== null) { - const candidate = - (error as { code?: unknown }).code ?? (error as { errorCode?: unknown }).errorCode; - if (typeof candidate === "string") { - return candidate; - } - } - return "unknown"; -} - -/** - * Times one Prisma operation and records it. Call from inside an existing - * `$allOperations` extension rather than adding a second `$extends` layer — each layer - * is another proxy on a hot path. - * - * Failures are timed as well as counted: a database that is timing out would otherwise - * look like a database serving no traffic. - */ -export async function recordOperation( - attributes: DbMetricAttributes, - operation: string, - model: string | undefined, - run: () => Promise -): Promise { - const { duration, errors } = getInstruments(); - - const base: Attributes = { - "db.datasource": attributes.datasource, - "db.client": attributes.client, - "db.operation": operation, - ...(includeModel && model ? { "db.model": model } : {}), - }; - - const startedAt = performance.now(); - - try { - const result = await run(); - duration.record(performance.now() - startedAt, base); - return result; - } catch (error) { - duration.record(performance.now() - startedAt, base); - errors.add(1, { ...base, "db.error_code": errorCode(error) }); - throw error; - } -} diff --git a/apps/webapp/app/v3/tracer.server.ts b/apps/webapp/app/v3/tracer.server.ts index a87abbb8567..53341b12bc6 100644 --- a/apps/webapp/app/v3/tracer.server.ts +++ b/apps/webapp/app/v3/tracer.server.ts @@ -58,9 +58,7 @@ import { LoggerSpanExporter } from "./telemetry/loggerExporter.server"; import { CompactMetricExporter } from "./telemetry/compactMetricExporter.server"; import { logger } from "~/services/logger.server"; import { flattenAttributes } from "@trigger.dev/core/v3"; -import { prisma } from "~/db.server"; import { metricsRegister } from "~/metrics.server"; -import type { Prisma } from "@trigger.dev/database"; import { performance } from "node:perf_hooks"; export const SEMINTATTRS_FORCE_RECORDING = "forceRecording"; @@ -376,231 +374,12 @@ function setupMetrics() { const meter = meterProvider.getMeter("trigger.dev", "3.3.12"); - configurePrismaMetrics({ meter }); configureNodejsMetrics({ meter }); configureHostMetrics({ meterProvider }); return meter; } -function configurePrismaMetrics({ meter }: { meter: Meter }) { - // Counters - const queriesTotal = meter.createObservableCounter("db.client.queries.total", { - description: "Total number of Prisma Client queries executed", - unit: "queries", - }); - const datasourceQueriesTotal = meter.createObservableCounter("db.datasource.queries.total", { - description: "Total number of datasource queries executed", - unit: "queries", - }); - const connectionsOpenedTotal = meter.createObservableCounter("db.pool.connections.opened.total", { - description: "Total number of pool connections opened", - unit: "connections", - }); - const connectionsClosedTotal = meter.createObservableCounter("db.pool.connections.closed.total", { - description: "Total number of pool connections closed", - unit: "connections", - }); - - // Gauges - const queriesActive = meter.createObservableGauge("db.client.queries.active", { - description: "Number of currently active Prisma Client queries", - unit: "queries", - }); - const queriesWait = meter.createObservableGauge("db.client.queries.wait", { - description: "Number of queries currently waiting for a connection", - unit: "queries", - }); - const totalGauge = meter.createObservableGauge("db.pool.connections.total", { - description: "Open Prisma-pool connections", - unit: "connections", - }); - const busyGauge = meter.createObservableGauge("db.pool.connections.busy", { - description: "Connections currently executing queries", - unit: "connections", - }); - const freeGauge = meter.createObservableGauge("db.pool.connections.free", { - description: "Idle (free) connections in the pool", - unit: "connections", - }); - - // Histogram statistics as gauges - const queriesWaitTimeCount = meter.createObservableGauge("db.client.queries.wait_time.count", { - description: "Number of wait time observations", - unit: "observations", - }); - const queriesWaitTimeSum = meter.createObservableGauge("db.client.queries.wait_time.sum", { - description: "Total wait time across all observations", - unit: "ms", - }); - const queriesWaitTimeMean = meter.createObservableGauge("db.client.queries.wait_time.mean", { - description: "Average wait time for a connection", - unit: "ms", - }); - - const queriesDurationCount = meter.createObservableGauge("db.client.queries.duration.count", { - description: "Number of query duration observations", - unit: "observations", - }); - const queriesDurationSum = meter.createObservableGauge("db.client.queries.duration.sum", { - description: "Total query duration across all observations", - unit: "ms", - }); - const queriesDurationMean = meter.createObservableGauge("db.client.queries.duration.mean", { - description: "Average duration of Prisma Client queries", - unit: "ms", - }); - - const datasourceQueriesDurationCount = meter.createObservableGauge( - "db.datasource.queries.duration.count", - { - description: "Number of datasource query duration observations", - unit: "observations", - } - ); - const datasourceQueriesDurationSum = meter.createObservableGauge( - "db.datasource.queries.duration.sum", - { - description: "Total datasource query duration across all observations", - unit: "ms", - } - ); - const datasourceQueriesDurationMean = meter.createObservableGauge( - "db.datasource.queries.duration.mean", - { - description: "Average duration of datasource queries", - unit: "ms", - } - ); - - // Single helper so we hit Prisma only once per scrape --------------------- - async function readPrismaMetrics() { - const metrics = await prisma.$metrics.json().catch(() => undefined); - - if (!metrics) { - return undefined; - } - - // Extract counter values - const counters: Record = {}; - for (const counter of metrics.counters) { - counters[counter.key] = counter.value; - } - - // Extract gauge values - const gauges: Record = {}; - for (const gauge of metrics.gauges) { - gauges[gauge.key] = gauge.value; - } - - // Extract histogram values - const histograms: Record = {}; - for (const histogram of metrics.histograms) { - histograms[histogram.key] = histogram.value; - } - - return { - counters: { - queriesTotal: counters["prisma_client_queries_total"] ?? 0, - datasourceQueriesTotal: counters["prisma_datasource_queries_total"] ?? 0, - connectionsOpenedTotal: counters["prisma_pool_connections_opened_total"] ?? 0, - connectionsClosedTotal: counters["prisma_pool_connections_closed_total"] ?? 0, - }, - gauges: { - queriesActive: gauges["prisma_client_queries_active"] ?? 0, - queriesWait: gauges["prisma_client_queries_wait"] ?? 0, - connectionsOpen: gauges["prisma_pool_connections_open"] ?? 0, - connectionsBusy: gauges["prisma_pool_connections_busy"] ?? 0, - connectionsIdle: gauges["prisma_pool_connections_idle"] ?? 0, - }, - histograms: { - queriesWait: histograms["prisma_client_queries_wait_histogram_ms"], - queriesDuration: histograms["prisma_client_queries_duration_histogram_ms"], - datasourceQueriesDuration: histograms["prisma_datasource_queries_duration_histogram_ms"], - }, - }; - } - - meter.addBatchObservableCallback( - async (res) => { - const prismaMetrics = await readPrismaMetrics(); - - if (!prismaMetrics) { - return; - } - - const { counters, gauges, histograms } = prismaMetrics; - - // Observe counters - res.observe(queriesTotal, counters.queriesTotal); - res.observe(datasourceQueriesTotal, counters.datasourceQueriesTotal); - res.observe(connectionsOpenedTotal, counters.connectionsOpenedTotal); - res.observe(connectionsClosedTotal, counters.connectionsClosedTotal); - - // Observe gauges - res.observe(queriesActive, gauges.queriesActive); - res.observe(queriesWait, gauges.queriesWait); - res.observe(totalGauge, gauges.connectionsOpen); - res.observe(busyGauge, gauges.connectionsBusy); - res.observe(freeGauge, gauges.connectionsIdle); - - // Observe histogram statistics as gauges - if (histograms.queriesWait) { - res.observe(queriesWaitTimeCount, histograms.queriesWait.count); - res.observe(queriesWaitTimeSum, histograms.queriesWait.sum); - res.observe( - queriesWaitTimeMean, - histograms.queriesWait.count > 0 - ? histograms.queriesWait.sum / histograms.queriesWait.count - : 0 - ); - } - - if (histograms.queriesDuration) { - res.observe(queriesDurationCount, histograms.queriesDuration.count); - res.observe(queriesDurationSum, histograms.queriesDuration.sum); - res.observe( - queriesDurationMean, - histograms.queriesDuration.count > 0 - ? histograms.queriesDuration.sum / histograms.queriesDuration.count - : 0 - ); - } - - if (histograms.datasourceQueriesDuration) { - res.observe(datasourceQueriesDurationCount, histograms.datasourceQueriesDuration.count); - res.observe(datasourceQueriesDurationSum, histograms.datasourceQueriesDuration.sum); - res.observe( - datasourceQueriesDurationMean, - histograms.datasourceQueriesDuration.count > 0 - ? histograms.datasourceQueriesDuration.sum / histograms.datasourceQueriesDuration.count - : 0 - ); - } - }, - [ - queriesTotal, - datasourceQueriesTotal, - connectionsOpenedTotal, - connectionsClosedTotal, - queriesActive, - queriesWait, - totalGauge, - busyGauge, - freeGauge, - queriesWaitTimeCount, - queriesWaitTimeSum, - queriesWaitTimeMean, - queriesDurationCount, - queriesDurationSum, - queriesDurationMean, - datasourceQueriesDurationCount, - datasourceQueriesDurationSum, - datasourceQueriesDurationMean, - ] - ); -} - function configureNodejsMetrics({ meter }: { meter: Meter }) { if (!env.INTERNAL_OTEL_NODEJS_METRICS_ENABLED) { return; diff --git a/internal-packages/database/prisma/schema.prisma b/internal-packages/database/prisma/schema.prisma index ca1d868ab04..5c3cec16374 100644 --- a/internal-packages/database/prisma/schema.prisma +++ b/internal-packages/database/prisma/schema.prisma @@ -8,7 +8,6 @@ generator client { provider = "prisma-client-js" output = "../generated/prisma" binaryTargets = ["native", "debian-openssl-1.1.x"] - previewFeatures = ["metrics"] } model User { diff --git a/internal-packages/run-ops-database/prisma/schema.prisma b/internal-packages/run-ops-database/prisma/schema.prisma index 4750efa392c..c7f8a10e170 100644 --- a/internal-packages/run-ops-database/prisma/schema.prisma +++ b/internal-packages/run-ops-database/prisma/schema.prisma @@ -7,7 +7,6 @@ generator client { provider = "prisma-client-js" output = "../generated/run-ops" binaryTargets = ["native", "debian-openssl-1.1.x"] - previewFeatures = ["metrics"] } // ───────────────────────────────────────────────────────────────────────────── From c0c21c2ebb096e9a1d74f7155deccfd2813818ba Mon Sep 17 00:00:00 2001 From: Eric Allam Date: Mon, 3 Aug 2026 10:46:35 +0100 Subject: [PATCH 4/8] revert(webapp): drop the in-list padding extension Padding was applied to the whole Prisma args object, so it rewrote values as well as predicates: a write payload or JSON comparison value containing a field named in or notIn had its trailing element duplicated. Bounding list arity is better handled per call site, where filter context is unambiguous. --- .server-changes/pad-prisma-in-list-filters.md | 6 - apps/webapp/app/db.server.ts | 5 +- .../app/utils/padInLists.server.test.ts | 124 ------------------ apps/webapp/app/utils/padInLists.server.ts | 120 ----------------- 4 files changed, 2 insertions(+), 253 deletions(-) delete mode 100644 .server-changes/pad-prisma-in-list-filters.md delete mode 100644 apps/webapp/app/utils/padInLists.server.test.ts delete mode 100644 apps/webapp/app/utils/padInLists.server.ts diff --git a/.server-changes/pad-prisma-in-list-filters.md b/.server-changes/pad-prisma-in-list-filters.md deleted file mode 100644 index 684548f83ab..00000000000 --- a/.server-changes/pad-prisma-in-list-filters.md +++ /dev/null @@ -1,6 +0,0 @@ ---- -area: webapp -type: improvement ---- - -Database queries that filter on a list of values now reuse cached query plans instead of forcing the database to re-plan for every distinct list length. diff --git a/apps/webapp/app/db.server.ts b/apps/webapp/app/db.server.ts index df4ddfb4e98..97a3ae7bb53 100644 --- a/apps/webapp/app/db.server.ts +++ b/apps/webapp/app/db.server.ts @@ -30,7 +30,6 @@ import { DATASOURCE_CONTEXT_KEY, startActiveSpan } from "./v3/tracer.server"; import type { Span } from "@opentelemetry/api"; import { context, trace } from "@opentelemetry/api"; import { queryPerformanceMonitor } from "./utils/queryPerformanceMonitor.server"; -import { padInLists } from "./utils/padInLists.server"; export type { PrismaTransactionClient, @@ -132,7 +131,7 @@ function tagDatasource(datasource: "writer" | "replica", trace.getActiveSpan()?.setAttribute("db.datasource", datasource); return context.with( context.active().setValue(DATASOURCE_CONTEXT_KEY, datasource), - async () => await query(padInLists(args)) + async () => await query(args) ); }, }, @@ -152,7 +151,7 @@ function tagDatasourceRunOps( trace.getActiveSpan()?.setAttribute("db.datasource", datasource); return context.with( context.active().setValue(DATASOURCE_CONTEXT_KEY, datasource), - async () => await query(padInLists(args)) + async () => await query(args) ); }, }, diff --git a/apps/webapp/app/utils/padInLists.server.test.ts b/apps/webapp/app/utils/padInLists.server.test.ts deleted file mode 100644 index 9639d1ba1af..00000000000 --- a/apps/webapp/app/utils/padInLists.server.test.ts +++ /dev/null @@ -1,124 +0,0 @@ -import { describe, expect, it } from "vitest"; -import { padInLists } from "./padInLists.server"; - -describe("padInLists", () => { - it("pads an in list up to the next power of two by repeating the last element", () => { - const args = { where: { id: { in: ["a", "b", "c"] } } }; - - expect(padInLists(args)).toEqual({ where: { id: { in: ["a", "b", "c", "c"] } } }); - }); - - it("pads notIn as well", () => { - const args = { where: { status: { notIn: ["A", "B", "C", "D", "E"] } } }; - - expect((padInLists(args) as typeof args).where.status.notIn).toHaveLength(8); - }); - - it("collapses arity 1..300 to 10 distinct lengths", () => { - const lengths = new Set(); - - for (let arity = 1; arity <= 300; arity++) { - const values = Array.from({ length: arity }, (_, i) => `id-${i}`); - const result = padInLists({ where: { id: { in: values } } }) as { - where: { id: { in: string[] } }; - }; - lengths.add(result.where.id.in.length); - } - - expect(lengths.size).toBe(10); - }); - - it("leaves the args reference untouched when nothing needs padding", () => { - const args = { where: { id: { in: ["a", "b"] } }, take: 10 }; - - expect(padInLists(args)).toBe(args); - }); - - it("does not mutate the caller's object", () => { - const values = ["a", "b", "c"]; - const args = { where: { id: { in: values } } }; - - padInLists(args); - - expect(values).toEqual(["a", "b", "c"]); - expect(args.where.id.in).toBe(values); - }); - - it("leaves empty and single-element lists alone", () => { - expect(padInLists({ where: { id: { in: [] } } })).toEqual({ where: { id: { in: [] } } }); - expect(padInLists({ where: { id: { in: ["only"] } } })).toEqual({ - where: { id: { in: ["only"] } }, - }); - }); - - it("leaves lists longer than the 1024 cap alone", () => { - const values = Array.from({ length: 1500 }, (_, i) => `id-${i}`); - const result = padInLists({ where: { id: { in: values } } }) as { - where: { id: { in: string[] } }; - }; - - expect(result.where.id.in).toHaveLength(1500); - }); - - it("pads a list that lands exactly on the cap boundary", () => { - const values = Array.from({ length: 700 }, (_, i) => `id-${i}`); - const result = padInLists({ where: { id: { in: values } } }) as { - where: { id: { in: string[] } }; - }; - - expect(result.where.id.in).toHaveLength(1024); - }); - - it("skips lists holding non-primitive values", () => { - const values = [{ id: "a" }, { id: "b" }, { id: "c" }]; - const result = padInLists({ where: { OR: { in: values } } }) as { - where: { OR: { in: unknown[] } }; - }; - - expect(result.where.OR.in).toHaveLength(3); - }); - - it("pads numbers and bigints", () => { - const numbers = padInLists({ where: { n: { in: [1, 2, 3] } } }) as { - where: { n: { in: number[] } }; - }; - expect(numbers.where.n.in).toEqual([1, 2, 3, 3]); - - const bigints = padInLists({ where: { n: { in: [1n, 2n, 3n] } } }) as { - where: { n: { in: bigint[] } }; - }; - expect(bigints.where.n.in).toHaveLength(4); - }); - - it("preserves class instances rather than rebuilding them as plain objects", () => { - const createdAt = new Date("2026-08-01T00:00:00.000Z"); - const args = { where: { createdAt: { gte: createdAt }, id: { in: ["a", "b", "c"] } } }; - - const result = padInLists(args) as typeof args; - - expect(result.where.createdAt.gte).toBe(createdAt); - expect(result.where.createdAt.gte).toBeInstanceOf(Date); - }); - - it("pads inside nested boolean filters", () => { - const args = { - where: { AND: [{ id: { in: ["a", "b", "c"] } }, { status: { in: ["X", "Y", "Z"] } }] }, - }; - - const result = padInLists(args) as { - where: { AND: Array<{ id?: { in: string[] }; status?: { in: string[] } }> }; - }; - - expect(result.where.AND[0].id!.in).toHaveLength(4); - expect(result.where.AND[1].status!.in).toHaveLength(4); - }); - - it("stops descending past the depth limit", () => { - let node: Record = { id: { in: ["a", "b", "c"] } }; - for (let i = 0; i < 12; i++) { - node = { nested: node }; - } - - expect(() => padInLists(node)).not.toThrow(); - }); -}); diff --git a/apps/webapp/app/utils/padInLists.server.ts b/apps/webapp/app/utils/padInLists.server.ts deleted file mode 100644 index 37cf4e5b21c..00000000000 --- a/apps/webapp/app/utils/padInLists.server.ts +++ /dev/null @@ -1,120 +0,0 @@ -/** - * Pads `in:` / `notIn:` filter arrays up to the next power of two so that a query shape - * stops minting a new prepared statement for every distinct list length. - * - * Prisma expands a list filter into one bind parameter per element, so - * `id: { in: [...] }` produces a different SQL string for every arity. On the run-graph - * batch loaders, where arity is the batch size, a single call site can produce hundreds of - * distinct statements and evict everything else from the pooler's prepared-statement cache - * (PlanetScale's default budget is 200). Measured on a local rig: arity 1..300 produced 300 - * distinct statements unpadded and 10 padded. - * - * Repeating the last element is semantically free — `IN` and `NOT IN` ignore duplicates — - * so results are unchanged. - */ - -const MAX_DEPTH = 8; - -/** - * Above this length the padding is not worth it: it is at worst 2x the bind parameters, so - * a large list would pay real bandwidth and parse cost to save a cache entry. Such lists - * are better served by a rewrite to `= ANY($1)`. - */ -const MAX_PADDED_LENGTH = 1024; - -const PADDED_KEYS = new Set(["in", "notIn"]); - -function isPaddableValue(value: unknown): boolean { - const type = typeof value; - return type === "string" || type === "number" || type === "bigint"; -} - -function padded(values: unknown[]): unknown[] | undefined { - const { length } = values; - - if (length < 2 || length > MAX_PADDED_LENGTH) { - return undefined; - } - - let target = 1; - while (target < length) { - target *= 2; - } - - if (target === length || target > MAX_PADDED_LENGTH) { - return undefined; - } - - if (!values.every(isPaddableValue)) { - return undefined; - } - - const result = values.slice(); - const last = values[length - 1]; - while (result.length < target) { - result.push(last); - } - - return result; -} - -/** - * Only plain objects are rebuilt. Class instances (Decimal, Date, Buffer, Prisma.sql, …) - * are returned untouched, because reconstructing them would drop their prototype. - */ -function isPlainObject(value: object): boolean { - const prototype = Object.getPrototypeOf(value); - return prototype === Object.prototype || prototype === null; -} - -function walk(node: unknown, depth: number): unknown { - if (depth > MAX_DEPTH || node === null || typeof node !== "object") { - return node; - } - - if (Array.isArray(node)) { - let changed = false; - const next = node.map((item) => { - const walked = walk(item, depth + 1); - changed ||= walked !== item; - return walked; - }); - return changed ? next : node; - } - - if (!isPlainObject(node)) { - return node; - } - - let changed = false; - const next: Record = {}; - - for (const [key, value] of Object.entries(node)) { - if (PADDED_KEYS.has(key) && Array.isArray(value)) { - const result = padded(value); - next[key] = result ?? value; - changed ||= result !== undefined; - continue; - } - - const walked = walk(value, depth + 1); - changed ||= walked !== value; - next[key] = walked; - } - - return changed ? next : node; -} - -/** - * Kill switch for a behaviour change on the query hot path. Set `DB_PAD_IN_LISTS=0` to - * fall straight through to the original args without a deploy. - */ -const enabled = process.env.DB_PAD_IN_LISTS !== "0"; - -/** - * Returns `args` unchanged (same reference) when there is nothing to pad, so the common - * path allocates nothing. - */ -export function padInLists(args: T): T { - return enabled ? (walk(args, 0) as T) : args; -} From 411fff949772e7cacd711ff58780a275cd230dd1 Mon Sep 17 00:00:00 2001 From: Eric Allam Date: Mon, 3 Aug 2026 12:55:35 +0100 Subject: [PATCH 5/8] feat(webapp,database): bound the arity of Prisma list filters Prisma expands a list filter into one bind parameter per element, so every distinct list length is a separate prepared statement. Where the length tracks data volume, a single call site can mint hundreds of them. Those entries are used once each, but inserting them evicts entries that were being reused, so the cost lands on unrelated queries sharing the pooler's statement cache. An unbounded list also risks the 65535 bind-parameter ceiling. Adds boundedIn(), which pads a filter list to the next power of two by repeating its last element. IN and NOT IN ignore duplicates, so results are unchanged, and a call site drops from one statement per length to at most log2(cap). It pads by repeating rather than with null because x NOT IN (a, b, NULL) is never true. Lists above 32768 are returned unchanged so padding can never push a query past the parameter limit. Two oxlint rules require it: a list filter must be an inline array literal or a boundedIn() call. The first covers filters reached through where/having/cursor and deliberately never descends into data, create, update, set or equals, where a key named "in" is user data rather than a predicate. The second covers bare filter objects passed to where-building helpers, which the first cannot see. Applies the helper to all 74 existing call sites. --- .oxlintrc.json | 44 +++- .../app/models/vercelIntegration.server.ts | 3 +- .../v3/ApiBatchResultsPresenter.server.ts | 7 +- .../v3/ApiRunListPresenter.server.ts | 9 +- .../EnvironmentVariablesPresenter.server.ts | 5 +- .../v3/ErrorsListPresenter.server.ts | 10 +- .../v3/PlaygroundPresenter.server.ts | 3 +- .../v3/QueueListPresenter.server.ts | 8 +- .../v3/SessionListPresenter.server.ts | 8 +- .../presenters/v3/SessionPresenter.server.ts | 4 +- .../presenters/v3/TestTaskPresenter.server.ts | 3 +- .../v3/WaitpointPresenter.server.ts | 3 +- .../route.tsx | 5 +- .../admin.api.v1.runs-replication.backfill.ts | 6 +- .../webapp/app/routes/admin.feature-flags.tsx | 3 +- apps/webapp/app/routes/api.v2.whoami.ts | 3 +- .../app/routes/engine.v1.dev.disconnect.ts | 4 +- .../app/routes/resources.runs.$runParam.ts | 3 +- .../app/services/realtime/runReader.server.ts | 3 +- .../app/services/realtime/sessions.server.ts | 3 +- .../app/services/runsBackfiller.server.ts | 3 +- .../clickhouseRunsRepository.server.ts | 5 +- .../services/secrets/secretStore.server.ts | 3 +- .../clickhouseSessionsRepository.server.ts | 3 +- .../services/taskIdentifierRegistry.server.ts | 5 +- .../controlPlaneResolver.server.ts | 3 +- .../alerts/errorAlertEvaluator.server.ts | 3 +- .../v3/services/bulk/BulkActionV2.server.ts | 5 +- .../services/createBackgroundWorker.server.ts | 3 +- .../app/v3/services/deployment.server.ts | 4 +- internal-packages/database/package.json | 6 +- .../database/src/boundedIn.test.ts | 64 ++++++ internal-packages/database/src/boundedIn.ts | 62 ++++++ internal-packages/database/src/index.ts | 1 + internal-packages/database/vitest.config.ts | 10 + .../run-engine/src/engine/index.ts | 5 +- .../engine/systems/executionSnapshotSystem.ts | 9 +- .../engine/systems/pendingVersionSystem.ts | 3 +- .../src/engine/systems/ttlSystem.ts | 3 +- .../src/engine/systems/waitpointSystem.ts | 4 +- .../run-store/src/PostgresRunStore.ts | 20 +- .../run-store/src/runOpsStore.ts | 13 +- oxlint-plugins/prisma-in-filter.mjs | 208 ++++++++++++++++++ pnpm-lock.yaml | 16 +- 44 files changed, 508 insertions(+), 90 deletions(-) create mode 100644 internal-packages/database/src/boundedIn.test.ts create mode 100644 internal-packages/database/src/boundedIn.ts create mode 100644 internal-packages/database/vitest.config.ts create mode 100644 oxlint-plugins/prisma-in-filter.mjs diff --git a/.oxlintrc.json b/.oxlintrc.json index d9b4cb2171f..3c21052639b 100644 --- a/.oxlintrc.json +++ b/.oxlintrc.json @@ -1,9 +1,14 @@ { "$schema": "./node_modules/oxlint/configuration_schema.json", - "plugins": ["typescript", "import", "react"], + "plugins": [ + "typescript", + "import", + "react" + ], "jsPlugins": [ "./oxlint-plugins/no-thrown-unawaited-redirect.mjs", - "./oxlint-plugins/runops-residency.mjs" + "./oxlint-plugins/runops-residency.mjs", + "./oxlint-plugins/prisma-in-filter.mjs" ], "ignorePatterns": [ "**/dist/**", @@ -30,28 +35,55 @@ "no-empty-pattern": "off", "no-control-regex": "off", "typescript/no-non-null-asserted-optional-chain": "off", - "no-unused-expressions": ["warn", { "allowShortCircuit": true, "allowTernary": true }], + "no-unused-expressions": [ + "warn", + { + "allowShortCircuit": true, + "allowTernary": true + } + ], "typescript/consistent-type-imports": "error", "import/no-duplicates": "error", "import/namespace": "off", "react-hooks/exhaustive-deps": "off", "react-hooks/rules-of-hooks": "off", - "trigger/no-thrown-unawaited-redirect": "error" + "trigger/no-thrown-unawaited-redirect": "error", + "trigger-prisma/no-unbounded-list-filter": "error", + "trigger-prisma/no-unbounded-list-filter-in-args-helper": "error" }, "overrides": [ { - "files": ["apps/webapp/app/**/*.ts", "apps/webapp/app/**/*.tsx"], + "files": [ + "apps/webapp/app/**/*.ts", + "apps/webapp/app/**/*.tsx" + ], "rules": { "trigger-runops/no-control-plane-run-graph-access": "error", "trigger-runops/no-control-plane-in-runops-slot": "error" } }, { - "files": ["apps/webapp/app/**/*.test.ts", "apps/webapp/app/**/*.test.tsx"], + "files": [ + "apps/webapp/app/**/*.test.ts", + "apps/webapp/app/**/*.test.tsx" + ], "rules": { "trigger-runops/no-control-plane-run-graph-access": "off", "trigger-runops/no-control-plane-in-runops-slot": "off" } + }, + { + "files": [ + "**/*.test.ts", + "**/*.test.tsx", + "**/test/**", + "**/tests/**", + "**/e2e/**" + ], + "rules": { + "trigger-prisma/no-unbounded-list-filter": "off", + "trigger-prisma/no-unbounded-list-filter-in-args-helper": "off" + } } ] } diff --git a/apps/webapp/app/models/vercelIntegration.server.ts b/apps/webapp/app/models/vercelIntegration.server.ts index 3a1aaf4b8ea..9365dc46de0 100644 --- a/apps/webapp/app/models/vercelIntegration.server.ts +++ b/apps/webapp/app/models/vercelIntegration.server.ts @@ -24,6 +24,7 @@ import { } from "~/v3/vercel/vercelProjectIntegrationSchema"; import { EnvironmentVariablesRepository } from "~/v3/environmentVariables/environmentVariablesRepository.server"; import { isReservedForExternalSync } from "~/v3/environmentVariableRules.server"; +import { boundedIn } from "@trigger.dev/database"; import { callVercelWithRecovery, wrapVercelCallWithRecovery, @@ -1415,7 +1416,7 @@ export class VercelIntegrationRepository { variable: { projectId: params.projectId, key: { - in: varsToSync.map((v) => v.key), + in: boundedIn(varsToSync.map((v) => v.key)), }, }, }, diff --git a/apps/webapp/app/presenters/v3/ApiBatchResultsPresenter.server.ts b/apps/webapp/app/presenters/v3/ApiBatchResultsPresenter.server.ts index c9179d59120..67ef45ebd27 100644 --- a/apps/webapp/app/presenters/v3/ApiBatchResultsPresenter.server.ts +++ b/apps/webapp/app/presenters/v3/ApiBatchResultsPresenter.server.ts @@ -12,6 +12,7 @@ import type { AuthenticatedEnvironment } from "~/services/apiAuth.server"; import { runStore as defaultRunStore } from "~/v3/runStore.server"; import { BasePresenter } from "./basePresenter.server"; +import { boundedIn } from "@trigger.dev/database"; /** * Run-ops read-through wiring. All optional; absent (or `splitEnabled` falsy) collapses `call` to * passthrough. `legacyReplica` is a READ REPLICA handle only — there is NO legacy-primary field. @@ -114,7 +115,7 @@ export class ApiBatchResultsPresenter extends BasePresenter { const taskRuns = await this.runStore.findRuns( { - where: { id: { in: taskRunIds } }, + where: { id: { in: boundedIn(taskRunIds) } }, select: memberRunSelect, }, this._prisma @@ -181,7 +182,7 @@ export class ApiBatchResultsPresenter extends BasePresenter { const taskRunIds = batchRun.items.map((item) => item.taskRunId); const newRows = (await newClient.taskRun.findMany({ - where: { id: { in: taskRunIds } }, + where: { id: { in: boundedIn(taskRunIds) } }, select: memberRunSelect, })) as TaskRunWithAttempts[]; const runsById = new Map(newRows.map((run) => [run.id, run])); @@ -193,7 +194,7 @@ export class ApiBatchResultsPresenter extends BasePresenter { ); if (legacyCandidateIds.length > 0) { const legacyRows = (await legacyReplica.taskRun.findMany({ - where: { id: { in: legacyCandidateIds } }, + where: { id: { in: boundedIn(legacyCandidateIds) } }, select: memberRunSelect, })) as TaskRunWithAttempts[]; for (const run of legacyRows) { diff --git a/apps/webapp/app/presenters/v3/ApiRunListPresenter.server.ts b/apps/webapp/app/presenters/v3/ApiRunListPresenter.server.ts index 58013703406..b345b456415 100644 --- a/apps/webapp/app/presenters/v3/ApiRunListPresenter.server.ts +++ b/apps/webapp/app/presenters/v3/ApiRunListPresenter.server.ts @@ -1,5 +1,10 @@ import { MachinePresetName, parsePacket, RunStatus } from "@trigger.dev/core/v3"; -import { type Project, type RuntimeEnvironment, type TaskRunStatus } from "@trigger.dev/database"; +import { + type Project, + type RuntimeEnvironment, + type TaskRunStatus, + boundedIn, +} from "@trigger.dev/database"; import assertNever from "assert-never"; import { z } from "zod"; import type { API_VERSIONS } from "~/api/versions"; @@ -208,7 +213,7 @@ export class ApiRunListPresenter extends BasePresenter { where: { projectId: project.id, slug: { - in: searchParams["filter[env]"], + in: boundedIn(searchParams["filter[env]"]), }, }, }); diff --git a/apps/webapp/app/presenters/v3/EnvironmentVariablesPresenter.server.ts b/apps/webapp/app/presenters/v3/EnvironmentVariablesPresenter.server.ts index 91966941fca..b6c22b9ab12 100644 --- a/apps/webapp/app/presenters/v3/EnvironmentVariablesPresenter.server.ts +++ b/apps/webapp/app/presenters/v3/EnvironmentVariablesPresenter.server.ts @@ -8,6 +8,7 @@ import type { SyncEnvVarsMapping, EnvSlug } from "~/v3/vercel/vercelProjectInteg import { VercelIntegrationService } from "~/services/vercelIntegration.server"; import { loadEnvironmentVariablesEnvironments } from "./environmentVariablesEnvironments.server"; +import { boundedIn } from "@trigger.dev/database"; type Result = Awaited>; export type EnvironmentVariableWithSetValues = Result["environmentVariables"][number]; @@ -72,7 +73,7 @@ export class EnvironmentVariablesPresenter { }, where: { environmentId: { - in: environmentIds, + in: boundedIn(environmentIds), }, }, }, @@ -103,7 +104,7 @@ export class EnvironmentVariablesPresenter { ? await this.#replicaClient.user.findMany({ where: { id: { - in: Array.from(userIds), + in: boundedIn(Array.from(userIds)), }, }, select: { diff --git a/apps/webapp/app/presenters/v3/ErrorsListPresenter.server.ts b/apps/webapp/app/presenters/v3/ErrorsListPresenter.server.ts index ea6e522dbd5..76a2319fee4 100644 --- a/apps/webapp/app/presenters/v3/ErrorsListPresenter.server.ts +++ b/apps/webapp/app/presenters/v3/ErrorsListPresenter.server.ts @@ -9,7 +9,11 @@ const errorsListGranularity = new TimeGranularity([ { max: "3 months", granularity: "1w" }, { max: "Infinity", granularity: "30d" }, ]); -import { type ErrorGroupStatus, type PrismaClientOrTransaction } from "@trigger.dev/database"; +import { + type ErrorGroupStatus, + type PrismaClientOrTransaction, + boundedIn, +} from "@trigger.dev/database"; import { type Direction } from "~/components/ListPagination"; import { timeFilterFromTo } from "~/components/runs/v3/SharedFilters"; import { findDisplayableEnvironment } from "~/models/runtimeEnvironment.server"; @@ -457,7 +461,7 @@ export class ErrorsListPresenter extends BasePresenter { if (statuses.includes("UNRESOLVED")) { const excluded = await this.replica.errorGroupState.findMany({ - where: { environmentId, status: { in: excludedStatuses } }, + where: { environmentId, status: { in: boundedIn(excludedStatuses) } }, select: { taskIdentifier: true, errorFingerprint: true }, }); if (excluded.length === 0) { @@ -470,7 +474,7 @@ export class ErrorsListPresenter extends BasePresenter { } const included = await this.replica.errorGroupState.findMany({ - where: { environmentId, status: { in: statuses } }, + where: { environmentId, status: { in: boundedIn(statuses) } }, select: { taskIdentifier: true, errorFingerprint: true }, }); if (included.length === 0) { diff --git a/apps/webapp/app/presenters/v3/PlaygroundPresenter.server.ts b/apps/webapp/app/presenters/v3/PlaygroundPresenter.server.ts index c98b5afb324..2a3566d7e80 100644 --- a/apps/webapp/app/presenters/v3/PlaygroundPresenter.server.ts +++ b/apps/webapp/app/presenters/v3/PlaygroundPresenter.server.ts @@ -8,6 +8,7 @@ import { findCurrentWorkerFromEnvironment } from "~/v3/models/workerDeployment.s import { runStore } from "~/v3/runStore.server"; import { isFinalRunStatus } from "~/v3/taskStatus"; +import { boundedIn } from "@trigger.dev/database"; export type PlaygroundAgent = { slug: string; filePath: string; @@ -135,7 +136,7 @@ export class PlaygroundPresenter { const runsById = new Map(); if (runIds.length > 0) { const runs = await runStore.findRuns({ - where: { id: { in: runIds } }, + where: { id: { in: boundedIn(runIds) } }, select: { id: true, friendlyId: true, status: true }, }); for (const run of runs) { diff --git a/apps/webapp/app/presenters/v3/QueueListPresenter.server.ts b/apps/webapp/app/presenters/v3/QueueListPresenter.server.ts index 6de35f2d45d..50278e8276e 100644 --- a/apps/webapp/app/presenters/v3/QueueListPresenter.server.ts +++ b/apps/webapp/app/presenters/v3/QueueListPresenter.server.ts @@ -1,6 +1,6 @@ import type { RunEngine } from "@internal/run-engine"; import type { Prisma } from "@trigger.dev/database"; -import { TaskQueueType } from "@trigger.dev/database"; +import { TaskQueueType, boundedIn } from "@trigger.dev/database"; import { type PrismaClientOrTransaction } from "~/db.server"; import { type AuthenticatedEnvironment } from "~/services/apiAuth.server"; import { clickhouseFactory } from "~/services/clickhouse/clickhouseFactoryInstance.server"; @@ -289,7 +289,7 @@ export class QueueListPresenter extends BasePresenter { // AND keeps the search's name filter intact alongside the exclusion (a spread // would overwrite one name condition with the other). tailQueues = await this._replica.taskQueue.findMany({ - where: { AND: [where, { name: { notIn: excludedNames } }] }, + where: { AND: [where, { name: { notIn: boundedIn(excludedNames) } }] }, select: queueListSelect, orderBy: { orderableName: "asc", @@ -321,7 +321,7 @@ export class QueueListPresenter extends BasePresenter { return []; } const queues = await this._replica.taskQueue.findMany({ - where: { AND: [where, { name: { in: names } }] }, + where: { AND: [where, { name: { in: boundedIn(names) } }] }, select: queueListSelect, }); const byName = new Map(queues.map((queue) => [queue.name, queue])); @@ -401,7 +401,7 @@ export class QueueListPresenter extends BasePresenter { const overriddenByIds = queues.map((q) => q.concurrencyLimitOverriddenBy).filter(Boolean); const overriddenByUsers = await this._replica.user.findMany({ where: { - id: { in: overriddenByIds }, + id: { in: boundedIn(overriddenByIds) }, }, }); diff --git a/apps/webapp/app/presenters/v3/SessionListPresenter.server.ts b/apps/webapp/app/presenters/v3/SessionListPresenter.server.ts index ec2ddd0eeb2..1e6d1fa2391 100644 --- a/apps/webapp/app/presenters/v3/SessionListPresenter.server.ts +++ b/apps/webapp/app/presenters/v3/SessionListPresenter.server.ts @@ -1,6 +1,10 @@ import { type Span } from "@opentelemetry/api"; import { type ClickHouse } from "@internal/clickhouse"; -import { type PrismaClient, type PrismaClientOrTransaction } from "@trigger.dev/database"; +import { + type PrismaClient, + type PrismaClientOrTransaction, + boundedIn, +} from "@trigger.dev/database"; import { type Direction } from "~/components/ListPagination"; import { timeFilters } from "~/components/runs/v3/SharedFilters"; import { findDisplayableEnvironment } from "~/models/runtimeEnvironment.server"; @@ -188,7 +192,7 @@ export class SessionListPresenter { ? runStore.findRuns( { where: { - id: { in: currentRunIds }, + id: { in: boundedIn(currentRunIds) }, projectId, runtimeEnvironmentId: environmentId, }, diff --git a/apps/webapp/app/presenters/v3/SessionPresenter.server.ts b/apps/webapp/app/presenters/v3/SessionPresenter.server.ts index 3a2f214faa0..5f0c0466cb9 100644 --- a/apps/webapp/app/presenters/v3/SessionPresenter.server.ts +++ b/apps/webapp/app/presenters/v3/SessionPresenter.server.ts @@ -1,5 +1,5 @@ import { type Span } from "@opentelemetry/api"; -import { type PrismaClientOrTransaction } from "@trigger.dev/database"; +import { type PrismaClientOrTransaction, boundedIn } from "@trigger.dev/database"; import { env } from "~/env.server"; import { findDisplayableEnvironment } from "~/models/runtimeEnvironment.server"; import { chatSnapshotStorageKey } from "~/services/realtime/chatSnapshot.server"; @@ -90,7 +90,7 @@ export class SessionPresenter { return runIds.length > 0 ? runStore.findRuns( { - where: { id: { in: runIds } }, + where: { id: { in: boundedIn(runIds) } }, select: { id: true, friendlyId: true, status: true }, }, this.replica diff --git a/apps/webapp/app/presenters/v3/TestTaskPresenter.server.ts b/apps/webapp/app/presenters/v3/TestTaskPresenter.server.ts index 430477ce582..6f60b4c3ebe 100644 --- a/apps/webapp/app/presenters/v3/TestTaskPresenter.server.ts +++ b/apps/webapp/app/presenters/v3/TestTaskPresenter.server.ts @@ -6,6 +6,7 @@ import { type RuntimeEnvironmentType, type TaskRunStatus, type TaskRunTemplate, + boundedIn, } from "@trigger.dev/database"; import { inferSchema } from "@jsonhero/schema-infer"; import parse from "parse-duration"; @@ -401,7 +402,7 @@ export class TestTaskPresenter { return this.runStore.findRuns( { where: { - id: { in: ids }, + id: { in: boundedIn(ids) }, payloadType: { in: ["application/json", "application/super+json"] }, }, select: RECENT_RUNS_SELECT, diff --git a/apps/webapp/app/presenters/v3/WaitpointPresenter.server.ts b/apps/webapp/app/presenters/v3/WaitpointPresenter.server.ts index 5cf5d91f742..aac8a5445bd 100644 --- a/apps/webapp/app/presenters/v3/WaitpointPresenter.server.ts +++ b/apps/webapp/app/presenters/v3/WaitpointPresenter.server.ts @@ -9,6 +9,7 @@ import { BasePresenter } from "./basePresenter.server"; import { NextRunListPresenter, type NextRunListItem } from "./NextRunListPresenter.server"; import { waitpointStatusToApiStatus } from "./WaitpointListPresenter.server"; +import { boundedIn } from "@trigger.dev/database"; export type WaitpointDetail = NonNullable>>; // Single-sourced display bound for a waitpoint's connected run friendlyIds. @@ -70,7 +71,7 @@ export class WaitpointPresenter extends BasePresenter { return []; } const runs = await this.runStore.findRuns({ - where: { id: { in: runIds } }, + where: { id: { in: boundedIn(runIds) } }, select: { friendlyId: true }, take: CONNECTED_RUNS_DISPLAY_LIMIT, }); diff --git a/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.environment-variables.new/route.tsx b/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.environment-variables.new/route.tsx index 02d6ef00c4d..80a864d08d2 100644 --- a/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.environment-variables.new/route.tsx +++ b/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.environment-variables.new/route.tsx @@ -57,6 +57,7 @@ import { EnvironmentVariablesRepository } from "~/v3/environmentVariables/enviro import { EnvironmentVariableKey } from "~/v3/environmentVariables/repository"; import { findUnauthorizedEnvironmentId } from "~/v3/writableEnvironments"; +import { boundedIn } from "@trigger.dev/database"; const Variable = z.object({ key: EnvironmentVariableKey, value: z.string().nonempty("Value is required"), @@ -128,7 +129,7 @@ export const action = dashboardAction( // that can't write a deployed tier can't create vars there via a direct // POST (the disabled checkboxes are not the boundary). const targetEnvironments = await prisma.runtimeEnvironment.findMany({ - where: { id: { in: submission.value.environmentIds } }, + where: { id: { in: boundedIn(submission.value.environmentIds) } }, select: { type: true }, }); const hasDeniedEnvironment = targetEnvironments.some( @@ -171,7 +172,7 @@ export const action = dashboardAction( const submittedEnvs = await prisma.runtimeEnvironment.findMany({ where: { projectId: project.id, - id: { in: submission.value.environmentIds }, + id: { in: boundedIn(submission.value.environmentIds) }, }, select: { id: true, type: true, orgMember: { select: { userId: true } } }, }); diff --git a/apps/webapp/app/routes/admin.api.v1.runs-replication.backfill.ts b/apps/webapp/app/routes/admin.api.v1.runs-replication.backfill.ts index 002da73c625..150fcaca3ee 100644 --- a/apps/webapp/app/routes/admin.api.v1.runs-replication.backfill.ts +++ b/apps/webapp/app/routes/admin.api.v1.runs-replication.backfill.ts @@ -1,5 +1,5 @@ import { type ActionFunctionArgs, json } from "@remix-run/server-runtime"; -import { type TaskRun } from "@trigger.dev/database"; +import { type TaskRun, boundedIn } from "@trigger.dev/database"; import { z } from "zod"; import { prisma } from "~/db.server"; import { runStore } from "~/v3/runStore.server"; @@ -30,9 +30,9 @@ export async function action({ request }: ActionFunctionArgs) { const batchRuns = await runStore.findRuns( { where: { - id: { in: batch }, + id: { in: boundedIn(batch) }, status: { - in: FINAL_RUN_STATUSES, + in: boundedIn(FINAL_RUN_STATUSES), }, }, }, diff --git a/apps/webapp/app/routes/admin.feature-flags.tsx b/apps/webapp/app/routes/admin.feature-flags.tsx index 76ba62ff8e9..6499f7ba4c1 100644 --- a/apps/webapp/app/routes/admin.feature-flags.tsx +++ b/apps/webapp/app/routes/admin.feature-flags.tsx @@ -28,6 +28,7 @@ import { DialogFooter, } from "~/components/primitives/Dialog"; import { cn } from "~/utils/cn"; +import { boundedIn } from "@trigger.dev/database"; import { UNSET_VALUE, BooleanControl, @@ -146,7 +147,7 @@ export const action = dashboardAction( await prisma.$transaction([ ...upsertOps, ...(keysToDelete.length > 0 - ? [prisma.featureFlag.deleteMany({ where: { key: { in: keysToDelete } } })] + ? [prisma.featureFlag.deleteMany({ where: { key: { in: boundedIn(keysToDelete) } } })] : []), ]); diff --git a/apps/webapp/app/routes/api.v2.whoami.ts b/apps/webapp/app/routes/api.v2.whoami.ts index 16629db0ec1..8b22f62463c 100644 --- a/apps/webapp/app/routes/api.v2.whoami.ts +++ b/apps/webapp/app/routes/api.v2.whoami.ts @@ -5,6 +5,7 @@ import { env } from "~/env.server"; import { v3ProjectPath } from "~/utils/pathBuilder"; import { authenticateRequest } from "~/services/apiAuth.server"; +import { boundedIn } from "@trigger.dev/database"; export async function loader({ request }: LoaderFunctionArgs) { const authenticationResult = await authenticateRequest(request, { personalAccessToken: true, @@ -112,7 +113,7 @@ async function getIdentityFromPAT( where: { externalRef: projectRef, organizationId: { - in: orgs.map((org) => org.id), + in: boundedIn(orgs.map((org) => org.id)), }, }, }); diff --git a/apps/webapp/app/routes/engine.v1.dev.disconnect.ts b/apps/webapp/app/routes/engine.v1.dev.disconnect.ts index 9f4a1d39d17..0c54eb34c91 100644 --- a/apps/webapp/app/routes/engine.v1.dev.disconnect.ts +++ b/apps/webapp/app/routes/engine.v1.dev.disconnect.ts @@ -3,7 +3,7 @@ import { Ratelimit } from "@upstash/ratelimit"; import { tryCatch } from "@trigger.dev/core"; import { DevDisconnectRequestBody } from "@trigger.dev/core/v3"; import { BulkActionId, RunId } from "@trigger.dev/core/v3/isomorphic"; -import { BulkActionNotificationType, BulkActionType } from "@trigger.dev/database"; +import { BulkActionNotificationType, BulkActionType, boundedIn } from "@trigger.dev/database"; import { prisma } from "~/db.server"; import { runStore } from "~/v3/runStore.server"; import { logger } from "~/services/logger.server"; @@ -106,7 +106,7 @@ async function cancelRunsInline(runFriendlyIds: string[], environmentId: string) const runs = await runStore.findRuns( { where: { - id: { in: runIds }, + id: { in: boundedIn(runIds) }, runtimeEnvironmentId: environmentId, }, select: { diff --git a/apps/webapp/app/routes/resources.runs.$runParam.ts b/apps/webapp/app/routes/resources.runs.$runParam.ts index 4b288d99c0e..e4328fe4b37 100644 --- a/apps/webapp/app/routes/resources.runs.$runParam.ts +++ b/apps/webapp/app/routes/resources.runs.$runParam.ts @@ -11,6 +11,7 @@ import { runStore } from "~/v3/runStore.server"; import { controlPlaneResolver } from "~/v3/runOpsMigration/controlPlaneResolver.server"; import { FINAL_ATTEMPT_STATUSES, isFinalRunStatus } from "~/v3/taskStatus"; +import { boundedIn } from "@trigger.dev/database"; export type RunInspectorData = UseDataFunctionReturn; export const loader = async ({ request, params }: LoaderFunctionArgs) => { @@ -113,7 +114,7 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => { error: true, }, where: { - status: { in: FINAL_ATTEMPT_STATUSES }, + status: { in: boundedIn(FINAL_ATTEMPT_STATUSES) }, taskRunId: run.id, }, orderBy: { diff --git a/apps/webapp/app/services/realtime/runReader.server.ts b/apps/webapp/app/services/realtime/runReader.server.ts index c215423b1d4..4308e3a7f14 100644 --- a/apps/webapp/app/services/realtime/runReader.server.ts +++ b/apps/webapp/app/services/realtime/runReader.server.ts @@ -2,6 +2,7 @@ import { type Prisma, type PrismaClient, type PrismaClientOrTransaction, + boundedIn, } from "@trigger.dev/database"; import type { RunStore } from "@internal/run-store"; import { BoundedTtlCache } from "./boundedTtlCache"; @@ -152,7 +153,7 @@ export class RunHydrator { { where: { runtimeEnvironmentId: environmentId, - id: { in: ids }, + id: { in: boundedIn(ids) }, }, select: buildHydratorSelect(skipColumns), }, diff --git a/apps/webapp/app/services/realtime/sessions.server.ts b/apps/webapp/app/services/realtime/sessions.server.ts index 7f50450c3a2..7bb7ee2f7cd 100644 --- a/apps/webapp/app/services/realtime/sessions.server.ts +++ b/apps/webapp/app/services/realtime/sessions.server.ts @@ -4,6 +4,7 @@ import type { RunStore } from "@internal/run-store"; import { $replica, prisma } from "~/db.server"; import { runStore as defaultRunStore } from "~/v3/runStore.server"; +import { boundedIn } from "@trigger.dev/database"; /** * Prefix that {@link SessionId.generate} attaches to every Session friendlyId. * Used to distinguish friendlyId lookups (`session_abc...`) from externalId @@ -176,7 +177,7 @@ export async function serializeSessionsWithFriendlyRunIds( runIds.length > 0 ? await runStore.findRuns({ where: { - id: { in: runIds }, + id: { in: boundedIn(runIds) }, projectId: scope.projectId, runtimeEnvironmentId: scope.runtimeEnvironmentId, }, diff --git a/apps/webapp/app/services/runsBackfiller.server.ts b/apps/webapp/app/services/runsBackfiller.server.ts index 3912a611368..8f1ed9790a8 100644 --- a/apps/webapp/app/services/runsBackfiller.server.ts +++ b/apps/webapp/app/services/runsBackfiller.server.ts @@ -6,6 +6,7 @@ import { startSpan } from "~/v3/tracing.server"; import { FINAL_RUN_STATUSES } from "../v3/taskStatus"; import { Logger } from "@trigger.dev/core/logger"; +import { boundedIn } from "@trigger.dev/database"; export class RunsBackfillerService { private readonly prisma: PrismaClientOrTransaction; private readonly runsReplicationInstance: RunsReplicationService; @@ -49,7 +50,7 @@ export class RunsBackfillerService { lte: to, }, status: { - in: FINAL_RUN_STATUSES, + in: boundedIn(FINAL_RUN_STATUSES), }, ...(cursor ? { id: { gt: cursor } } : {}), }, diff --git a/apps/webapp/app/services/runsRepository/clickhouseRunsRepository.server.ts b/apps/webapp/app/services/runsRepository/clickhouseRunsRepository.server.ts index c9fefd1da10..f9db41e4f0b 100644 --- a/apps/webapp/app/services/runsRepository/clickhouseRunsRepository.server.ts +++ b/apps/webapp/app/services/runsRepository/clickhouseRunsRepository.server.ts @@ -15,6 +15,7 @@ import { decodeRunsCursor, encodeRunsCursor } from "./runsCursor.server"; import { runStore } from "~/v3/runStore.server"; import { type PrismaClientOrTransaction } from "~/db.server"; +import { boundedIn } from "@trigger.dev/database"; type RunCursorRow = { runId: string; createdAt: number }; /** @@ -248,7 +249,7 @@ export class ClickHouseRunsRepository implements IRunsRepository { const runs = await this.#hydrateRunsByIds(runIds, (client, ids) => store.findRuns( { - where: { id: { in: ids } }, + where: { id: { in: boundedIn(ids) } }, select: { id: true, friendlyId: true }, }, client @@ -268,7 +269,7 @@ export class ClickHouseRunsRepository implements IRunsRepository { { where: { id: { - in: ids, + in: boundedIn(ids), }, }, select: { diff --git a/apps/webapp/app/services/secrets/secretStore.server.ts b/apps/webapp/app/services/secrets/secretStore.server.ts index f4d5aac5ef8..629007cf5e4 100644 --- a/apps/webapp/app/services/secrets/secretStore.server.ts +++ b/apps/webapp/app/services/secrets/secretStore.server.ts @@ -7,6 +7,7 @@ import { safeJsonParse } from "~/utils/json"; import { logger } from "../logger.server"; import type { SecretStoreOptions } from "./secretStoreOptionsSchema.server"; +import { boundedIn } from "@trigger.dev/database"; type ProviderInitializationOptions = { DATABASE: { prismaClient?: PrismaClientOrTransaction; @@ -118,7 +119,7 @@ class PrismaSecretStore implements SecretStoreProvider { const secrets = await this.#prismaClient.secretStore.findMany({ where: { key: { - in: keys, + in: boundedIn(keys), }, }, }); diff --git a/apps/webapp/app/services/sessionsRepository/clickhouseSessionsRepository.server.ts b/apps/webapp/app/services/sessionsRepository/clickhouseSessionsRepository.server.ts index 10086c52f36..7e983a25dfa 100644 --- a/apps/webapp/app/services/sessionsRepository/clickhouseSessionsRepository.server.ts +++ b/apps/webapp/app/services/sessionsRepository/clickhouseSessionsRepository.server.ts @@ -1,5 +1,6 @@ import { type ClickhouseQueryBuilder } from "@internal/clickhouse"; import parseDuration from "parse-duration"; +import { boundedIn } from "@trigger.dev/database"; import { convertSessionListInputOptionsToFilterOptions, type FilterSessionsOptions, @@ -83,7 +84,7 @@ export class ClickHouseSessionsRepository implements ISessionsRepository { let sessions = await this.options.prisma.session.findMany({ where: { - id: { in: idsToReturn }, + id: { in: boundedIn(idsToReturn) }, runtimeEnvironmentId: options.environmentId, }, orderBy: { createdAt: "desc" }, diff --git a/apps/webapp/app/services/taskIdentifierRegistry.server.ts b/apps/webapp/app/services/taskIdentifierRegistry.server.ts index d7dc93ba31e..527460439c1 100644 --- a/apps/webapp/app/services/taskIdentifierRegistry.server.ts +++ b/apps/webapp/app/services/taskIdentifierRegistry.server.ts @@ -2,6 +2,7 @@ import { type TaskTriggerSource, type PrismaClient, type PrismaClientOrTransaction, + boundedIn, } from "@trigger.dev/database"; import { $replica, prisma } from "~/db.server"; import { getAllTaskIdentifiers } from "~/models/task.server"; @@ -59,7 +60,7 @@ export async function syncTaskIdentifiers( db.taskIdentifier.updateMany({ where: { runtimeEnvironmentId: environmentId, - slug: { in: taskSlugs }, + slug: { in: boundedIn(taskSlugs) }, }, data: { currentTriggerSource: source, @@ -73,7 +74,7 @@ export async function syncTaskIdentifiers( db.taskIdentifier.updateMany({ where: { runtimeEnvironmentId: environmentId, - slug: { notIn: slugs }, + slug: { notIn: boundedIn(slugs) }, isInLatestDeployment: true, }, data: { isInLatestDeployment: false }, diff --git a/apps/webapp/app/v3/runOpsMigration/controlPlaneResolver.server.ts b/apps/webapp/app/v3/runOpsMigration/controlPlaneResolver.server.ts index eb19a7fb6c1..97d7a93ac4c 100644 --- a/apps/webapp/app/v3/runOpsMigration/controlPlaneResolver.server.ts +++ b/apps/webapp/app/v3/runOpsMigration/controlPlaneResolver.server.ts @@ -17,6 +17,7 @@ import { } from "./controlPlaneCache.server"; import { authIncludeWithParent, toAuthenticated } from "~/models/runtimeEnvironment.server"; +import { boundedIn } from "@trigger.dev/database"; /** * App-level control-plane resolution + cache layer. Replaces the run-ops -> control-plane * Prisma joins (env/project/org, the pinned/current worker version + its tasks/queues, the @@ -304,7 +305,7 @@ export class ControlPlaneResolver { ids: string[] ): Promise> { const rows = await client.backgroundWorker.findMany({ - where: { id: { in: ids } }, + where: { id: { in: boundedIn(ids) } }, select: { id: true, version: true, diff --git a/apps/webapp/app/v3/services/alerts/errorAlertEvaluator.server.ts b/apps/webapp/app/v3/services/alerts/errorAlertEvaluator.server.ts index 94bb10c7b8e..f56341ea688 100644 --- a/apps/webapp/app/v3/services/alerts/errorAlertEvaluator.server.ts +++ b/apps/webapp/app/v3/services/alerts/errorAlertEvaluator.server.ts @@ -4,6 +4,7 @@ import { type PrismaClientOrTransaction, type ProjectAlertChannel, type RuntimeEnvironmentType, + boundedIn, } from "@trigger.dev/database"; import { $replica, prisma } from "~/db.server"; import { ErrorAlertConfig } from "~/models/projectAlert.server"; @@ -293,7 +294,7 @@ export class ErrorAlertEvaluator { const envs = await this._replica.runtimeEnvironment.findMany({ where: { projectId, - type: { in: types }, + type: { in: boundedIn(types) }, }, select: { id: true, diff --git a/apps/webapp/app/v3/services/bulk/BulkActionV2.server.ts b/apps/webapp/app/v3/services/bulk/BulkActionV2.server.ts index 9531912b9b6..362975a60b8 100644 --- a/apps/webapp/app/v3/services/bulk/BulkActionV2.server.ts +++ b/apps/webapp/app/v3/services/bulk/BulkActionV2.server.ts @@ -4,6 +4,7 @@ import { BulkActionStatus, BulkActionType, type PrismaClient, + boundedIn, } from "@trigger.dev/database"; import { clickhouseFactory } from "~/services/clickhouse/clickhouseFactoryInstance.server"; import { @@ -313,7 +314,7 @@ export class BulkActionService extends BaseService { // still be cuid-resident, and merges (disjoint by construction). In single-DB mode it // reads the collapsed store's replica, byte-identical to the pre-migration read. const runs = await this.runStore.findRuns({ - where: { id: { in: runIdsToProcess } }, + where: { id: { in: boundedIn(runIdsToProcess) } }, select: { id: true, engine: true, @@ -362,7 +363,7 @@ export class BulkActionService extends BaseService { // Route the member hydration through the run store (NEW-first, legacy-replica probe for // the misses, disjoint merge). Full-row read: replay needs the whole TaskRun. const runs = await this.runStore.findRuns({ - where: { id: { in: runIdsToProcess } }, + where: { id: { in: boundedIn(runIdsToProcess) } }, }); await pMap( diff --git a/apps/webapp/app/v3/services/createBackgroundWorker.server.ts b/apps/webapp/app/v3/services/createBackgroundWorker.server.ts index 43551e849f0..12eca0585de 100644 --- a/apps/webapp/app/v3/services/createBackgroundWorker.server.ts +++ b/apps/webapp/app/v3/services/createBackgroundWorker.server.ts @@ -767,7 +767,7 @@ export async function syncDeclarativeSchedules( const potentiallyDeletableSchedules = await prisma.taskSchedule.findMany({ where: { id: { - in: Array.from(missingSchedules), + in: boundedIn(Array.from(missingSchedules)), }, }, include: { @@ -851,6 +851,7 @@ export async function createBackgroundFiles( import { createHash } from "crypto"; +import { boundedIn } from "@trigger.dev/database"; function hashContent(content: string): string { return createHash("sha256").update(content).digest("hex").slice(0, 16); } diff --git a/apps/webapp/app/v3/services/deployment.server.ts b/apps/webapp/app/v3/services/deployment.server.ts index f726bba3d6d..c67d7778568 100644 --- a/apps/webapp/app/v3/services/deployment.server.ts +++ b/apps/webapp/app/v3/services/deployment.server.ts @@ -1,7 +1,7 @@ import { type AuthenticatedEnvironment } from "~/services/apiAuth.server"; import { BaseService } from "./baseService.server"; import { errAsync, fromPromise, okAsync, type ResultAsync } from "neverthrow"; -import { type WorkerDeployment, type Project } from "@trigger.dev/database"; +import { type WorkerDeployment, type Project, boundedIn } from "@trigger.dev/database"; import { BuildServerMetadata, logger, @@ -220,7 +220,7 @@ export class DeploymentService extends BaseService { where: { id: deployment.id, status: { - notIn: FINAL_DEPLOYMENT_STATUSES, // status could've changed in the meantime, we're not locking the row + notIn: boundedIn(FINAL_DEPLOYMENT_STATUSES), // status could've changed in the meantime, we're not locking the row }, }, data: { diff --git a/internal-packages/database/package.json b/internal-packages/database/package.json index d2fc05131b6..9cff6d17870 100644 --- a/internal-packages/database/package.json +++ b/internal-packages/database/package.json @@ -11,7 +11,8 @@ }, "devDependencies": { "@types/decimal.js": "^7.4.3", - "rimraf": "6.0.1" + "rimraf": "6.0.1", + "vitest": "4.1.7" }, "scripts": { "clean": "rimraf dist", @@ -24,6 +25,7 @@ "db:reset": "prisma migrate reset", "typecheck": "tsc --noEmit", "build": "pnpm run clean && tsc -p tsconfig.build.json", - "dev": "tsc --noEmit false --outDir dist --declaration --watch" + "dev": "tsc --noEmit false --outDir dist --declaration --watch", + "test": "vitest run" } } diff --git a/internal-packages/database/src/boundedIn.test.ts b/internal-packages/database/src/boundedIn.test.ts new file mode 100644 index 00000000000..a8d33b606f7 --- /dev/null +++ b/internal-packages/database/src/boundedIn.test.ts @@ -0,0 +1,64 @@ +import { describe, expect, it } from "vitest"; +import { boundedIn } from "./boundedIn.js"; + +describe("boundedIn", () => { + it("pads up to the next power of two by repeating the last element", () => { + expect(boundedIn(["a", "b", "c"])).toEqual(["a", "b", "c", "c"]); + expect(boundedIn([1, 2, 3, 4, 5])).toEqual([1, 2, 3, 4, 5, 5, 5, 5]); + }); + + it("never pads with null, which would break NOT IN", () => { + const padded = boundedIn(["a", "b", "c"]); + + expect(padded).not.toContain(null); + expect(padded).not.toContain(undefined); + expect(padded.every((value) => value === "a" || value === "b" || value === "c")).toBe(true); + }); + + it("collapses arity 1..300 to 10 distinct lengths", () => { + const lengths = new Set(); + + for (let arity = 1; arity <= 300; arity++) { + lengths.add(boundedIn(Array.from({ length: arity }, (_, i) => `id-${i}`)).length); + } + + expect(lengths.size).toBe(10); + expect([...lengths].sort((a, b) => a - b)).toEqual([1, 2, 4, 8, 16, 32, 64, 128, 256, 512]); + }); + + it("returns the same reference when no padding is needed", () => { + const empty: string[] = []; + const single = ["only"]; + const exact = ["a", "b", "c", "d"]; + + expect(boundedIn(empty)).toBe(empty); + expect(boundedIn(single)).toBe(single); + expect(boundedIn(exact)).toBe(exact); + }); + + it("does not mutate the input", () => { + const values = ["a", "b", "c"]; + + boundedIn(values); + + expect(values).toEqual(["a", "b", "c"]); + }); + + it("leaves lists above the bind-parameter cap unchanged", () => { + const huge = Array.from({ length: 40_000 }, (_, i) => i); + + expect(boundedIn(huge)).toBe(huge); + }); + + it("pads the largest list that still fits under the cap", () => { + const values = Array.from({ length: 20_000 }, (_, i) => i); + + expect(boundedIn(values)).toHaveLength(32_768); + }); + + it("preserves the original values in order", () => { + const padded = boundedIn(["x", "y", "z"]); + + expect(padded.slice(0, 3)).toEqual(["x", "y", "z"]); + }); +}); diff --git a/internal-packages/database/src/boundedIn.ts b/internal-packages/database/src/boundedIn.ts new file mode 100644 index 00000000000..015e94b5163 --- /dev/null +++ b/internal-packages/database/src/boundedIn.ts @@ -0,0 +1,62 @@ +/** + * Bounds the bind-parameter count of a Prisma `in` / `notIn` list filter. + * + * Prisma expands a list filter into one bind parameter per element, so every distinct list + * length is a separate prepared statement. Where the length tracks data volume (a batch + * size, a run-graph fan-out, a prior query's id set) one call site can mint hundreds of + * statements. Those entries are used once, but inserting them evicts entries that were + * being reused, so the cost lands on unrelated queries competing for the same pooler cache. + * + * Padding to the next power of two caps a call site at roughly log2(cap) statements instead + * of one per length. `IN` and `NOT IN` ignore duplicates, so repeating the last element + * leaves results unchanged. + * + * Call it at the filter itself, never on a whole args object: + * + * where: { id: { in: boundedIn(ids) } } + * + * Applying this by walking Prisma's args generically is not equivalent and is not safe: a + * key named `in` inside `data`, or inside a JSON `equals` value, is user data rather than a + * predicate, and padding it corrupts what gets stored or compared. + */ + +/** + * Postgres accepts at most 65535 bind parameters in one statement. Padding past half of + * that risks turning a working query into a protocol error, so lists above the cap are + * returned unchanged; a site that can reach this size wants chunking, not padding. + */ +const MAX_PADDED_LENGTH = 32768; + +/** + * Pads `values` up to the next power of two by repeating the last element. + * + * Returns the input array unchanged when it is empty, has a single element, is already a + * power of two, or exceeds the cap, so the common path allocates nothing. + * + * Pads by repeating rather than with null deliberately: `x NOT IN (a, b, NULL)` is never + * true, so null-padding a `notIn` filter would silently match no rows. + */ +export function boundedIn(values: T[]): T[] { + const { length } = values; + + if (length < 2 || length > MAX_PADDED_LENGTH) { + return values; + } + + let target = 1; + while (target < length) { + target *= 2; + } + + if (target === length || target > MAX_PADDED_LENGTH) { + return values; + } + + const padded = values.slice(); + const last = values[length - 1]!; + while (padded.length < target) { + padded.push(last); + } + + return padded; +} diff --git a/internal-packages/database/src/index.ts b/internal-packages/database/src/index.ts index 94e211e91aa..fa6872c12e6 100644 --- a/internal-packages/database/src/index.ts +++ b/internal-packages/database/src/index.ts @@ -1,2 +1,3 @@ export * from "../generated/prisma"; +export * from "./boundedIn"; export * from "./transaction"; diff --git a/internal-packages/database/vitest.config.ts b/internal-packages/database/vitest.config.ts new file mode 100644 index 00000000000..16d38181a8f --- /dev/null +++ b/internal-packages/database/vitest.config.ts @@ -0,0 +1,10 @@ +import { defineConfig } from "vitest/config"; + +export default defineConfig({ + test: { + include: ["src/**/*.test.ts"], + globals: true, + isolate: true, + testTimeout: 10_000, + }, +}); diff --git a/internal-packages/run-engine/src/engine/index.ts b/internal-packages/run-engine/src/engine/index.ts index b88f4f276e4..3c8cf32934c 100644 --- a/internal-packages/run-engine/src/engine/index.ts +++ b/internal-packages/run-engine/src/engine/index.ts @@ -37,6 +37,7 @@ import { type TaskRunExecutionSnapshot, type Waitpoint, Prisma, + boundedIn, } from "@trigger.dev/database"; import { Worker } from "@trigger.dev/redis-worker"; import { assertNever } from "assert-never"; @@ -2955,7 +2956,7 @@ export class RunEngine { ): Promise> { const runs = await this.runStore.findRuns({ where: { - id: { in: runIds }, + id: { in: boundedIn(runIds) }, completedAt: { lte: new Date(Date.now() - completedAtOffsetMs), // This only finds runs that were completed more than 10 minutes ago }, @@ -2963,7 +2964,7 @@ export class RunEngine { not: null, }, status: { - in: getFinalRunStatuses(), + in: boundedIn(getFinalRunStatuses()), }, }, select: { diff --git a/internal-packages/run-engine/src/engine/systems/executionSnapshotSystem.ts b/internal-packages/run-engine/src/engine/systems/executionSnapshotSystem.ts index dca4c66b2e7..48299ac220c 100644 --- a/internal-packages/run-engine/src/engine/systems/executionSnapshotSystem.ts +++ b/internal-packages/run-engine/src/engine/systems/executionSnapshotSystem.ts @@ -15,6 +15,7 @@ import { ExecutionSnapshotNotFoundError, ServiceValidationError } from "../error import type { HeartbeatTimeouts } from "../types.js"; import type { SystemResources } from "./systems.js"; +import { boundedIn } from "@trigger.dev/database"; /** Chunk size for fetching waitpoints to avoid NAPI string conversion limits */ const WAITPOINT_CHUNK_SIZE = 100; @@ -186,9 +187,13 @@ async function fetchWaitpointsInChunks( for (let i = 0; i < waitpointIds.length; i += WAITPOINT_CHUNK_SIZE) { const chunk = waitpointIds.slice(i, i + WAITPOINT_CHUNK_SIZE); const waitpoints = runStore - ? await runStore.findManyWaitpoints({ where: { id: { in: chunk } } }, prisma, runId) + ? await runStore.findManyWaitpoints( + { where: { id: { in: boundedIn(chunk) } } }, + prisma, + runId + ) : await prisma.waitpoint.findMany({ - where: { id: { in: chunk } }, + where: { id: { in: boundedIn(chunk) } }, }); allWaitpoints.push(...waitpoints); } diff --git a/internal-packages/run-engine/src/engine/systems/pendingVersionSystem.ts b/internal-packages/run-engine/src/engine/systems/pendingVersionSystem.ts index 1636394a8b2..1984c82ef27 100644 --- a/internal-packages/run-engine/src/engine/systems/pendingVersionSystem.ts +++ b/internal-packages/run-engine/src/engine/systems/pendingVersionSystem.ts @@ -1,6 +1,7 @@ import type { EnqueueSystem } from "./enqueueSystem.js"; import type { SystemResources } from "./systems.js"; +import { boundedIn } from "@trigger.dev/database"; export type PendingVersionSystemOptions = { resources: SystemResources; enqueueSystem: EnqueueSystem; @@ -96,7 +97,7 @@ export class PendingVersionSystem { const pendingRuns = await this.$.runStore.findRuns( { where: { - id: { in: candidateIds }, + id: { in: boundedIn(candidateIds) }, status: "PENDING_VERSION", }, orderBy: { diff --git a/internal-packages/run-engine/src/engine/systems/ttlSystem.ts b/internal-packages/run-engine/src/engine/systems/ttlSystem.ts index 0fb3b8387cb..0f8920c4649 100644 --- a/internal-packages/run-engine/src/engine/systems/ttlSystem.ts +++ b/internal-packages/run-engine/src/engine/systems/ttlSystem.ts @@ -8,6 +8,7 @@ import type { WaitpointSystem } from "./waitpointSystem.js"; import { startSpan } from "@internal/tracing"; import pMap from "p-map"; +import { boundedIn } from "@trigger.dev/database"; export type TtlSystemOptions = { resources: SystemResources; waitpointSystem: WaitpointSystem; @@ -160,7 +161,7 @@ export class TtlSystem { // Fetch all runs in a single query (no snapshot data needed) const runs = await this.$.runStore.findRuns( { - where: { id: { in: runIds } }, + where: { id: { in: boundedIn(runIds) } }, select: { id: true, spanId: true, diff --git a/internal-packages/run-engine/src/engine/systems/waitpointSystem.ts b/internal-packages/run-engine/src/engine/systems/waitpointSystem.ts index 9cf372b7b15..5d5a80772a6 100644 --- a/internal-packages/run-engine/src/engine/systems/waitpointSystem.ts +++ b/internal-packages/run-engine/src/engine/systems/waitpointSystem.ts @@ -7,7 +7,7 @@ import type { TaskRunExecutionStatus, Waitpoint, } from "@trigger.dev/database"; -import { Prisma } from "@trigger.dev/database"; +import { Prisma, boundedIn } from "@trigger.dev/database"; import type { RunStore } from "@internal/run-store"; import { assertNever } from "assert-never"; import { nanoid } from "nanoid"; @@ -929,7 +929,7 @@ export class WaitpointSystem { await this.$.runStore.deleteManyTaskRunWaitpoints({ where: { taskRunId: runId, - id: { in: blockingWaitpoints.map((b) => b.id) }, + id: { in: boundedIn(blockingWaitpoints.map((b) => b.id)) }, }, }); diff --git a/internal-packages/run-store/src/PostgresRunStore.ts b/internal-packages/run-store/src/PostgresRunStore.ts index 676e345ad5a..338bc4960f2 100644 --- a/internal-packages/run-store/src/PostgresRunStore.ts +++ b/internal-packages/run-store/src/PostgresRunStore.ts @@ -1,4 +1,4 @@ -import { Prisma } from "@trigger.dev/database"; +import { Prisma, boundedIn } from "@trigger.dev/database"; import type { BatchTaskRun, BatchTaskRunItemStatus, @@ -242,7 +242,7 @@ async function batchHydrateJoinRelation( } const targetIds = [...new Set(links.map((l) => l[joinTargetField]))]; const rows = (await targetDelegate.findMany( - targetFindManyArgs({ id: { in: targetIds } }, projection, ["id"]) + targetFindManyArgs({ id: { in: boundedIn(targetIds) } }, projection, ["id"]) )) as Record[]; const byTargetId = new Map(rows.map((r) => [r.id as string, r])); for (const link of links) { @@ -267,7 +267,7 @@ const hydrateAssociatedWaitpoint: DedicatedRelationHydrator = async ( return byParent; } const rows = (await client.waitpoint.findMany( - targetFindManyArgs({ completedByTaskRunId: { in: parentIds } }, projection, [ + targetFindManyArgs({ completedByTaskRunId: { in: boundedIn(parentIds) } }, projection, [ "completedByTaskRunId", ]) )) as Record[]; @@ -311,7 +311,7 @@ const hydrateBlockingTaskRuns: DedicatedRelationHydrator = async (client, parent return byParent; } const edges = (await client.taskRunWaitpoint.findMany({ - where: { waitpointId: { in: parentIds } }, + where: { waitpointId: { in: boundedIn(parentIds) } }, })) as Record[]; const nestedTaskRun = projection?.select?.taskRun; const runProjection = nestedTaskRun ? projectionOf(nestedTaskRun as SubProjection) : undefined; @@ -321,7 +321,7 @@ const hydrateBlockingTaskRuns: DedicatedRelationHydrator = async (client, parent const runs = ( runIds.length > 0 ? await client.taskRun.findMany( - targetFindManyArgs({ id: { in: runIds } }, runProjection, ["id"]) + targetFindManyArgs({ id: { in: boundedIn(runIds) } }, runProjection, ["id"]) ) : [] ) as Record[]; @@ -371,7 +371,7 @@ const hydrateConnectedRuns: DedicatedRelationHydrator = async (client, parents, } const targetIds = [...new Set(links.map((l) => l.taskRunId))]; const rows = (await client.taskRun.findMany( - targetFindManyArgs({ id: { in: targetIds } }, projection, ["id"]) + targetFindManyArgs({ id: { in: boundedIn(targetIds) } }, projection, ["id"]) )) as Record[]; const byTargetId = new Map(rows.map((r) => [r.id as string, r])); for (const link of links) { @@ -1435,7 +1435,7 @@ export class PostgresRunStore implements RunStore { // byFriendlyIds — only clears idempotencyKey, not idempotencyKeyExpiresAt const result = await prisma.taskRun.updateMany({ - where: { friendlyId: { in: params.byFriendlyIds } }, + where: { friendlyId: { in: boundedIn(params.byFriendlyIds) } }, data: { idempotencyKey: null }, }); return { count: result.count }; @@ -1668,7 +1668,9 @@ export class PostgresRunStore implements RunStore { ? { include: args.include } : {}; const rows = (await this.findRuns( - { where: { id: { in: ids } }, ...projected } as Parameters[0], + { where: { id: { in: boundedIn(ids) } }, ...projected } as Parameters< + PostgresRunStore["findRuns"] + >[0], client )) as Record[]; const byId = new Map(); @@ -1760,7 +1762,7 @@ export class PostgresRunStore implements RunStore { return []; } return client.waitpoint.findMany({ - where: { id: { in: links.map((l) => l.waitpointId) } }, + where: { id: { in: boundedIn(links.map((l) => l.waitpointId)) } }, }); } diff --git a/internal-packages/run-store/src/runOpsStore.ts b/internal-packages/run-store/src/runOpsStore.ts index 467df81df26..28df27a4f2d 100644 --- a/internal-packages/run-store/src/runOpsStore.ts +++ b/internal-packages/run-store/src/runOpsStore.ts @@ -32,6 +32,7 @@ import type { import { isReadReplicaClient } from "./readReplicaClient.js"; import { CONNECTED_RUNS_LIMIT } from "./PostgresRunStore.js"; +import { boundedIn } from "@trigger.dev/database"; /** * Run-ops routing substrate for the TaskRun-core method group. Implements {@link RunStore} * by selecting between a NEW store (the dedicated run-ops DB, where new runs are born) and @@ -401,7 +402,7 @@ export class RoutingRunStore implements RunStore { ? { include: args.include } : {}; const rows = (await this.findRuns( - { where: { id: { in: ids } }, ...projected } as FindRunsArgs, + { where: { id: { in: boundedIn(ids) } }, ...projected } as FindRunsArgs, client )) as Record[]; const byId = new Map(); @@ -886,7 +887,7 @@ export class RoutingRunStore implements RunStore { return; // all completed tokens co-resident → owning-store hydration is complete } const recovered = (await this.findManyWaitpoints( - { where: { id: { in: missing } } }, + { where: { id: { in: boundedIn(missing) } } }, client )) as Record[]; snapshot.completedWaitpoints = [...completed, ...recovered]; @@ -1412,7 +1413,7 @@ export class RoutingRunStore implements RunStore { return this.findManyExecutionSnapshots( { ...(findArgs as Prisma.TaskRunExecutionSnapshotFindManyArgs), - where: { id: { in: snapshotIds } }, + where: { id: { in: boundedIn(snapshotIds) } }, }, client ); @@ -1552,7 +1553,7 @@ export class RoutingRunStore implements RunStore { return; } const waitpoints = (await this.findManyWaitpoints( - { where: { id: { in: ids } } }, + { where: { id: { in: boundedIn(ids) } } }, client )) as Record[]; const byId = new Map(waitpoints.map((w) => [w.id as string, w])); @@ -2005,7 +2006,7 @@ function idListFromWhere(where: Prisma.TaskRunWhereInput): string[] | undefined } function narrowToIds(args: FindRunsArgs, ids: string[]): FindRunsArgs { - return { ...args, where: { ...args.where, id: { in: ids } } }; + return { ...args, where: { ...args.where, id: { in: boundedIn(ids) } } }; } // Clone find-many args, replacing the `id` filter with `{ in: ids }` while keeping any other `where` @@ -2013,7 +2014,7 @@ function narrowToIds(args: FindRunsArgs, ids: string[]): FindRunsArgs { function narrowArgsToIds(args: Record, ids: string[]): Record { return { ...args, - where: { ...((args.where as Record) ?? {}), id: { in: ids } }, + where: { ...((args.where as Record) ?? {}), id: { in: boundedIn(ids) } }, }; } diff --git a/oxlint-plugins/prisma-in-filter.mjs b/oxlint-plugins/prisma-in-filter.mjs new file mode 100644 index 00000000000..3193550b85c --- /dev/null +++ b/oxlint-plugins/prisma-in-filter.mjs @@ -0,0 +1,208 @@ +/** + * oxlint plugin: trigger-prisma — flags `in:` / `notIn:` list filters. + * + * Prisma expands a list filter into one bind parameter per element, so every distinct list + * length is a separate prepared statement. Where the list length tracks data volume (batch + * size, run-graph fan-out, a prior query's id set) a single call site can mint hundreds of + * statements, and the pooler's prepared-statement cache evicts entries that were being + * reused to make room for ones that never will be. + * + * The fix is per call site: bound the list, chunk it to a fixed size, or rewrite to + * `= ANY($1)` so arity stops changing the SQL. This rule enumerates the sites that need + * that treatment and stops new ones appearing. + * + * Deliberately scoped to filter position. A key named `in` inside `data`, `create`, + * `update`, `set` or a JSON `equals` value is user data, not a predicate, and must never be + * touched — rewriting those corrupts what gets stored or compared. + */ + +/** Subtrees that hold predicates. Descend into these. */ +const FILTER_ROOTS = new Set(["where", "having", "cursor"]); + +/** + * Keys whose values are stored or compared verbatim. Never descend into these, even inside + * a `where`: a JSON column's `equals` value is data, not a predicate. + */ +const VALUE_POSITION = new Set([ + "data", + "create", + "update", + "set", + "equals", + "connect", + "connectOrCreate", + "select", + "include", + "_count", +]); + +const LIST_FILTERS = new Set(["in", "notIn"]); + +/** + * Helpers whose first argument IS a where clause, so the filter arrives as a bare object + * with no `where:` key for the main rule to key off. Repo-specific by design, in the same + * spirit as the delegate list in runops-residency.mjs: an explicit list cannot silently + * stop matching the way a heuristic can. + */ +const FILTER_ARG_HELPERS = new Set(["targetFindManyArgs"]); + +/** Fallback for helpers that follow the naming convention but are not listed above. */ +const FILTER_ARG_HELPER_PATTERN = + /(?:FindMany|FindFirst|FindUnique|Count|DeleteMany|UpdateMany)Args$/; + +function isFilterArgHelper(callee) { + const name = + callee.type === "Identifier" + ? callee.name + : callee.type === "MemberExpression" && + !callee.computed && + callee.property.type === "Identifier" + ? callee.property.name + : undefined; + if (!name) return false; + return FILTER_ARG_HELPERS.has(name) || FILTER_ARG_HELPER_PATTERN.test(name); +} + +/** The sanctioned bounding helper from `@trigger.dev/database`. */ +const BOUNDING_HELPER = "boundedIn"; + +/** + * A list filter is acceptable when its arity cannot vary at runtime: an inline array + * literal (fixed in the source) or a `boundedIn()` call (padded to a power of two). + * Type-only wrappers are unwrapped so `boundedIn(ids) as string[]` still counts. + */ +function isBounded(node) { + let current = node; + while ( + current && + (current.type === "TSAsExpression" || + current.type === "TSSatisfiesExpression" || + current.type === "TSNonNullExpression") + ) { + current = current.expression; + } + if (!current) return false; + + if (current.type === "ArrayExpression") return true; + + if (current.type === "CallExpression") { + const callee = current.callee; + if (callee.type === "Identifier") return callee.name === BOUNDING_HELPER; + if (callee.type === "MemberExpression" && !callee.computed) { + return callee.property.type === "Identifier" && callee.property.name === BOUNDING_HELPER; + } + } + + return false; +} + +function propertyKeyName(node) { + if (!node || node.type !== "Property") return undefined; + const key = node.key; + if (!node.computed && key.type === "Identifier") return key.name; + if (key.type === "Literal" && typeof key.value === "string") return key.value; + return undefined; +} + +/** + * Reports every `in` / `notIn` reachable from a filter root without passing through a + * value-position key. Depth-bounded so a pathological args object cannot stall the linter. + */ +function reportListFilters(node, context, depth, messageId = "listFilter", extra = {}) { + if (!node || typeof node !== "object" || depth > 12) return; + + if (node.type === "ArrayExpression") { + for (const element of node.elements) { + reportListFilters(element, context, depth + 1, messageId, extra); + } + return; + } + + if (node.type !== "ObjectExpression") return; + + for (const property of node.properties) { + if (property.type !== "Property") continue; + + const name = propertyKeyName(property); + if (!name || VALUE_POSITION.has(name)) continue; + + if (LIST_FILTERS.has(name)) { + if (!isBounded(property.value)) { + context.report({ + node: property, + messageId, + data: { filter: name, ...extra }, + }); + } + continue; + } + + reportListFilters(property.value, context, depth + 1, messageId, extra); + } +} + +/** @type {import("eslint").Rule.RuleModule} */ +const noUnboundedListFilter = { + meta: { + type: "problem", + docs: { + description: + "Disallow `in` / `notIn` list filters, whose arity changes the generated SQL and churns the prepared-statement cache.", + }, + messages: { + listFilter: + "Prisma `{{filter}}:` filter. Its length becomes the bind-parameter count, so each distinct length is a separate prepared statement. Bound or chunk the list, or rewrite to `= ANY($1)`. If the length is genuinely fixed and small, disable this line with a reason.", + }, + schema: [], + }, + create(context) { + return { + Property(node) { + const name = propertyKeyName(node); + if (!name || !FILTER_ROOTS.has(name)) return; + reportListFilters(node.value, context, 0); + }, + }; + }, +}; + +/** @type {import("eslint").Rule.RuleModule} */ +const noUnboundedListFilterInArgsHelper = { + meta: { + type: "problem", + docs: { + description: + "Disallow `in` / `notIn` in a bare filter object passed to a where-building helper, which the where-keyed rule cannot see.", + }, + messages: { + listFilter: + "Prisma `{{filter}}:` filter passed to `{{helper}}()` as a bare where clause. Its length becomes the bind-parameter count, so each distinct length is a separate prepared statement. Bound or chunk the list, or rewrite to `= ANY($1)`.", + }, + schema: [], + }, + create(context) { + return { + CallExpression(node) { + if (!isFilterArgHelper(node.callee)) return; + const first = node.arguments[0]; + if (!first || first.type !== "ObjectExpression") return; + + const helper = + node.callee.type === "Identifier" ? node.callee.name : node.callee.property.name; + + reportListFilters(first, context, 0, "listFilter", { helper }); + }, + }; + }, +}; + +/** @type {import("eslint").ESLint.Plugin} */ +const plugin = { + meta: { name: "trigger-prisma" }, + rules: { + "no-unbounded-list-filter": noUnboundedListFilter, + "no-unbounded-list-filter-in-args-helper": noUnboundedListFilterInArgsHelper, + }, +}; + +export default plugin; diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index f37e3250982..b04e65fa1e6 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -1060,6 +1060,9 @@ importers: rimraf: specifier: 6.0.1 version: 6.0.1 + vitest: + specifier: 4.1.7 + version: 4.1.7(@opentelemetry/api@1.9.1)(@types/node@24.13.3)(@vitest/coverage-v8@4.1.7)(vite@6.4.2(@types/node@24.13.3)(jiti@2.7.0)(lightningcss@1.32.0)(terser@5.46.1)(tsx@4.22.4)(yaml@2.9.0)) internal-packages/emails: dependencies: @@ -14625,10 +14628,6 @@ packages: resolution: {integrity: sha512-mEwzpUgrLySlveBwEVDMKk5B57bhLPYovRfPAXD5gA/98Opn0rCDj3GtLwFvCvH5RK9uPCExUROW5NjDwvqkxw==} engines: {node: '>=12.0.0'} - tinyglobby@0.2.16: - resolution: {integrity: sha512-pn99VhoACYR8nFHhxqix+uvsbXineAasWm5ojXoN8xEwK5Kd3/TrhNn1wByuD52UxWRLy8pu+kRMniEi6Eq9Zg==} - engines: {node: '>=12.0.0'} - tinyglobby@0.2.17: resolution: {integrity: sha512-wXR/dYpcqKmfWpEdZjiKJOwCNFndD0DMnrW/cYjVGttEkBfVgcLFHoNrlj47mjOVic9yyNu65alsgF4NQyTa2g==} engines: {node: '>=12.0.0'} @@ -30608,11 +30607,6 @@ snapshots: fdir: 6.5.0(picomatch@4.0.4) picomatch: 4.0.4 - tinyglobby@0.2.16: - dependencies: - fdir: 6.5.0(picomatch@4.0.4) - picomatch: 4.0.4 - tinyglobby@0.2.17: dependencies: fdir: 6.5.0(picomatch@4.0.4) @@ -31387,7 +31381,7 @@ snapshots: std-env: 4.1.0 tinybench: 2.9.0 tinyexec: 1.2.3 - tinyglobby: 0.2.16 + tinyglobby: 0.2.17 tinyrainbow: 3.1.0 vite: 6.4.2(@types/node@24.13.3)(jiti@2.7.0)(lightningcss@1.32.0)(terser@5.46.1)(tsx@3.12.2)(yaml@2.9.0) why-is-node-running: 2.3.0 @@ -31416,7 +31410,7 @@ snapshots: std-env: 4.1.0 tinybench: 2.9.0 tinyexec: 1.2.3 - tinyglobby: 0.2.16 + tinyglobby: 0.2.17 tinyrainbow: 3.1.0 vite: 6.4.2(@types/node@24.13.3)(jiti@2.7.0)(lightningcss@1.32.0)(terser@5.46.1)(tsx@4.22.4)(yaml@2.9.0) why-is-node-running: 2.3.0 From 19f00f458e125483ea09ca5865c142e9f3a5e84c Mon Sep 17 00:00:00 2001 From: Eric Allam Date: Mon, 3 Aug 2026 13:21:07 +0100 Subject: [PATCH 6/8] chore(webapp): reformat lint config and refresh the run-ops guard baseline The boundedIn import shifted four line numbers in ApiBatchResultsPresenter, so the guard read its existing baseline entries as new violations. Same four violations, same file, one line lower. --- .oxlintrc.json | 24 ++++--------------- .../v3/runOpsMigration/track1-baseline.json | 8 +++---- 2 files changed, 8 insertions(+), 24 deletions(-) diff --git a/.oxlintrc.json b/.oxlintrc.json index 3c21052639b..61e3a684a2d 100644 --- a/.oxlintrc.json +++ b/.oxlintrc.json @@ -1,10 +1,6 @@ { "$schema": "./node_modules/oxlint/configuration_schema.json", - "plugins": [ - "typescript", - "import", - "react" - ], + "plugins": ["typescript", "import", "react"], "jsPlugins": [ "./oxlint-plugins/no-thrown-unawaited-redirect.mjs", "./oxlint-plugins/runops-residency.mjs", @@ -53,33 +49,21 @@ }, "overrides": [ { - "files": [ - "apps/webapp/app/**/*.ts", - "apps/webapp/app/**/*.tsx" - ], + "files": ["apps/webapp/app/**/*.ts", "apps/webapp/app/**/*.tsx"], "rules": { "trigger-runops/no-control-plane-run-graph-access": "error", "trigger-runops/no-control-plane-in-runops-slot": "error" } }, { - "files": [ - "apps/webapp/app/**/*.test.ts", - "apps/webapp/app/**/*.test.tsx" - ], + "files": ["apps/webapp/app/**/*.test.ts", "apps/webapp/app/**/*.test.tsx"], "rules": { "trigger-runops/no-control-plane-run-graph-access": "off", "trigger-runops/no-control-plane-in-runops-slot": "off" } }, { - "files": [ - "**/*.test.ts", - "**/*.test.tsx", - "**/test/**", - "**/tests/**", - "**/e2e/**" - ], + "files": ["**/*.test.ts", "**/*.test.tsx", "**/test/**", "**/tests/**", "**/e2e/**"], "rules": { "trigger-prisma/no-unbounded-list-filter": "off", "trigger-prisma/no-unbounded-list-filter-in-args-helper": "off" diff --git a/apps/webapp/app/v3/runOpsMigration/track1-baseline.json b/apps/webapp/app/v3/runOpsMigration/track1-baseline.json index 9127f2d48ee..63d27dbadca 100644 --- a/apps/webapp/app/v3/runOpsMigration/track1-baseline.json +++ b/apps/webapp/app/v3/runOpsMigration/track1-baseline.json @@ -80,7 +80,7 @@ "violations": [ { "file": "apps/webapp/app/presenters/v3/ApiBatchResultsPresenter.server.ts", - "line": 88, + "line": 89, "model": "BatchTaskRun", "delegate": "batchTaskRun", "callKind": "read", @@ -89,7 +89,7 @@ }, { "file": "apps/webapp/app/presenters/v3/ApiBatchResultsPresenter.server.ts", - "line": 149, + "line": 150, "model": "BatchTaskRun", "delegate": "batchTaskRun", "callKind": "read", @@ -98,7 +98,7 @@ }, { "file": "apps/webapp/app/presenters/v3/ApiBatchResultsPresenter.server.ts", - "line": 183, + "line": 184, "model": "TaskRun", "delegate": "taskRun", "callKind": "read", @@ -107,7 +107,7 @@ }, { "file": "apps/webapp/app/presenters/v3/ApiBatchResultsPresenter.server.ts", - "line": 195, + "line": 196, "model": "TaskRun", "delegate": "taskRun", "callKind": "read", From d447f963f62aaa76fcd3b4c2286f391ff73dead6 Mon Sep 17 00:00:00 2001 From: Eric Allam Date: Mon, 3 Aug 2026 13:22:12 +0100 Subject: [PATCH 7/8] docs: add release note for bounded list filter arity --- .server-changes/bounded-list-filter-arity.md | 6 ++++++ 1 file changed, 6 insertions(+) create mode 100644 .server-changes/bounded-list-filter-arity.md diff --git a/.server-changes/bounded-list-filter-arity.md b/.server-changes/bounded-list-filter-arity.md new file mode 100644 index 00000000000..219440b1d1e --- /dev/null +++ b/.server-changes/bounded-list-filter-arity.md @@ -0,0 +1,6 @@ +--- +area: webapp +type: improvement +--- + +Database queries that filter on a list of values now reuse cached query plans more consistently, instead of forcing the database to re-plan whenever the list length changes. From eb4604d4a95d4137fc4b9a531bd61ccea5471c7a Mon Sep 17 00:00:00 2001 From: Eric Allam Date: Mon, 3 Aug 2026 15:41:01 +0100 Subject: [PATCH 8/8] fix(webapp): close three blind spots in the list-filter lint rule The rule accepted any array literal as fixed-arity, but a literal containing a spread has runtime-variable length, so [...new Set(ids)] passed. It also walked only plain object properties, leaving filters assembled conditionally invisible: spread-conditional properties, ternary-valued properties, and logical-and objects. Ten further call sites were unbounded behind those shapes, including one in PostgresRunStore whose four sibling hydrators had all been converted. --- apps/webapp/app/models/member.server.ts | 3 +- .../v3/BatchListPresenter.server.ts | 4 +- .../presenters/v3/RegionsPresenter.server.ts | 4 +- .../v3/ScheduleListPresenter.server.ts | 6 +-- .../v3/WaitpointListPresenter.server.ts | 3 +- ...nts.$environmentId.engine.repair-queues.ts | 3 +- ...billingLimitConvergeEnvironments.server.ts | 3 +- .../billingLimitQueuedRuns.server.ts | 3 +- .../run-store/src/PostgresRunStore.ts | 2 +- oxlint-plugins/prisma-in-filter.mjs | 41 ++++++++++++++++--- 10 files changed, 53 insertions(+), 19 deletions(-) diff --git a/apps/webapp/app/models/member.server.ts b/apps/webapp/app/models/member.server.ts index 3be5f7ce09c..e4e6bff2bf2 100644 --- a/apps/webapp/app/models/member.server.ts +++ b/apps/webapp/app/models/member.server.ts @@ -11,6 +11,7 @@ import { getDefaultEnvironmentConcurrencyLimit } from "~/services/platform.v3.se import { rbac } from "~/services/rbac.server"; import { ssoController } from "~/services/sso.server"; +import { boundedIn } from "@trigger.dev/database"; export const INVITE_NOT_FOUND = "Invite not found"; export const INVITE_BLOCKED_DIRECTORY_MANAGED = "Membership for this organization is managed by Directory Sync, so invites can't be accepted."; @@ -134,7 +135,7 @@ export async function inviteMembers({ const existingMembers = await prisma.orgMember.findMany({ where: { organizationId: org.id, - user: { email: { in: [...uniqueEmails] } }, + user: { email: { in: boundedIn([...uniqueEmails]) } }, }, select: { user: { select: { email: true } } }, }); diff --git a/apps/webapp/app/presenters/v3/BatchListPresenter.server.ts b/apps/webapp/app/presenters/v3/BatchListPresenter.server.ts index 6d7f60316c2..6de786159a7 100644 --- a/apps/webapp/app/presenters/v3/BatchListPresenter.server.ts +++ b/apps/webapp/app/presenters/v3/BatchListPresenter.server.ts @@ -1,4 +1,4 @@ -import { type BatchTaskRunStatus } from "@trigger.dev/database"; +import { type BatchTaskRunStatus, boundedIn } from "@trigger.dev/database"; import { type RunOpsPrismaClient } from "@internal/run-ops-database"; import parse from "parse-duration"; import { type PrismaClientOrTransaction } from "~/db.server"; @@ -263,7 +263,7 @@ export class BatchListPresenter extends BasePresenter { : {}), ...(friendlyId ? { friendlyId } : {}), ...(statuses && statuses.length > 0 - ? { status: { in: statuses }, batchVersion: { not: "v1" } } + ? { status: { in: boundedIn(statuses) }, batchVersion: { not: "v1" } } : {}), ...(createdAtGte !== undefined || createdAtLte !== undefined ? { diff --git a/apps/webapp/app/presenters/v3/RegionsPresenter.server.ts b/apps/webapp/app/presenters/v3/RegionsPresenter.server.ts index 538bd2d3c8a..818ab445233 100644 --- a/apps/webapp/app/presenters/v3/RegionsPresenter.server.ts +++ b/apps/webapp/app/presenters/v3/RegionsPresenter.server.ts @@ -1,4 +1,4 @@ -import { type WorkloadType } from "@trigger.dev/database"; +import { type WorkloadType, boundedIn } from "@trigger.dev/database"; import { type Project } from "~/models/project.server"; import { type User } from "~/models/user.server"; import { FEATURE_FLAG } from "~/v3/featureFlags"; @@ -87,7 +87,7 @@ export class RegionsPresenter extends BasePresenter { : // Hide hidden unless they're allowed to use them project.allowedWorkerQueues.length > 0 ? { - masterQueue: { in: project.allowedWorkerQueues }, + masterQueue: { in: boundedIn(project.allowedWorkerQueues) }, } : defaultVisibilityFilter(hasComputeAccess), orderBy: { diff --git a/apps/webapp/app/presenters/v3/ScheduleListPresenter.server.ts b/apps/webapp/app/presenters/v3/ScheduleListPresenter.server.ts index e36e7abb99e..ab394b76ec1 100644 --- a/apps/webapp/app/presenters/v3/ScheduleListPresenter.server.ts +++ b/apps/webapp/app/presenters/v3/ScheduleListPresenter.server.ts @@ -1,4 +1,4 @@ -import { type RuntimeEnvironmentType, type ScheduleType } from "@trigger.dev/database"; +import { type RuntimeEnvironmentType, type ScheduleType, boundedIn } from "@trigger.dev/database"; import { type ScheduleListFilters } from "~/components/runs/v3/ScheduleFilters"; import { displayableEnvironment } from "~/models/runtimeEnvironment.server"; import { getTaskIdentifiers } from "~/models/task.server"; @@ -164,7 +164,7 @@ export class ScheduleListPresenter extends BasePresenter { const totalCount = await this._replica.taskSchedule.count({ where: { projectId: project.id, - taskIdentifier: tasks ? { in: tasks } : undefined, + taskIdentifier: tasks ? { in: boundedIn(tasks) } : undefined, instances: { some: { environmentId, @@ -227,7 +227,7 @@ export class ScheduleListPresenter extends BasePresenter { }, where: { projectId: project.id, - taskIdentifier: tasks ? { in: tasks } : undefined, + taskIdentifier: tasks ? { in: boundedIn(tasks) } : undefined, instances: { some: { environmentId, diff --git a/apps/webapp/app/presenters/v3/WaitpointListPresenter.server.ts b/apps/webapp/app/presenters/v3/WaitpointListPresenter.server.ts index 6c132f0b4f5..980f4f42e4a 100644 --- a/apps/webapp/app/presenters/v3/WaitpointListPresenter.server.ts +++ b/apps/webapp/app/presenters/v3/WaitpointListPresenter.server.ts @@ -3,6 +3,7 @@ import { type RunEngineVersion, type RuntimeEnvironmentType, type WaitpointStatus, + boundedIn, } from "@trigger.dev/database"; import { type Direction } from "~/components/ListPagination"; import { type PrismaClientOrTransaction } from "~/db.server"; @@ -186,7 +187,7 @@ export class WaitpointListPresenter extends BasePresenter { type: "MANUAL", ...(cursor ? { id: direction === "forward" ? { lt: cursor } : { gt: cursor } } : {}), ...(id ? { friendlyId: id } : {}), - ...(statusesToFilter.length ? { status: { in: statusesToFilter } } : {}), + ...(statusesToFilter.length ? { status: { in: boundedIn(statusesToFilter) } } : {}), ...(filterOutputIsError !== undefined ? { outputIsError: filterOutputIsError } : {}), ...(idempotencyKey ? { OR: [{ idempotencyKey }, { inactiveIdempotencyKey: idempotencyKey }] } diff --git a/apps/webapp/app/routes/admin.api.v1.environments.$environmentId.engine.repair-queues.ts b/apps/webapp/app/routes/admin.api.v1.environments.$environmentId.engine.repair-queues.ts index 6748655c025..a9ac295aed6 100644 --- a/apps/webapp/app/routes/admin.api.v1.environments.$environmentId.engine.repair-queues.ts +++ b/apps/webapp/app/routes/admin.api.v1.environments.$environmentId.engine.repair-queues.ts @@ -7,6 +7,7 @@ import { requireAdminApiRequest } from "~/services/personalAccessToken.server"; import { determineEngineVersion } from "~/v3/engineVersion.server"; import { engine } from "~/v3/runEngine.server"; +import { boundedIn } from "@trigger.dev/database"; const ParamsSchema = z.object({ environmentId: z.string(), }); @@ -49,7 +50,7 @@ export async function action({ request, params }: ActionFunctionArgs) { where: { runtimeEnvironmentId: environment.id, version: "V2", - name: parsedBody.queues.length > 0 ? { in: parsedBody.queues } : undefined, + name: parsedBody.queues.length > 0 ? { in: boundedIn(parsedBody.queues) } : undefined, }, select: { friendlyId: true, diff --git a/apps/webapp/app/v3/services/billingLimit/billingLimitConvergeEnvironments.server.ts b/apps/webapp/app/v3/services/billingLimit/billingLimitConvergeEnvironments.server.ts index 0b59d4c7fae..031841edf83 100644 --- a/apps/webapp/app/v3/services/billingLimit/billingLimitConvergeEnvironments.server.ts +++ b/apps/webapp/app/v3/services/billingLimit/billingLimitConvergeEnvironments.server.ts @@ -4,6 +4,7 @@ import { type PrismaClient, type Project, type RuntimeEnvironment, + boundedIn, } from "@trigger.dev/database"; import { prisma } from "~/db.server"; import { logger } from "~/services/logger.server"; @@ -71,7 +72,7 @@ async function pauseBillingLimitEnvironments( const environments = await db.runtimeEnvironment.findMany({ where: { organizationId, - type: { in: [...BILLABLE_ENVIRONMENT_TYPES] }, + type: { in: boundedIn([...BILLABLE_ENVIRONMENT_TYPES]) }, paused: false, }, take: batchSize, diff --git a/apps/webapp/app/v3/services/billingLimit/billingLimitQueuedRuns.server.ts b/apps/webapp/app/v3/services/billingLimit/billingLimitQueuedRuns.server.ts index 60abe490f81..7cf3cf7cd99 100644 --- a/apps/webapp/app/v3/services/billingLimit/billingLimitQueuedRuns.server.ts +++ b/apps/webapp/app/v3/services/billingLimit/billingLimitQueuedRuns.server.ts @@ -5,6 +5,7 @@ import { clickhouseFactory } from "~/services/clickhouse/clickhouseFactoryInstan import { RunsRepository } from "~/services/runsRepository/runsRepository.server"; import { BILLABLE_ENVIRONMENT_TYPES } from "./billingLimitConstants"; +import { boundedIn } from "@trigger.dev/database"; export type BillableEnvironmentRef = { id: string; projectId: string; @@ -17,7 +18,7 @@ export async function getBillableEnvironmentsForBillingLimit( return prismaClient.runtimeEnvironment.findMany({ where: { organizationId, - type: { in: [...BILLABLE_ENVIRONMENT_TYPES] }, + type: { in: boundedIn([...BILLABLE_ENVIRONMENT_TYPES]) }, }, select: { id: true, diff --git a/internal-packages/run-store/src/PostgresRunStore.ts b/internal-packages/run-store/src/PostgresRunStore.ts index 338bc4960f2..ec9a0734fb6 100644 --- a/internal-packages/run-store/src/PostgresRunStore.ts +++ b/internal-packages/run-store/src/PostgresRunStore.ts @@ -427,7 +427,7 @@ async function batchHydrateEdgeTarget( return byParent; } const rows = (await targetDelegate.findMany( - targetFindManyArgs({ id: { in: [...new Set(targetIds)] } }, projection, ["id"]) + targetFindManyArgs({ id: { in: boundedIn([...new Set(targetIds)]) } }, projection, ["id"]) )) as Record[]; const byTargetId = new Map(rows.map((r) => [r.id as string, r])); for (const p of parents) { diff --git a/oxlint-plugins/prisma-in-filter.mjs b/oxlint-plugins/prisma-in-filter.mjs index 3193550b85c..586187d2f5d 100644 --- a/oxlint-plugins/prisma-in-filter.mjs +++ b/oxlint-plugins/prisma-in-filter.mjs @@ -70,6 +70,10 @@ const BOUNDING_HELPER = "boundedIn"; * A list filter is acceptable when its arity cannot vary at runtime: an inline array * literal (fixed in the source) or a `boundedIn()` call (padded to a power of two). * Type-only wrappers are unwrapped so `boundedIn(ids) as string[]` still counts. + * + * An array literal counts only when nothing spreads into it. `[...new Set(ids)]` is an + * ArrayExpression whose length is decided at runtime, which is precisely the case the + * helper exists for. */ function isBounded(node) { let current = node; @@ -83,7 +87,9 @@ function isBounded(node) { } if (!current) return false; - if (current.type === "ArrayExpression") return true; + if (current.type === "ArrayExpression") { + return current.elements.every((element) => !element || element.type !== "SpreadElement"); + } if (current.type === "CallExpression") { const callee = current.callee; @@ -107,20 +113,43 @@ function propertyKeyName(node) { /** * Reports every `in` / `notIn` reachable from a filter root without passing through a * value-position key. Depth-bounded so a pathological args object cannot stall the linter. + * + * Filters are routinely assembled conditionally, so the walk follows the shapes that carry + * them: `cond ? { … } : {}`, `cond && { … }`, and `...(cond ? { … } : {})`. Stopping at a + * plain ObjectExpression would leave those permanently invisible to the rule. */ function reportListFilters(node, context, depth, messageId = "listFilter", extra = {}) { if (!node || typeof node !== "object" || depth > 12) return; - if (node.type === "ArrayExpression") { - for (const element of node.elements) { - reportListFilters(element, context, depth + 1, messageId, extra); - } - return; + const descend = (child) => reportListFilters(child, context, depth + 1, messageId, extra); + + switch (node.type) { + case "TSAsExpression": + case "TSSatisfiesExpression": + case "TSNonNullExpression": + return descend(node.expression); + case "ConditionalExpression": + descend(node.consequent); + return descend(node.alternate); + case "LogicalExpression": + descend(node.left); + return descend(node.right); + case "ArrayExpression": + for (const element of node.elements) descend(element); + return; + case "SpreadElement": + return descend(node.argument); + default: + break; } if (node.type !== "ObjectExpression") return; for (const property of node.properties) { + if (property.type === "SpreadElement") { + descend(property.argument); + continue; + } if (property.type !== "Property") continue; const name = propertyKeyName(property);