diff --git a/doc/DATABASE.md b/doc/DATABASE.md index b5034991b1..6c80f1f500 100644 --- a/doc/DATABASE.md +++ b/doc/DATABASE.md @@ -113,7 +113,18 @@ DATABASE_URL=postgres://postgres.[PROJECT-REF]:[PASSWORD]@aws-0-[REGION].pooler. DATABASE_MIGRATION_URL=postgres://postgres.[PROJECT-REF]:[PASSWORD]@aws-0-[REGION].pooler.supabase.com:5432/postgres ``` -If your hosted database requires transaction-pooling-only connections, use a direct or session-pooled connection for Paperclip until runtime pooling support is documented in this guide. Do not edit database client source files as part of deployment setup. +If your hosted database requires transaction-pooling-only connections (pgbouncer transaction mode, Supavisor port 6543, Neon `-pooler` endpoints), set `DATABASE_PREPARED_STATEMENTS=false` so the client does not rely on session-scoped prepared statements, and keep `DATABASE_MIGRATION_URL` on a direct connection. Do not edit database client source files as part of deployment setup. + +### Client tuning (optional) + +All of these are optional; when unset, the driver defaults apply and behavior is unchanged — typical self-hosted setups need none of them: + +```sh +DATABASE_PREPARED_STATEMENTS=false # required for transaction-mode poolers; default: enabled +DATABASE_POOL_MAX=25 # connection pool size; default: 10 +DATABASE_IDLE_TIMEOUT_SECONDS=60 # close idle pooled connections; default: keep open +DATABASE_CONNECT_TIMEOUT_SECONDS=10 # default: 30 +``` ### Push the schema diff --git a/docs/deploy/database.md b/docs/deploy/database.md index a454de9b33..0d4ad5731e 100644 --- a/docs/deploy/database.md +++ b/docs/deploy/database.md @@ -56,16 +56,14 @@ For production, use a hosted provider like [Supabase](https://supabase.com/). Use the **direct connection** (port 5432) for migrations and the **pooled connection** (port 6543) for the application. -If using connection pooling, disable prepared statements: +If using connection pooling (transaction mode), disable prepared statements via the environment — no source edits needed: -```ts -// packages/db/src/client.ts -export function createDb(url: string) { - const sql = postgres(url, { prepare: false }); - return drizzlePg(sql, { schema }); -} +```sh +DATABASE_PREPARED_STATEMENTS=false ``` +Related optional client tuning (driver defaults apply when unset): `DATABASE_POOL_MAX`, `DATABASE_IDLE_TIMEOUT_SECONDS`, `DATABASE_CONNECT_TIMEOUT_SECONDS`. + ## Switching Between Modes | `DATABASE_URL` | Mode | diff --git a/packages/db/src/client-options.test.ts b/packages/db/src/client-options.test.ts new file mode 100644 index 0000000000..267d14dc17 --- /dev/null +++ b/packages/db/src/client-options.test.ts @@ -0,0 +1,57 @@ +import { describe, expect, it } from "vitest"; +import { databaseClientOptionsFromEnv, postgresJsOptions } from "./client.js"; + +describe("databaseClientOptionsFromEnv", () => { + it("returns no options when nothing is set, preserving driver defaults", () => { + expect(databaseClientOptionsFromEnv({})).toEqual({}); + expect(postgresJsOptions(databaseClientOptionsFromEnv({}))).toEqual({}); + }); + + it("ignores empty values", () => { + expect( + databaseClientOptionsFromEnv({ + DATABASE_PREPARED_STATEMENTS: "", + DATABASE_POOL_MAX: "", + }), + ).toEqual({}); + }); + + it("parses prepared-statement toggles", () => { + expect(databaseClientOptionsFromEnv({ DATABASE_PREPARED_STATEMENTS: "false" })).toEqual({ prepare: false }); + expect(databaseClientOptionsFromEnv({ DATABASE_PREPARED_STATEMENTS: "0" })).toEqual({ prepare: false }); + expect(databaseClientOptionsFromEnv({ DATABASE_PREPARED_STATEMENTS: "true" })).toEqual({ prepare: true }); + expect(databaseClientOptionsFromEnv({ DATABASE_PREPARED_STATEMENTS: "TRUE" })).toEqual({ prepare: true }); + }); + + it("parses pool and timeout settings", () => { + expect( + databaseClientOptionsFromEnv({ + DATABASE_POOL_MAX: "25", + DATABASE_IDLE_TIMEOUT_SECONDS: "60", + DATABASE_CONNECT_TIMEOUT_SECONDS: "10", + }), + ).toEqual({ maxConnections: 25, idleTimeoutSeconds: 60, connectTimeoutSeconds: 10 }); + }); + + it("rejects malformed values instead of silently ignoring them", () => { + expect(() => databaseClientOptionsFromEnv({ DATABASE_PREPARED_STATEMENTS: "maybe" })).toThrow( + /DATABASE_PREPARED_STATEMENTS/, + ); + expect(() => databaseClientOptionsFromEnv({ DATABASE_POOL_MAX: "0" })).toThrow(/DATABASE_POOL_MAX/); + expect(() => databaseClientOptionsFromEnv({ DATABASE_POOL_MAX: "-3" })).toThrow(/DATABASE_POOL_MAX/); + expect(() => databaseClientOptionsFromEnv({ DATABASE_CONNECT_TIMEOUT_SECONDS: "1.5" })).toThrow( + /DATABASE_CONNECT_TIMEOUT_SECONDS/, + ); + }); + + it("maps to postgres.js option names", () => { + expect( + postgresJsOptions({ + prepare: false, + maxConnections: 25, + idleTimeoutSeconds: 60, + connectTimeoutSeconds: 10, + }), + ).toEqual({ prepare: false, max: 25, idle_timeout: 60, connect_timeout: 10 }); + }); +}); diff --git a/packages/db/src/client.ts b/packages/db/src/client.ts index f2ec3fa32c..99b8f8a17d 100644 --- a/packages/db/src/client.ts +++ b/packages/db/src/client.ts @@ -45,8 +45,72 @@ export type MigrationState = reason: "no-migration-journal-empty-db" | "no-migration-journal-non-empty-db" | "pending-migrations"; }; -export function createDb(url: string) { - const sql = postgres(url); +export interface DatabaseClientOptions { + /** + * postgres.js `prepare`. Set false when connecting through a + * transaction-mode pooler (pgbouncer / Neon `-pooler` endpoints / + * Supabase Supavisor transaction ports) so the client does not rely on + * session-scoped prepared statements. Defaults to the driver default + * (enabled), preserving existing behavior on direct connections. + */ + prepare?: boolean; + /** postgres.js `max` — connection pool size (driver default: 10). */ + maxConnections?: number; + /** postgres.js `idle_timeout` in seconds (driver default: disabled). */ + idleTimeoutSeconds?: number; + /** postgres.js `connect_timeout` in seconds (driver default: 30). */ + connectTimeoutSeconds?: number; +} + +function envBoolean(env: NodeJS.ProcessEnv, name: string): boolean | undefined { + const value = env[name]?.trim().toLowerCase(); + if (value === undefined || value === "") return undefined; + if (value === "true" || value === "1") return true; + if (value === "false" || value === "0") return false; + throw new Error(`${name} must be "true" or "false", got: ${env[name]}`); +} + +function envPositiveInteger(env: NodeJS.ProcessEnv, name: string): number | undefined { + const value = env[name]?.trim(); + if (value === undefined || value === "") return undefined; + if (!/^[1-9]\d*$/.test(value)) { + throw new Error(`${name} must be a positive integer, got: ${env[name]}`); + } + return Number.parseInt(value, 10); +} + +/** + * Database client tuning from the environment, so hosted deployments can + * adapt to their connection topology (pooled endpoints, network latency) + * without editing source. Every variable is optional; when unset the + * driver defaults apply and behavior is identical to a bare + * `postgres(url)` — self-hosted setups need none of these. + */ +export function databaseClientOptionsFromEnv(env: NodeJS.ProcessEnv = process.env): DatabaseClientOptions { + const options: DatabaseClientOptions = {}; + const prepare = envBoolean(env, "DATABASE_PREPARED_STATEMENTS"); + if (prepare !== undefined) options.prepare = prepare; + const maxConnections = envPositiveInteger(env, "DATABASE_POOL_MAX"); + if (maxConnections !== undefined) options.maxConnections = maxConnections; + const idleTimeoutSeconds = envPositiveInteger(env, "DATABASE_IDLE_TIMEOUT_SECONDS"); + if (idleTimeoutSeconds !== undefined) options.idleTimeoutSeconds = idleTimeoutSeconds; + const connectTimeoutSeconds = envPositiveInteger(env, "DATABASE_CONNECT_TIMEOUT_SECONDS"); + if (connectTimeoutSeconds !== undefined) options.connectTimeoutSeconds = connectTimeoutSeconds; + return options; +} + +export function postgresJsOptions(options: DatabaseClientOptions): Record { + const driverOptions: Record = {}; + if (options.prepare !== undefined) driverOptions.prepare = options.prepare; + if (options.maxConnections !== undefined) driverOptions.max = options.maxConnections; + if (options.idleTimeoutSeconds !== undefined) driverOptions.idle_timeout = options.idleTimeoutSeconds; + if (options.connectTimeoutSeconds !== undefined) driverOptions.connect_timeout = options.connectTimeoutSeconds; + return driverOptions; +} + +export function createDb(url: string, options?: DatabaseClientOptions) { + const resolved = options ?? databaseClientOptionsFromEnv(); + const sql = postgres(url, postgresJsOptions(resolved)); return drizzlePg(sql, { schema }); } diff --git a/server/src/services/attention.ts b/server/src/services/attention.ts index d6dae780f7..f43729b7f4 100644 --- a/server/src/services/attention.ts +++ b/server/src/services/attention.ts @@ -1023,8 +1023,10 @@ export function attentionService(db: Db, serviceOptions: AttentionServiceOptions if (options.all && !options.queue && !options.allowUnscopedAll) { throw badRequest("all requires a queue filter"); } - const prefix = await companyPrefix(db, companyId); - const dismissals = await dismissalByKey(db, companyId, options.userId); + const [prefix, dismissals] = await Promise.all([ + companyPrefix(db, companyId), + dismissalByKey(db, companyId, options.userId), + ]); const includeDismissed = options.includeDismissed === true; const now = serviceOptions.now?.() ?? Date.now(); const collected: AttentionItem[] = []; @@ -1126,9 +1128,11 @@ export function attentionService(db: Db, serviceOptions: AttentionServiceOptions )) .orderBy(desc(issueThreadInteractions.updatedAt), desc(issueThreadInteractions.id)); const visibleInteractionRows = collapsePendingConfirmationsToNewest(interactionRows); - const interactionIssueMap = await issueSummaryMap(db, companyId, visibleInteractionRows.map((row) => row.issueId)); - const interactionImageMap = await issueImageMap(db, companyId, visibleInteractionRows.map((row) => row.issueId)); - const interactionPlanDocumentMap = await planDocumentMap(db, companyId, visibleInteractionRows.map((row) => row.issueId)); + const [interactionIssueMap, interactionImageMap, interactionPlanDocumentMap] = await Promise.all([ + issueSummaryMap(db, companyId, visibleInteractionRows.map((row) => row.issueId)), + issueImageMap(db, companyId, visibleInteractionRows.map((row) => row.issueId)), + planDocumentMap(db, companyId, visibleInteractionRows.map((row) => row.issueId)), + ]); for (const interaction of visibleInteractionRows) { const issue = interactionIssueMap.get(interaction.issueId) ?? null; @@ -1194,16 +1198,18 @@ export function attentionService(db: Db, serviceOptions: AttentionServiceOptions const openDecisions = options.all ? await openDecisionQuery : await openDecisionQuery.limit(openDecisionLimit); - const decisionIssueMap = await issueSummaryMap(db, companyId, openDecisions.map((decision) => decision.originIssueId)); // Bundle titles let the feed render a single "Agent proposed N decisions" // group header over sibling decisions (v1 still decides each independently). const bundleIds = [...new Set(openDecisions.map((decision) => decision.bundleId).filter((value): value is string => Boolean(value)))]; const bundleTitleMap = new Map(); - if (bundleIds.length > 0) { - const bundleRows = await db.select({ id: decisionBundles.id, title: decisionBundles.title }) - .from(decisionBundles).where(and(eq(decisionBundles.companyId, companyId), inArray(decisionBundles.id, bundleIds))); - for (const row of bundleRows) bundleTitleMap.set(row.id, row.title); - } + const [decisionIssueMap, bundleRows] = await Promise.all([ + issueSummaryMap(db, companyId, openDecisions.map((decision) => decision.originIssueId)), + bundleIds.length > 0 + ? db.select({ id: decisionBundles.id, title: decisionBundles.title }) + .from(decisionBundles).where(and(eq(decisionBundles.companyId, companyId), inArray(decisionBundles.id, bundleIds))) + : Promise.resolve([]), + ]); + for (const row of bundleRows) bundleTitleMap.set(row.id, row.title); for (const decision of openDecisions) { const issue = decisionIssueMap.get(decision.originIssueId) ?? null; add(createItem({ @@ -1301,12 +1307,14 @@ export function attentionService(db: Db, serviceOptions: AttentionServiceOptions inArray(issueRecoveryActions.ownerType, [...HUMAN_RECOVERY_OWNER_TYPES]), )) .orderBy(desc(issueRecoveryActions.updatedAt), desc(issueRecoveryActions.id)); - const recoveryIssueMap = await issueSummaryMap( - db, - companyId, - recoveryRows.flatMap((row) => [row.sourceIssueId, row.recoveryIssueId]), - ); - const recoveryImageMap = await issueImageMap(db, companyId, recoveryRows.map((row) => row.sourceIssueId)); + const [recoveryIssueMap, recoveryImageMap] = await Promise.all([ + issueSummaryMap( + db, + companyId, + recoveryRows.flatMap((row) => [row.sourceIssueId, row.recoveryIssueId]), + ), + issueImageMap(db, companyId, recoveryRows.map((row) => row.sourceIssueId)), + ]); for (const recovery of recoveryRows) { const sourceIssue = recoveryIssueMap.get(recovery.sourceIssueId) ?? null; @@ -1378,9 +1386,11 @@ export function attentionService(db: Db, serviceOptions: AttentionServiceOptions notInArray(issues.status, [...PRODUCTIVITY_REVIEW_TERMINAL_STATUSES]), )) .orderBy(desc(issues.updatedAt), desc(issues.id)); - const productivitySourceMap = await issueSummaryMap(db, companyId, productivityRows.map((row) => row.originId)); - const productivityReviewMap = await issueSummaryMap(db, companyId, productivityRows.map((row) => row.id)); - const productivityImageMap = await issueImageMap(db, companyId, productivityRows.map((row) => row.id)); + const [productivitySourceMap, productivityReviewMap, productivityImageMap] = await Promise.all([ + issueSummaryMap(db, companyId, productivityRows.map((row) => row.originId)), + issueSummaryMap(db, companyId, productivityRows.map((row) => row.id)), + issueImageMap(db, companyId, productivityRows.map((row) => row.id)), + ]); for (const review of productivityRows) { const reviewIssue = productivityReviewMap.get(review.id); @@ -1427,14 +1437,16 @@ export function attentionService(db: Db, serviceOptions: AttentionServiceOptions const terminalBlockerIssueIds = typedBlockedIssues .map((issue) => issue.blockerAttention?.terminalBlockerIssueId) .filter((issueId): issueId is string => Boolean(issueId)); - const blockedIssueSummaries = await issueSummaryMap(db, companyId, blockedIssues.map((issue) => issue.id)); - const terminalBlockerSummaries = await issueSummaryMap(db, companyId, terminalBlockerIssueIds); - const blockerImageMap = await issueImageMap( - db, - companyId, - [...blockedIssues.map((issue) => issue.id), ...terminalBlockerIssueIds], - ); - const blockingIssues = await blockingIssueMap(db, companyId, blockedIssues.map((issue) => issue.id)); + const [blockedIssueSummaries, terminalBlockerSummaries, blockerImageMap, blockingIssues] = await Promise.all([ + issueSummaryMap(db, companyId, blockedIssues.map((issue) => issue.id)), + issueSummaryMap(db, companyId, terminalBlockerIssueIds), + issueImageMap( + db, + companyId, + [...blockedIssues.map((issue) => issue.id), ...terminalBlockerIssueIds], + ), + blockingIssueMap(db, companyId, blockedIssues.map((issue) => issue.id)), + ]); const terminalCandidates = new Map [row.issueId, row.approvalId])); - const reviewIssueMap = await issueSummaryMap(db, companyId, reviewIssueIds); - const reviewImageMap = await issueImageMap(db, companyId, reviewIssueIds); + const [reviewIssueMap, reviewImageMap] = await Promise.all([ + issueSummaryMap(db, companyId, reviewIssueIds), + issueImageMap(db, companyId, reviewIssueIds), + ]); for (const review of reviewRows) { const state = parseIssueExecutionState(review.executionState); @@ -1634,37 +1648,39 @@ export function attentionService(db: Db, serviceOptions: AttentionServiceOptions } const failedRows = [...latestExhaustedByRunId.values()]; const failedIssueIds = failedRows.map((row) => readRunIssueId(row.contextSnapshot)); - const failedIssueMap = await issueSummaryMap( - db, - companyId, - failedIssueIds, - ); - const failedImageMap = await issueImageMap(db, companyId, failedIssueIds); const failedAgentIds = [...new Set(failedRows.map((row) => row.agentId))]; const oldestFailedRunCreatedAt = failedRows.reduce((oldest, row) => { if (!oldest || row.createdAt < oldest) return row.createdAt; return oldest; }, null); + const [failedIssueMap, failedImageMap, newerRuns] = await Promise.all([ + issueSummaryMap( + db, + companyId, + failedIssueIds, + ), + issueImageMap(db, companyId, failedIssueIds), + oldestFailedRunCreatedAt && failedAgentIds.length > 0 + ? db + .select({ + agentId: heartbeatRuns.agentId, + createdAt: heartbeatRuns.createdAt, + contextSnapshot: heartbeatRuns.contextSnapshot, + }) + .from(heartbeatRuns) + .where(and( + eq(heartbeatRuns.companyId, companyId), + inArray(heartbeatRuns.agentId, failedAgentIds), + gt(heartbeatRuns.createdAt, oldestFailedRunCreatedAt), + )) + : Promise.resolve([]), + ]); const latestRunCreatedAtByKey = new Map(); - if (oldestFailedRunCreatedAt && failedAgentIds.length > 0) { - const newerRuns = await db - .select({ - agentId: heartbeatRuns.agentId, - createdAt: heartbeatRuns.createdAt, - contextSnapshot: heartbeatRuns.contextSnapshot, - }) - .from(heartbeatRuns) - .where(and( - eq(heartbeatRuns.companyId, companyId), - inArray(heartbeatRuns.agentId, failedAgentIds), - gt(heartbeatRuns.createdAt, oldestFailedRunCreatedAt), - )); - for (const newerRun of newerRuns) { - const newerRunKey = `${newerRun.agentId}:${readRunIssueId(newerRun.contextSnapshot) ?? ""}`; - const latestCreatedAt = latestRunCreatedAtByKey.get(newerRunKey); - if (!latestCreatedAt || newerRun.createdAt > latestCreatedAt) { - latestRunCreatedAtByKey.set(newerRunKey, newerRun.createdAt); - } + for (const newerRun of newerRuns) { + const newerRunKey = `${newerRun.agentId}:${readRunIssueId(newerRun.contextSnapshot) ?? ""}`; + const latestCreatedAt = latestRunCreatedAtByKey.get(newerRunKey); + if (!latestCreatedAt || newerRun.createdAt > latestCreatedAt) { + latestRunCreatedAtByKey.set(newerRunKey, newerRun.createdAt); } } for (const run of failedRows) {