From 2ab8f845fb3c1e9e7a5a5ba4159d4dc6be8be91e Mon Sep 17 00:00:00 2001 From: Eric Allam Date: Sat, 8 Aug 2026 14:46:13 +0100 Subject: [PATCH 1/5] feat(webapp,database): opt-in per-client Prisma driver adapters Add per-client env vars to route each Prisma client through @prisma/adapter-pg (node-postgres) instead of the built-in engine driver. All default off, so behavior is unchanged unless a flag is set: - CONTROL_PLANE_DATABASE_WRITER_DRIVER_ADAPTER - CONTROL_PLANE_DATABASE_REPLICA_DRIVER_ADAPTER - RUN_OPS_DATABASE_WRITER_DRIVER_ADAPTER - RUN_OPS_DATABASE_REPLICA_DRIVER_ADAPTER - RUN_OPS_LEGACY_DATABASE_WRITER_DRIVER_ADAPTER - RUN_OPS_LEGACY_DATABASE_REPLICA_DRIVER_ADAPTER Enables the driverAdapters preview feature on both schemas (keeps the Rust query engine; does not add queryCompiler). Each adapter pool is built with a bounded connectionTimeoutMillis and an onPoolError handler. Handle the connect-failure differences the adapter introduces: - isInfrastructureError now recognizes the adapter's connect-failure shapes (P2010 'not reachable' and raw ECONNREFUSED/ENOTFOUND-class errors) so the DB host is still scrubbed from API-client errors and infra failures are logged. - isPrismaRetriableError treats the adapter pool-acquire timeout as retriable, preserving the P2024 retry behavior. refs TRI-13039 Co-Authored-By: Claude Opus 4.8 (1M context) --- .../prisma-driver-adapter-per-client.md | 6 + apps/webapp/app/db.server.ts | 405 +++++++++++------- apps/webapp/app/env.server.ts | 6 + apps/webapp/app/utils/prismaErrors.ts | 26 +- apps/webapp/package.json | 2 + .../database/prisma/schema.prisma | 2 +- internal-packages/database/src/transaction.ts | 9 +- .../run-ops-database/prisma/schema.prisma | 2 +- pnpm-lock.yaml | 31 +- 9 files changed, 330 insertions(+), 159 deletions(-) create mode 100644 .server-changes/prisma-driver-adapter-per-client.md diff --git a/.server-changes/prisma-driver-adapter-per-client.md b/.server-changes/prisma-driver-adapter-per-client.md new file mode 100644 index 00000000000..6fdd88138e7 --- /dev/null +++ b/.server-changes/prisma-driver-adapter-per-client.md @@ -0,0 +1,6 @@ +--- +area: webapp +type: improvement +--- + +Each database client can now optionally connect through the node-postgres driver, configurable independently per client and off by default, so default behavior is unchanged. diff --git a/apps/webapp/app/db.server.ts b/apps/webapp/app/db.server.ts index 3cbff8ff17b..f2a2d93b98a 100644 --- a/apps/webapp/app/db.server.ts +++ b/apps/webapp/app/db.server.ts @@ -10,6 +10,8 @@ import { } from "@trigger.dev/database"; import { RunOpsPrismaClient } from "@internal/run-ops-database"; import { markReadReplicaClient } from "@internal/run-store"; +import { PrismaPg } from "@prisma/adapter-pg"; +import { Pool } from "pg"; import invariant from "tiny-invariant"; import { z } from "zod"; import { env } from "./env.server"; @@ -307,7 +309,14 @@ const runOpsTopology: RunOpsTopology = singleton("runOpsTopology", () => { controlPlane: { writer: prisma, replica: $replica }, buildNewWriter: (url, clientType) => captureInfraErrorsRunOps( - tagDatasourceRunOps("run-ops-writer", buildRunOpsWriterClient({ url, clientType })) + tagDatasourceRunOps( + "run-ops-writer", + buildRunOpsWriterClient({ + url, + clientType, + useDriverAdapter: env.RUN_OPS_DATABASE_WRITER_DRIVER_ADAPTER === "1", + }) + ) ), // 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 — @@ -315,7 +324,14 @@ const runOpsTopology: RunOpsTopology = singleton("runOpsTopology", () => { buildNewReplica: (url, clientType) => markReadReplicaClient( captureInfraErrorsRunOps( - tagDatasourceRunOps("run-ops-replica", buildRunOpsReplicaClient({ url, clientType })) + tagDatasourceRunOps( + "run-ops-replica", + buildRunOpsReplicaClient({ + url, + clientType, + useDriverAdapter: env.RUN_OPS_DATABASE_REPLICA_DRIVER_ADAPTER === "1", + }) + ) ) ), // Legacy client shares the exact control-plane wrapper stack (the legacy DB carries the full @@ -329,6 +345,7 @@ const runOpsTopology: RunOpsTopology = singleton("runOpsTopology", () => { clientType, poolTimeout: env.RUN_OPS_LEGACY_DATABASE_WRITER_POOL_TIMEOUT, connectTimeout: env.RUN_OPS_LEGACY_DATABASE_WRITER_CONNECTION_TIMEOUT, + useDriverAdapter: env.RUN_OPS_LEGACY_DATABASE_WRITER_DRIVER_ADAPTER === "1", }) ) ), @@ -342,6 +359,7 @@ const runOpsTopology: RunOpsTopology = singleton("runOpsTopology", () => { clientType, poolTimeout: env.RUN_OPS_LEGACY_DATABASE_READ_REPLICA_POOL_TIMEOUT, connectTimeout: env.RUN_OPS_LEGACY_DATABASE_READ_REPLICA_CONNECTION_TIMEOUT, + useDriverAdapter: env.RUN_OPS_LEGACY_DATABASE_REPLICA_DRIVER_ADAPTER === "1", }) ) ) @@ -415,7 +433,29 @@ function getClient() { clientType: "writer", poolTimeout: env.DATABASE_WRITER_POOL_TIMEOUT, connectTimeout: env.DATABASE_WRITER_CONNECTION_TIMEOUT, + useDriverAdapter: env.CONTROL_PLANE_DATABASE_WRITER_DRIVER_ADAPTER === "1", + }); +} + +function buildDriverAdapterPool( + connectionString: string, + clientType: string, + poolTimeoutSeconds: number +): PrismaPg { + const pool = new Pool({ + connectionString, + max: env.DATABASE_CONNECTION_LIMIT, + connectionTimeoutMillis: poolTimeoutSeconds * 1000, + application_name: env.SERVICE_NAME, + }); + pool.on("error", (error) => { + logger.error("prisma driver adapter pool error", { + clientType, + error: error instanceof Error ? error.message : String(error), + ignoreError: true, + }); }); + return new PrismaPg(pool); } // Generalized writer builder shared by the control-plane client and the run-ops @@ -426,11 +466,13 @@ export function buildWriterClient({ clientType, poolTimeout, connectTimeout, + useDriverAdapter = false, }: { url: string; clientType: string; poolTimeout?: number; connectTimeout?: number; + useDriverAdapter?: boolean; }): PrismaClient { const databaseUrl = buildPrismaConnectionUrl(url, { connectionLimit: env.DATABASE_CONNECTION_LIMIT.toString(), @@ -439,66 +481,77 @@ export function buildWriterClient({ applicationName: env.SERVICE_NAME, }); - console.log(`🔌 setting up prisma client to ${redactUrlSecrets(databaseUrl)}`); + console.log( + `🔌 setting up prisma client to ${redactUrlSecrets(databaseUrl)}${ + useDriverAdapter ? " (pg driver adapter)" : "" + }` + ); - const client = new PrismaClient({ - datasources: { - db: { - url: databaseUrl.href, - }, + const logConfig = [ + // events + { + emit: "event", + level: "error", }, - log: [ - // events - { - emit: "event", - level: "error", - }, - { - emit: "event", - level: "info", - }, - { - emit: "event", - level: "warn", - }, - // stdout - ...((process.env.PRISMA_LOG_TO_STDOUT === "1" - ? [ - { - emit: "stdout", - level: "error", - }, - { - emit: "stdout", - level: "info", - }, - { - emit: "stdout", - level: "warn", - }, - ] - : []) satisfies Prisma.LogDefinition[]), - // Query performance monitoring - ...((process.env.VERBOSE_PRISMA_LOGS === "1" || - process.env.VERY_SLOW_QUERY_THRESHOLD_MS !== undefined - ? [ - { - emit: "event", - level: "query", - }, - ] - : []) satisfies Prisma.LogDefinition[]), - // verbose - ...((process.env.VERBOSE_PRISMA_LOGS === "1" - ? [ - { - emit: "stdout", - level: "query", - }, - ] - : []) satisfies Prisma.LogDefinition[]), - ], - }); + { + emit: "event", + level: "info", + }, + { + emit: "event", + level: "warn", + }, + // stdout + ...((process.env.PRISMA_LOG_TO_STDOUT === "1" + ? [ + { + emit: "stdout", + level: "error", + }, + { + emit: "stdout", + level: "info", + }, + { + emit: "stdout", + level: "warn", + }, + ] + : []) satisfies Prisma.LogDefinition[]), + // Query performance monitoring + ...((process.env.VERBOSE_PRISMA_LOGS === "1" || + process.env.VERY_SLOW_QUERY_THRESHOLD_MS !== undefined + ? [ + { + emit: "event", + level: "query", + }, + ] + : []) satisfies Prisma.LogDefinition[]), + // verbose + ...((process.env.VERBOSE_PRISMA_LOGS === "1" + ? [ + { + emit: "stdout", + level: "query", + }, + ] + : []) satisfies Prisma.LogDefinition[]), + ] satisfies Prisma.LogDefinition[]; + + const client = useDriverAdapter + ? new PrismaClient({ + adapter: buildDriverAdapterPool( + databaseUrl.href, + clientType, + poolTimeout ?? env.DATABASE_POOL_TIMEOUT + ), + log: logConfig, + }) + : new PrismaClient({ + datasources: { db: { url: databaseUrl.href } }, + log: logConfig, + }); // Only use structured logging if we're not already logging to stdout if (process.env.PRISMA_LOG_TO_STDOUT !== "1") { @@ -571,6 +624,7 @@ function getReplicaClient() { clientType: "reader", poolTimeout: env.DATABASE_READ_REPLICA_POOL_TIMEOUT, connectTimeout: env.DATABASE_READ_REPLICA_CONNECTION_TIMEOUT, + useDriverAdapter: env.CONTROL_PLANE_DATABASE_REPLICA_DRIVER_ADAPTER === "1", }); } @@ -582,11 +636,13 @@ export function buildReplicaClient({ clientType, poolTimeout, connectTimeout, + useDriverAdapter = false, }: { url: string; clientType: string; poolTimeout?: number; connectTimeout?: number; + useDriverAdapter?: boolean; }): PrismaClient { const replicaUrl = buildPrismaConnectionUrl(url, { connectionLimit: env.DATABASE_CONNECTION_LIMIT.toString(), @@ -595,66 +651,77 @@ export function buildReplicaClient({ applicationName: env.SERVICE_NAME, }); - console.log(`🔌 setting up read replica connection to ${redactUrlSecrets(replicaUrl)}`); + console.log( + `🔌 setting up read replica connection to ${redactUrlSecrets(replicaUrl)}${ + useDriverAdapter ? " (pg driver adapter)" : "" + }` + ); - const replicaClient = new PrismaClient({ - datasources: { - db: { - url: replicaUrl.href, - }, + const logConfig = [ + // events + { + emit: "event", + level: "error", }, - log: [ - // events - { - emit: "event", - level: "error", - }, - { - emit: "event", - level: "info", - }, - { - emit: "event", - level: "warn", - }, - // stdout - ...((process.env.PRISMA_LOG_TO_STDOUT === "1" - ? [ - { - emit: "stdout", - level: "error", - }, - { - emit: "stdout", - level: "info", - }, - { - emit: "stdout", - level: "warn", - }, - ] - : []) satisfies Prisma.LogDefinition[]), - // Query performance monitoring - ...((process.env.VERBOSE_PRISMA_LOGS === "1" || - process.env.VERY_SLOW_QUERY_THRESHOLD_MS !== undefined - ? [ - { - emit: "event", - level: "query", - }, - ] - : []) satisfies Prisma.LogDefinition[]), - // verbose - ...((process.env.VERBOSE_PRISMA_LOGS === "1" - ? [ - { - emit: "stdout", - level: "query", - }, - ] - : []) satisfies Prisma.LogDefinition[]), - ], - }); + { + emit: "event", + level: "info", + }, + { + emit: "event", + level: "warn", + }, + // stdout + ...((process.env.PRISMA_LOG_TO_STDOUT === "1" + ? [ + { + emit: "stdout", + level: "error", + }, + { + emit: "stdout", + level: "info", + }, + { + emit: "stdout", + level: "warn", + }, + ] + : []) satisfies Prisma.LogDefinition[]), + // Query performance monitoring + ...((process.env.VERBOSE_PRISMA_LOGS === "1" || + process.env.VERY_SLOW_QUERY_THRESHOLD_MS !== undefined + ? [ + { + emit: "event", + level: "query", + }, + ] + : []) satisfies Prisma.LogDefinition[]), + // verbose + ...((process.env.VERBOSE_PRISMA_LOGS === "1" + ? [ + { + emit: "stdout", + level: "query", + }, + ] + : []) satisfies Prisma.LogDefinition[]), + ] satisfies Prisma.LogDefinition[]; + + const replicaClient = useDriverAdapter + ? new PrismaClient({ + adapter: buildDriverAdapterPool( + replicaUrl.href, + clientType, + poolTimeout ?? env.DATABASE_POOL_TIMEOUT + ), + log: logConfig, + }) + : new PrismaClient({ + datasources: { db: { url: replicaUrl.href } }, + log: logConfig, + }); // Only use structured logging if we're not already logging to stdout if (process.env.PRISMA_LOG_TO_STDOUT !== "1") { @@ -714,9 +781,11 @@ export function buildReplicaClient({ function buildRunOpsWriterClient({ url, clientType, + useDriverAdapter = false, }: { url: string; clientType: string; + useDriverAdapter?: boolean; }): RunOpsPrismaClient { const databaseUrl = buildPrismaConnectionUrl(url, { connectionLimit: env.DATABASE_CONNECTION_LIMIT.toString(), @@ -727,20 +796,41 @@ function buildRunOpsWriterClient({ applicationName: env.SERVICE_NAME, }); - console.log(`🔌 setting up run-ops prisma client to ${redactUrlSecrets(databaseUrl)}`); - - const client = new RunOpsPrismaClient({ - datasources: { db: { url: databaseUrl.href } }, - log: [ - { emit: "event", level: "error" }, - { emit: "event", level: "info" }, - { emit: "event", level: "warn" }, - ...((process.env.VERBOSE_PRISMA_LOGS === "1" || - process.env.VERY_SLOW_QUERY_THRESHOLD_MS !== undefined - ? [{ emit: "event", level: "query" }] - : []) as { emit: "event"; level: "query" }[]), - ], - }); + console.log( + `🔌 setting up run-ops prisma client to ${redactUrlSecrets(databaseUrl)}${ + useDriverAdapter ? " (pg driver adapter)" : "" + }` + ); + + const client = useDriverAdapter + ? new RunOpsPrismaClient({ + adapter: buildDriverAdapterPool( + databaseUrl.href, + clientType, + env.RUN_OPS_DATABASE_WRITER_POOL_TIMEOUT ?? env.DATABASE_POOL_TIMEOUT + ), + log: [ + { emit: "event", level: "error" }, + { emit: "event", level: "info" }, + { emit: "event", level: "warn" }, + ...((process.env.VERBOSE_PRISMA_LOGS === "1" || + process.env.VERY_SLOW_QUERY_THRESHOLD_MS !== undefined + ? [{ emit: "event", level: "query" }] + : []) as { emit: "event"; level: "query" }[]), + ], + }) + : new RunOpsPrismaClient({ + datasources: { db: { url: databaseUrl.href } }, + log: [ + { emit: "event", level: "error" }, + { emit: "event", level: "info" }, + { emit: "event", level: "warn" }, + ...((process.env.VERBOSE_PRISMA_LOGS === "1" || + process.env.VERY_SLOW_QUERY_THRESHOLD_MS !== undefined + ? [{ emit: "event", level: "query" }] + : []) as { emit: "event"; level: "query" }[]), + ], + }); if (process.env.PRISMA_LOG_TO_STDOUT !== "1") { client.$on("info", (log) => logger.info("RunOpsPrismaClient info", { clientType, event: log })); @@ -767,9 +857,11 @@ function buildRunOpsWriterClient({ function buildRunOpsReplicaClient({ url, clientType, + useDriverAdapter = false, }: { url: string; clientType: string; + useDriverAdapter?: boolean; }): RunOpsPrismaClient { const replicaUrl = buildPrismaConnectionUrl(url, { connectionLimit: ( @@ -784,20 +876,41 @@ function buildRunOpsReplicaClient({ applicationName: env.SERVICE_NAME, }); - console.log(`🔌 setting up run-ops read replica connection to ${redactUrlSecrets(replicaUrl)}`); - - const client = new RunOpsPrismaClient({ - datasources: { db: { url: replicaUrl.href } }, - log: [ - { emit: "event", level: "error" }, - { emit: "event", level: "info" }, - { emit: "event", level: "warn" }, - ...((process.env.VERBOSE_PRISMA_LOGS === "1" || - process.env.VERY_SLOW_QUERY_THRESHOLD_MS !== undefined - ? [{ emit: "event", level: "query" }] - : []) as { emit: "event"; level: "query" }[]), - ], - }); + console.log( + `🔌 setting up run-ops read replica connection to ${redactUrlSecrets(replicaUrl)}${ + useDriverAdapter ? " (pg driver adapter)" : "" + }` + ); + + const client = useDriverAdapter + ? new RunOpsPrismaClient({ + adapter: buildDriverAdapterPool( + replicaUrl.href, + clientType, + env.RUN_OPS_DATABASE_READ_REPLICA_POOL_TIMEOUT ?? env.DATABASE_POOL_TIMEOUT + ), + log: [ + { emit: "event", level: "error" }, + { emit: "event", level: "info" }, + { emit: "event", level: "warn" }, + ...((process.env.VERBOSE_PRISMA_LOGS === "1" || + process.env.VERY_SLOW_QUERY_THRESHOLD_MS !== undefined + ? [{ emit: "event", level: "query" }] + : []) as { emit: "event"; level: "query" }[]), + ], + }) + : new RunOpsPrismaClient({ + datasources: { db: { url: replicaUrl.href } }, + log: [ + { emit: "event", level: "error" }, + { emit: "event", level: "info" }, + { emit: "event", level: "warn" }, + ...((process.env.VERBOSE_PRISMA_LOGS === "1" || + process.env.VERY_SLOW_QUERY_THRESHOLD_MS !== undefined + ? [{ emit: "event", level: "query" }] + : []) as { emit: "event"; level: "query" }[]), + ], + }); if (process.env.PRISMA_LOG_TO_STDOUT !== "1") { client.$on("info", (log) => logger.info("RunOpsPrismaClient info", { clientType, event: log })); diff --git a/apps/webapp/app/env.server.ts b/apps/webapp/app/env.server.ts index 3ae566b1b00..c64c39f2d03 100644 --- a/apps/webapp/app/env.server.ts +++ b/apps/webapp/app/env.server.ts @@ -238,6 +238,12 @@ const EnvironmentSchema = z ) .optional(), CONTROL_PLANE_DATABASE_READ_REPLICA_URL: z.string().optional(), + CONTROL_PLANE_DATABASE_WRITER_DRIVER_ADAPTER: z.string().default("0"), + CONTROL_PLANE_DATABASE_REPLICA_DRIVER_ADAPTER: z.string().default("0"), + RUN_OPS_DATABASE_WRITER_DRIVER_ADAPTER: z.string().default("0"), + RUN_OPS_DATABASE_REPLICA_DRIVER_ADAPTER: z.string().default("0"), + RUN_OPS_LEGACY_DATABASE_WRITER_DRIVER_ADAPTER: z.string().default("0"), + RUN_OPS_LEGACY_DATABASE_REPLICA_DRIVER_ADAPTER: z.string().default("0"), // Control-plane cache relax knobs. Unset -> defaults (DEFAULT_CP_CACHE_TTL_MS / _MAX_ENTRIES). CONTROL_PLANE_CACHE_TTL_MS: z.coerce.number().int().optional(), CONTROL_PLANE_CACHE_MAX_ENTRIES: z.coerce.number().int().optional(), diff --git a/apps/webapp/app/utils/prismaErrors.ts b/apps/webapp/app/utils/prismaErrors.ts index 6c128b259d7..3fa6d53ada9 100644 --- a/apps/webapp/app/utils/prismaErrors.ts +++ b/apps/webapp/app/utils/prismaErrors.ts @@ -27,6 +27,25 @@ const INFRASTRUCTURE_PRISMA_CODES = new Set([ * (which both scrubs the message and is retryable by the SDK) instead of * folding `.message` into a client-facing error. */ +const CONNECTIVITY_ERRNO = new Set([ + "ECONNREFUSED", + "ENOTFOUND", + "ETIMEDOUT", + "ECONNRESET", + "EHOSTUNREACH", + "EPIPE", +]); +const CONNECTIVITY_MESSAGE = + /ECONNREFUSED|ENOTFOUND|ETIMEDOUT|ECONNRESET|EHOSTUNREACH|not reachable|can't reach database|connection terminated|server has closed|closed the connection|timed out fetching a new connection/i; + +function looksLikeConnectivityError(error: unknown): boolean { + const e = error as { code?: unknown; message?: unknown }; + if (typeof e?.code === "string" && CONNECTIVITY_ERRNO.has(e.code)) { + return true; + } + return typeof e?.message === "string" && CONNECTIVITY_MESSAGE.test(e.message); +} + export function isInfrastructureError(error: unknown): boolean { if ( error instanceof Prisma.PrismaClientInitializationError || @@ -37,10 +56,13 @@ export function isInfrastructureError(error: unknown): boolean { } if (error instanceof Prisma.PrismaClientKnownRequestError) { - return INFRASTRUCTURE_PRISMA_CODES.has(error.code); + if (INFRASTRUCTURE_PRISMA_CODES.has(error.code)) { + return true; + } + return error.code === "P2010" && looksLikeConnectivityError(error); } - return false; + return looksLikeConnectivityError(error); } // One-shot marker so a single infra error is logged exactly once: the client diff --git a/apps/webapp/package.json b/apps/webapp/package.json index 79ac530f3d0..ffcdf4e2099 100644 --- a/apps/webapp/package.json +++ b/apps/webapp/package.json @@ -175,6 +175,7 @@ "p-retry": "^4.6.1", "parse-duration": "^2.1.0", "pg": "8.15.6", + "@prisma/adapter-pg": "6.14.0", "posthog-js": "^1.93.3", "posthog-node": "5.35.6", "prism-react-renderer": "^2.3.1", @@ -243,6 +244,7 @@ "@types/marked": "^4.0.3", "@types/morgan": "^1.9.3", "@types/node-fetch": "^2.6.2", + "@types/pg": "^8.11.10", "@types/prismjs": "^1.26.0", "@types/qs": "^6.9.7", "@types/react": "18.2.69", diff --git a/internal-packages/database/prisma/schema.prisma b/internal-packages/database/prisma/schema.prisma index 8ea3ea2fc3a..545b0259539 100644 --- a/internal-packages/database/prisma/schema.prisma +++ b/internal-packages/database/prisma/schema.prisma @@ -8,7 +8,7 @@ generator client { provider = "prisma-client-js" output = "../generated/prisma" binaryTargets = ["native", "debian-openssl-1.1.x"] - previewFeatures = ["metrics"] + previewFeatures = ["metrics", "driverAdapters"] } model User { diff --git a/internal-packages/database/src/transaction.ts b/internal-packages/database/src/transaction.ts index fad07d1102b..4d271845393 100644 --- a/internal-packages/database/src/transaction.ts +++ b/internal-packages/database/src/transaction.ts @@ -37,12 +37,15 @@ export function isPrismaKnownError(error: unknown): error is PrismaClientKnownRe */ const retryCodes = ["P2024", "P2028", "P2034"]; +const ADAPTER_ACQUIRE_TIMEOUT = /timeout exceeded when trying to connect/i; + export function isPrismaRetriableError(error: unknown): boolean { - if (!isPrismaKnownError(error)) { - return false; + if (isPrismaKnownError(error)) { + return retryCodes.includes(error.code); } - return retryCodes.includes(error.code); + const message = (error as { message?: unknown })?.message; + return typeof message === "string" && ADAPTER_ACQUIRE_TIMEOUT.test(message); } /* diff --git a/internal-packages/run-ops-database/prisma/schema.prisma b/internal-packages/run-ops-database/prisma/schema.prisma index 4750efa392c..0a6c10dd034 100644 --- a/internal-packages/run-ops-database/prisma/schema.prisma +++ b/internal-packages/run-ops-database/prisma/schema.prisma @@ -7,7 +7,7 @@ generator client { provider = "prisma-client-js" output = "../generated/run-ops" binaryTargets = ["native", "debian-openssl-1.1.x"] - previewFeatures = ["metrics"] + previewFeatures = ["metrics", "driverAdapters"] } // ───────────────────────────────────────────────────────────────────────────── diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index f7532695b4e..124bf9ca755 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -391,6 +391,9 @@ importers: '@popperjs/core': specifier: ^2.11.8 version: 2.11.8 + '@prisma/adapter-pg': + specifier: 6.14.0 + version: 6.14.0 '@prisma/instrumentation': specifier: ^6.14.0 version: 6.14.0(@opentelemetry/api@1.9.1) @@ -854,6 +857,9 @@ importers: '@types/node-fetch': specifier: ^2.6.2 version: 2.6.2 + '@types/pg': + specifier: ^8.11.10 + version: 8.11.14 '@types/prismjs': specifier: ^1.26.0 version: 1.26.0 @@ -5679,6 +5685,9 @@ packages: '@posthog/types@1.376.4': resolution: {integrity: sha512-EoDEvA925lf6yxPpbP4wozlXgu4b9WEqxZlFBUDd4k2akP5R/RWyHpvQT8aYyfY6BtSLn8TnVwxPQOM4b90isA==} + '@prisma/adapter-pg@6.14.0': + resolution: {integrity: sha512-heUCNPZ3f2Iv/JId280HCoN7NECWkYckC3K1cOKpTRRjiJv0mN0w99V91vBq7aHHTtlOVpfDPQbUMNAjMegIWg==} + '@prisma/client@6.14.0': resolution: {integrity: sha512-8E/Nk3eL5g7RQIg/LUj1ICyDmhD053STjxrPxUtCRybs2s/2sOEcx9NpITuAOPn07HEpWBfhAVe1T/HYWXUPOw==} engines: {node: '>=18.18'} @@ -5700,6 +5709,9 @@ packages: '@prisma/debug@6.14.0': resolution: {integrity: sha512-j4Lf+y+5QIJgQD4sJWSbkOD7geKx9CakaLp/TyTy/UDu9Wo0awvWCBH/BAxTHUaCpIl9USA5VS/KJhDqKJSwug==} + '@prisma/driver-adapter-utils@6.14.0': + resolution: {integrity: sha512-On9vTNiJ7J/O1kVqedLtfdhhrfRYprkUOhxjlmmWEv12WNdG6v5x4PsrfZXdBtZqRZaqK1i1TigO6IdtYh8z+A==} + '@prisma/engines-version@6.14.0-25.717184b7b35ea05dfa71a3236b7af656013e1e49': resolution: {integrity: sha512-EgN9ODJpiX45yvwcngoStp3uQPJ3l+AEVoQ6dMMO2QvmwIlnxfApzKmJQExzdo7/hqQANrz5txHJdGYHzOnGHA==} @@ -13324,9 +13336,6 @@ packages: pg-protocol@1.10.3: resolution: {integrity: sha512-6DIBgBQaTKDJyxnXaLiLR8wBpQQcGWuAESkRBX/t6OwA8YsqP+iVSiond2EDy6Y/dsGk8rh/jtax3js5NeV7JQ==} - pg-protocol@1.9.5: - resolution: {integrity: sha512-DYTWtWpfd5FOro3UnAfwvhD8jh59r2ig8bPtc9H8Ds7MscE/9NYruUQWFAOuraRl29jwcT2kyMFQ3MxeaVjUhg==} - pg-types@2.2.0: resolution: {integrity: sha512-qTAAlrEsl8s4OiEQY69wDvcMIdQN6wdz5ojQiOy6YRMuynxenON0O5oCpJI6lshc6scgAY8qvJ2On/p+CXY0GA==} engines: {node: '>=4'} @@ -20315,6 +20324,14 @@ snapshots: '@posthog/types@1.376.4': {} + '@prisma/adapter-pg@6.14.0': + dependencies: + '@prisma/driver-adapter-utils': 6.14.0 + pg: 8.15.6 + postgres-array: 3.0.4 + transitivePeerDependencies: + - pg-native + '@prisma/client@6.14.0(prisma@6.14.0(magicast@0.3.5)(typescript@7.0.2))(typescript@7.0.2)': optionalDependencies: prisma: 6.14.0(magicast@0.3.5)(typescript@7.0.2) @@ -20340,6 +20357,10 @@ snapshots: '@prisma/debug@6.14.0': {} + '@prisma/driver-adapter-utils@6.14.0': + dependencies: + '@prisma/debug': 6.14.0 + '@prisma/engines-version@6.14.0-25.717184b7b35ea05dfa71a3236b7af656013e1e49': {} '@prisma/engines@6.14.0': @@ -23169,7 +23190,7 @@ snapshots: '@types/pg@8.11.14': dependencies: '@types/node': 24.13.3 - pg-protocol: 1.9.5 + pg-protocol: 1.10.3 pg-types: 4.0.2 '@types/pg@8.6.1': @@ -29209,8 +29230,6 @@ snapshots: pg-protocol@1.10.3: {} - pg-protocol@1.9.5: {} - pg-types@2.2.0: dependencies: pg-int8: 1.0.1 From f07ac7c9e9ab366a0cf3c483306529780db6d4c5 Mon Sep 17 00:00:00 2001 From: Eric Allam Date: Sat, 8 Aug 2026 17:25:32 +0100 Subject: [PATCH 2/5] fix(webapp): address review on driver-adapter pools - Pass the per-client resolved connection limit into the adapter pool instead of always using DATABASE_CONNECTION_LIMIT, so per-client overrides (e.g. RUN_OPS_DATABASE_READ_REPLICA_CONNECTION_LIMIT) are honored on the adapter path. - Build the adapter pool from the base DSN (drops prisma-only URL params the pg driver ignores and the duplicate application_name). - Scope the connectivity message match to 'database not reachable' so a generic 'not reachable' error is no longer misclassified as infrastructure. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../prisma-driver-adapter-per-client.md | 2 +- apps/webapp/app/db.server.ts | 25 +++++++++++-------- apps/webapp/app/utils/prismaErrors.ts | 2 +- 3 files changed, 17 insertions(+), 12 deletions(-) diff --git a/.server-changes/prisma-driver-adapter-per-client.md b/.server-changes/prisma-driver-adapter-per-client.md index 6fdd88138e7..91b4fc5c6da 100644 --- a/.server-changes/prisma-driver-adapter-per-client.md +++ b/.server-changes/prisma-driver-adapter-per-client.md @@ -3,4 +3,4 @@ area: webapp type: improvement --- -Each database client can now optionally connect through the node-postgres driver, configurable independently per client and off by default, so default behavior is unchanged. +Groundwork for an alternative database connection driver, gated behind configuration and disabled by default, so there is no change to default behavior. diff --git a/apps/webapp/app/db.server.ts b/apps/webapp/app/db.server.ts index f2a2d93b98a..20583e1e2ef 100644 --- a/apps/webapp/app/db.server.ts +++ b/apps/webapp/app/db.server.ts @@ -440,11 +440,12 @@ function getClient() { function buildDriverAdapterPool( connectionString: string, clientType: string, - poolTimeoutSeconds: number + poolTimeoutSeconds: number, + connectionLimit: number ): PrismaPg { const pool = new Pool({ connectionString, - max: env.DATABASE_CONNECTION_LIMIT, + max: connectionLimit, connectionTimeoutMillis: poolTimeoutSeconds * 1000, application_name: env.SERVICE_NAME, }); @@ -542,9 +543,10 @@ export function buildWriterClient({ const client = useDriverAdapter ? new PrismaClient({ adapter: buildDriverAdapterPool( - databaseUrl.href, + url, clientType, - poolTimeout ?? env.DATABASE_POOL_TIMEOUT + poolTimeout ?? env.DATABASE_POOL_TIMEOUT, + env.DATABASE_CONNECTION_LIMIT ), log: logConfig, }) @@ -712,9 +714,10 @@ export function buildReplicaClient({ const replicaClient = useDriverAdapter ? new PrismaClient({ adapter: buildDriverAdapterPool( - replicaUrl.href, + url, clientType, - poolTimeout ?? env.DATABASE_POOL_TIMEOUT + poolTimeout ?? env.DATABASE_POOL_TIMEOUT, + env.DATABASE_CONNECTION_LIMIT ), log: logConfig, }) @@ -805,9 +808,10 @@ function buildRunOpsWriterClient({ const client = useDriverAdapter ? new RunOpsPrismaClient({ adapter: buildDriverAdapterPool( - databaseUrl.href, + url, clientType, - env.RUN_OPS_DATABASE_WRITER_POOL_TIMEOUT ?? env.DATABASE_POOL_TIMEOUT + env.RUN_OPS_DATABASE_WRITER_POOL_TIMEOUT ?? env.DATABASE_POOL_TIMEOUT, + env.DATABASE_CONNECTION_LIMIT ), log: [ { emit: "event", level: "error" }, @@ -885,9 +889,10 @@ function buildRunOpsReplicaClient({ const client = useDriverAdapter ? new RunOpsPrismaClient({ adapter: buildDriverAdapterPool( - replicaUrl.href, + url, clientType, - env.RUN_OPS_DATABASE_READ_REPLICA_POOL_TIMEOUT ?? env.DATABASE_POOL_TIMEOUT + env.RUN_OPS_DATABASE_READ_REPLICA_POOL_TIMEOUT ?? env.DATABASE_POOL_TIMEOUT, + env.RUN_OPS_DATABASE_READ_REPLICA_CONNECTION_LIMIT ?? env.DATABASE_CONNECTION_LIMIT ), log: [ { emit: "event", level: "error" }, diff --git a/apps/webapp/app/utils/prismaErrors.ts b/apps/webapp/app/utils/prismaErrors.ts index 3fa6d53ada9..c57687d13ce 100644 --- a/apps/webapp/app/utils/prismaErrors.ts +++ b/apps/webapp/app/utils/prismaErrors.ts @@ -36,7 +36,7 @@ const CONNECTIVITY_ERRNO = new Set([ "EPIPE", ]); const CONNECTIVITY_MESSAGE = - /ECONNREFUSED|ENOTFOUND|ETIMEDOUT|ECONNRESET|EHOSTUNREACH|not reachable|can't reach database|connection terminated|server has closed|closed the connection|timed out fetching a new connection/i; + /ECONNREFUSED|ENOTFOUND|ETIMEDOUT|ECONNRESET|EHOSTUNREACH|database not reachable|can't reach database|connection terminated|server has closed the connection|timed out fetching a new connection/i; function looksLikeConnectivityError(error: unknown): boolean { const e = error as { code?: unknown; message?: unknown }; From 1c72e5777d37d71a96f359e345e2e66feda24fc0 Mon Sep 17 00:00:00 2001 From: Eric Allam Date: Sat, 8 Aug 2026 19:38:15 +0100 Subject: [PATCH 3/5] fix(webapp,database): address Devin review on the driver-adapter path - Pass the datasource schema (?schema=) to PrismaPg as its {schema} option so custom-schema installs keep talking to the right schema on the adapter (the adapter does not honor ?schema= in the connection string). - Set disposeExternalPool: true so $disconnect() closes the pg pool instead of leaking sockets, matching the engine-driver path. - Guard the two $metrics consumers (the /metrics route and the OTel batch observable callback) so a client on a driver adapter degrades to empty metrics instead of failing the scrape / rejecting the callback. - isPrismaRetriableError checks the adapter acquire-timeout message independently of the coded-error branch, so the pool-acquire retry still engages if the timeout arrives wrapped as a coded error. Co-Authored-By: Claude Opus 4.8 (1M context) --- apps/webapp/app/db.server.ts | 10 +++++++++- apps/webapp/app/routes/metrics.ts | 7 ++++++- apps/webapp/app/v3/tracer.server.ts | 8 +++++++- internal-packages/database/src/transaction.ts | 4 ++-- 4 files changed, 24 insertions(+), 5 deletions(-) diff --git a/apps/webapp/app/db.server.ts b/apps/webapp/app/db.server.ts index 20583e1e2ef..249b8514569 100644 --- a/apps/webapp/app/db.server.ts +++ b/apps/webapp/app/db.server.ts @@ -456,7 +456,15 @@ function buildDriverAdapterPool( ignoreError: true, }); }); - return new PrismaPg(pool); + + let schema: string | undefined; + try { + schema = new URL(connectionString).searchParams.get("schema") ?? undefined; + } catch { + schema = undefined; + } + + return new PrismaPg(pool, { schema, disposeExternalPool: true }); } // Generalized writer builder shared by the control-plane client and the run-ops diff --git a/apps/webapp/app/routes/metrics.ts b/apps/webapp/app/routes/metrics.ts index 62d8befe5f6..038d9e75fc0 100644 --- a/apps/webapp/app/routes/metrics.ts +++ b/apps/webapp/app/routes/metrics.ts @@ -14,7 +14,12 @@ 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, ""); + let prismaMetrics = ""; + try { + prismaMetrics = (await prisma.$metrics.prometheus()).replace(/^\s*[\r\n]/gm, ""); + } catch { + prismaMetrics = ""; + } const coreMetrics = await metricsRegister.metrics(); // Order matters, core metrics end with `# EOF`, prisma metrics don't diff --git a/apps/webapp/app/v3/tracer.server.ts b/apps/webapp/app/v3/tracer.server.ts index 81ae47f3b1d..d91c9ee2ff7 100644 --- a/apps/webapp/app/v3/tracer.server.ts +++ b/apps/webapp/app/v3/tracer.server.ts @@ -551,7 +551,13 @@ function configurePrismaMetrics({ meter }: { meter: Meter }) { meter.addBatchObservableCallback( async (res) => { - const { counters, gauges, histograms } = await readPrismaMetrics(); + let prismaMetrics: Awaited>; + try { + prismaMetrics = await readPrismaMetrics(); + } catch { + return; + } + const { counters, gauges, histograms } = prismaMetrics; // Observe counters res.observe(queriesTotal, counters.queriesTotal); diff --git a/internal-packages/database/src/transaction.ts b/internal-packages/database/src/transaction.ts index 4d271845393..c58bf1a0405 100644 --- a/internal-packages/database/src/transaction.ts +++ b/internal-packages/database/src/transaction.ts @@ -40,8 +40,8 @@ const retryCodes = ["P2024", "P2028", "P2034"]; const ADAPTER_ACQUIRE_TIMEOUT = /timeout exceeded when trying to connect/i; export function isPrismaRetriableError(error: unknown): boolean { - if (isPrismaKnownError(error)) { - return retryCodes.includes(error.code); + if (isPrismaKnownError(error) && retryCodes.includes(error.code)) { + return true; } const message = (error as { message?: unknown })?.message; From 24286a0c2213580dbf2f34baf44c719d463052b7 Mon Sep 17 00:00:00 2001 From: Eric Allam Date: Sat, 8 Aug 2026 19:54:03 +0100 Subject: [PATCH 4/5] fix(database): retry adapter pool-acquire timeouts in $transaction The retry decision used retryCodes.includes(error.code) directly, inside the isPrismaKnownError branch, so the broadened isPrismaRetriableError check never governed retries. Route the retry decision through isPrismaRetriableError so the adapter's pool-acquire timeout is retried like P2024 was, while keeping prismaError()/swallow behavior for coded errors only. Co-Authored-By: Claude Opus 4.8 (1M context) --- internal-packages/database/src/transaction.ts | 16 ++++++++-------- 1 file changed, 8 insertions(+), 8 deletions(-) diff --git a/internal-packages/database/src/transaction.ts b/internal-packages/database/src/transaction.ts index c58bf1a0405..806ae604038 100644 --- a/internal-packages/database/src/transaction.ts +++ b/internal-packages/database/src/transaction.ts @@ -93,15 +93,15 @@ export async function $transaction( try { return await (prisma as PrismaClient).$transaction(fn, options); } catch (error) { - if (isPrismaKnownError(error)) { - if ( - retryCodes.includes(error.code) && - typeof options?.maxRetries === "number" && - attempt < options.maxRetries - ) { - return $transaction(prisma, fn, prismaError, options, attempt + 1); - } + if ( + isPrismaRetriableError(error) && + typeof options?.maxRetries === "number" && + attempt < options.maxRetries + ) { + return $transaction(prisma, fn, prismaError, options, attempt + 1); + } + if (isPrismaKnownError(error)) { prismaError(error); if (options?.swallowPrismaErrors) { From afbd62b7b4f1a6c9c456de052e1cf2b28f05d221 Mon Sep 17 00:00:00 2001 From: Eric Allam Date: Sat, 8 Aug 2026 20:10:17 +0100 Subject: [PATCH 5/5] fix(webapp): retry driver-adapter connectivity errors in the mollifier drainer isRetryablePgError only recognized the Rust engine's DB-unreachable shapes (P1001 / "Can't reach database server"). Under a driver adapter the same outage surfaces as P2010 "Database not reachable" / ECONNREFUSED / ENOTFOUND, so buffered runs were permanently failed on a transient outage. Reuse the shared looksLikeConnectivityError predicate (now exported from prismaErrors) so those are retried too. Co-Authored-By: Claude Opus 4.8 (1M context) --- apps/webapp/app/utils/prismaErrors.ts | 2 +- apps/webapp/app/v3/mollifier/mollifierDrainerHandler.server.ts | 2 ++ 2 files changed, 3 insertions(+), 1 deletion(-) diff --git a/apps/webapp/app/utils/prismaErrors.ts b/apps/webapp/app/utils/prismaErrors.ts index c57687d13ce..e1f11d55129 100644 --- a/apps/webapp/app/utils/prismaErrors.ts +++ b/apps/webapp/app/utils/prismaErrors.ts @@ -38,7 +38,7 @@ const CONNECTIVITY_ERRNO = new Set([ const CONNECTIVITY_MESSAGE = /ECONNREFUSED|ENOTFOUND|ETIMEDOUT|ECONNRESET|EHOSTUNREACH|database not reachable|can't reach database|connection terminated|server has closed the connection|timed out fetching a new connection/i; -function looksLikeConnectivityError(error: unknown): boolean { +export function looksLikeConnectivityError(error: unknown): boolean { const e = error as { code?: unknown; message?: unknown }; if (typeof e?.code === "string" && CONNECTIVITY_ERRNO.has(e.code)) { return true; diff --git a/apps/webapp/app/v3/mollifier/mollifierDrainerHandler.server.ts b/apps/webapp/app/v3/mollifier/mollifierDrainerHandler.server.ts index fd9daa5d615..302ac0fa9f2 100644 --- a/apps/webapp/app/v3/mollifier/mollifierDrainerHandler.server.ts +++ b/apps/webapp/app/v3/mollifier/mollifierDrainerHandler.server.ts @@ -7,6 +7,7 @@ import type { MollifierDrainerTerminalFailureHandler, } from "@trigger.dev/redis-worker"; import { logger } from "~/services/logger.server"; +import { looksLikeConnectivityError } from "~/utils/prismaErrors"; import { recordRunDebugLog } from "~/v3/eventRepository/index.server"; import { PerformTaskRunAlertsService } from "~/v3/services/alerts/performTaskRunAlerts.server"; import { startSpan } from "~/v3/tracing.server"; @@ -29,6 +30,7 @@ export function isRetryablePgError(err: unknown): boolean { if (msg.includes("Can't reach database server")) return true; if (msg.includes("Connection lost")) return true; if (msg.includes("ECONNRESET")) return true; + if (looksLikeConnectivityError(err)) return true; return false; }