diff --git a/doc/DATABASE.md b/doc/DATABASE.md index cb341f8ea79..2032268da2d 100644 --- a/doc/DATABASE.md +++ b/doc/DATABASE.md @@ -458,14 +458,14 @@ 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 (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. +If your transaction-mode pooler does not support protocol-level named prepared statements, set `DATABASE_PREPARED_STATEMENTS=false`. If your pooler supports them, you can opt in with `DATABASE_PREPARED_STATEMENTS=true`. Check your provider's pooler configuration before you enable this setting. 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_PREPARED_STATEMENTS=false # use false when the pooler does not support named prepared statements DATABASE_POOL_MAX=25 # connection pool size; default: 10 DATABASE_IDLE_TIMEOUT_SECONDS=60 # close idle pooled connections; default: 60 (0 = keep open) DATABASE_CONNECT_TIMEOUT_SECONDS=10 # default: 30 @@ -473,6 +473,14 @@ DATABASE_MAX_LIFETIME_SECONDS=1800 # recycle a pooled connection after this lo DATABASE_APPLICATION_NAME=paperclip # application_name in pg_stat_activity; default: paperclip ``` +An explicit `DATABASE_PREPARED_STATEMENTS=true` enables named prepared statements +for parameterized ORM queries, transactions, reserved connections, and dedicated +clients. Postgres.js manages the statement cache. This does not cache query +results. Bound values and returned data stay fresh. Explicit per-query options +still apply, including `prepare: false` and simple-protocol execution. With the +setting unset, ORM queries keep their existing unprepared behavior and tagged +SQL uses the driver default. With `false`, preparation stays disabled. + ### Bounded response diagnostics `PAPERCLIP_DB_RESPONSE_CAPTURE_WINDOW=/` enables one explicit UTC diff --git a/docs/deploy/database.md b/docs/deploy/database.md index 4f72b70c097..6e85cebce7e 100644 --- a/docs/deploy/database.md +++ b/docs/deploy/database.md @@ -56,12 +56,19 @@ 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 (transaction mode), disable prepared statements via the environment — no source edits needed: +If your transaction-mode pooler does not support protocol-level named prepared statements, disable them via the environment: ```sh DATABASE_PREPARED_STATEMENTS=false ``` +For a direct connection or a pooler that supports named prepared statements, +`DATABASE_PREPARED_STATEMENTS=true` also prepares parameterized ORM queries and +queries inside transactions and reserved or dedicated connections. Check the +provider's configuration before you enable it. Preparation caches statement +metadata, not query results. Unset preserves the existing ORM behavior and the +driver default for tagged SQL. Explicit per-query options still apply. + Related optional client tuning: `DATABASE_POOL_MAX`, `DATABASE_IDLE_TIMEOUT_SECONDS`, `DATABASE_CONNECT_TIMEOUT_SECONDS`, `DATABASE_MAX_LIFETIME_SECONDS`, `DATABASE_APPLICATION_NAME`. Driver defaults apply when unset, except that idle pooled connections close after 60 seconds (`DATABASE_IDLE_TIMEOUT_SECONDS=0` keeps them open) and the pool reports `application_name=paperclip`. See [Connection pool settings](#connection-pool-settings). ## Connection Pool Settings @@ -75,7 +82,7 @@ The server opens one postgres.js pool for its own queries (and a second one when | `DATABASE_CONNECT_TIMEOUT_SECONDS` | `30` (driver) | Give up on a connection attempt after this long. | | `DATABASE_MAX_LIFETIME_SECONDS` | 30–60 min, randomized (driver) | Recycle a pooled connection once it is this old. | | `DATABASE_APPLICATION_NAME` | `paperclip` | Value of `application_name` in `pg_stat_activity`, so you can find Paperclip's backends: `SELECT * FROM pg_stat_activity WHERE application_name = 'paperclip';` | -| `DATABASE_PREPARED_STATEMENTS` | `true` (driver) | Set `false` behind a transaction-mode pooler (see above). | +| `DATABASE_PREPARED_STATEMENTS` | Unset | `true` opts parameterized ORM queries into preparation. `false` disables preparation. Use `false` when the pooler does not support named statements. Unset preserves existing behavior. | The server ends its pools during shutdown (SIGINT/SIGTERM) and when startup fails after the pool was opened, so a restarting server does not leave idle backends behind. Size `max_connections` on the PostgreSQL side for at least `DATABASE_POOL_MAX` per server process plus your other clients. Each server process also opens up to 2 connections (`application_name=paperclip-controller-lease`) for agent-run lease renewals, so they never queue behind slow application queries; count those too. diff --git a/packages/db/src/client.ts b/packages/db/src/client.ts index f1b8079db45..7f067cbfc1c 100644 --- a/packages/db/src/client.ts +++ b/packages/db/src/client.ts @@ -5,6 +5,7 @@ import { fileURLToPath } from "node:url"; import postgres from "postgres"; import { installDatabaseWorkSignals } from "./work-signals.js"; import { installDatabaseResponseCapture } from "./response-capture.js"; +import { installPreparedQueries } from "./prepared-queries.js"; import * as schema from "./schema/index.js"; const MIGRATIONS_FOLDER = fileURLToPath(new URL("./migrations", import.meta.url)); @@ -129,13 +130,6 @@ export type MigrationState = }; 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; @@ -301,6 +295,7 @@ export function createDb(url: string, options?: DatabaseClientOptions) { function createDbClient(url: string, resolved: DatabaseClientOptions, owner?: object) { const sql = postgres(url, postgresJsOptions(resolved)); + if (resolved.prepare === true) installPreparedQueries(sql); const key = hostPortKeyOrNull(url); if (key) registerClient(key, sql); // A disconnect can lose the response after a statement has committed. diff --git a/packages/db/src/prepared-queries-postgres.test.ts b/packages/db/src/prepared-queries-postgres.test.ts new file mode 100644 index 00000000000..edef6dc060e --- /dev/null +++ b/packages/db/src/prepared-queries-postgres.test.ts @@ -0,0 +1,175 @@ +import { eq, sql } from "drizzle-orm"; +import { integer, jsonb, pgTable, text } from "drizzle-orm/pg-core"; +import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it } from "vitest"; +import { closeLivenessDb, createDb, databaseClientOptionsFromEnv, livenessDb, withDedicatedDbConnection, type DatabaseClientOptions, type Db } from "./client.js"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./test-embedded-postgres.js"; +import { runEventPayload } from "./run-event-payload.js"; + +const records = pgTable("prepared_query_fixture", { id: text("id").primaryKey(), companyId: text("company_id").notNull(), value: integer("value").notNull(), payload: jsonb("payload") }); +const encoded = pgTable("prepared_encoded_fixture", { id: text("id").primaryKey(), payload: runEventPayload("payload") }); +const support = await getEmbeddedPostgresTestSupport(); + +describe.skipIf(!support.supported)("explicit ORM prepared statements on PostgreSQL", () => { + let temporary: Awaited>; + let setup: Db; + const clients: Db[] = []; + beforeAll(async () => { + temporary = await startEmbeddedPostgresTestDatabase("prepared-orm-"); + setup = createDb(temporary.connectionString, { prepare: false, maxConnections: 1 }); + await setup.execute(sql`create table prepared_query_fixture(id text primary key, company_id text not null, value integer not null, payload jsonb)`); + await setup.execute(sql`create table prepared_encoded_fixture(id text primary key, payload jsonb)`); + await setup.execute(sql`create table prepared_schema_change_fixture(id text primary key)`); + await setup.execute(sql`insert into prepared_schema_change_fixture(id) values ('retained')`); + }, 90_000); + beforeEach(async () => { + await setup.execute(sql`truncate prepared_query_fixture, prepared_encoded_fixture`); + await setup.insert(records).values([ + { id: "a", companyId: "company-a", value: 1, payload: { owner: "a" } }, + { id: "b", companyId: "company-b", value: 2, payload: { owner: "b" } }, + ]); + }); + afterEach(async () => { + for (const db of clients.splice(0)) { + await closeLivenessDb(db, 2); + await db.$client.end(); + } + }); + afterAll(async () => { await setup?.$client.end(); await temporary?.cleanup(); }); + function fixture(options: DatabaseClientOptions = {}) { + const db = createDb(temporary.connectionString, { ...options, maxConnections: 1 }); + clients.push(db); return db; + } + async function statements(db: Db, pattern = '%"prepared_query_fixture"%') { + return db.$client.unsafe("select name,statement from pg_prepared_statements where statement like $1", [pattern]); + } + + it.each([undefined, false])("keeps the existing ORM mode when prepare is %s", async prepare => { + const db = fixture({ prepare }); + expect(await db.select().from(records).where(eq(records.companyId, "company-a"))).toEqual([ + { id: "a", companyId: "company-a", value: 1, payload: { owner: "a" } }, + ]); + expect(await statements(db)).toHaveLength(0); + }); + + it("honors the explicit environment opt-in for ORM queries with fresh company bindings", async () => { + const db = fixture(databaseClientOptionsFromEnv({ DATABASE_PREPARED_STATEMENTS: "true" })); + for (let i = 0; i < 12; i++) { + const companyId = i % 2 ? "company-b" : "company-a"; + const rows = await db.select().from(records).where(eq(records.companyId, companyId)); + expect(rows).toEqual([{ id: i % 2 ? "b" : "a", companyId, value: i % 2 ? 2 : 1, payload: { owner: i % 2 ? "b" : "a" } }]); + } + expect(await statements(db)).toHaveLength(1); + }); + + it("preserves explicit per-query opt-out", async () => { + const db = fixture({ prepare: true }); + expect((await db.$client.unsafe("select $1::integer as explicit_opt_out_marker", [7], { prepare: false }))[0].explicit_opt_out_marker).toBe(7); + expect(await statements(db, "%explicit_opt_out_marker%")).toHaveLength(0); + }); + + it("preserves simple queries and the two-argument options overload", async () => { + const db = fixture({ prepare: true }); + const simpleOptions = { prepare: true, simple: true }; + const simple = await db.$client.unsafe("select 1 as first; select 2 as second", [], simpleOptions); + expect(simple).toEqual([[{ first: 1 }], [{ second: 2 }]]); + const unsafe = db.$client.unsafe as unknown as (query: string, options: { simple: boolean }) => Promise; + expect(await unsafe("select 3 as third; select 4 as fourth", { simple: true })) + .toEqual([[{ third: 3 }], [{ fourth: 4 }]]); + }); + + it("does not replay a rejected prepared write", async () => { + const db = fixture({ prepare: true }); + await expect(db.insert(records).values({ id: "a", companyId: "company-b", value: 9 })) + .rejects.toMatchObject({ cause: { code: "23505" } }); + expect(await db.select({ companyId: records.companyId, value: records.value }).from(records).where(eq(records.id, "a"))) + .toEqual([{ companyId: "company-a", value: 1 }]); + }); + + it("keeps write queries lazy and repeated awaits from repeating writes", async () => { + const db = fixture({ prepare: true }); + const pending = db.$client.unsafe("insert into prepared_query_fixture(id,company_id,value) values ($1,$2,$3) returning id", ["lazy", "company-a", 3]); + expect(typeof pending.values).toBe("function"); + expect(await setup.select().from(records).where(eq(records.id, "lazy"))).toEqual([]); + expect(await pending).toEqual([{ id: "lazy" }]); + expect(await pending).toEqual([{ id: "lazy" }]); + expect(await setup.select({ id: records.id }).from(records).where(eq(records.id, "lazy"))).toEqual([{ id: "lazy" }]); + expect(await statements(db, "%insert into prepared_query_fixture%")).toHaveLength(1); + }); + + it("preserves transaction commits, nested savepoint rollback, and scoped reads", async () => { + const db = fixture({ prepare: true }); + await db.transaction(async tx => { + expect(await tx.select({ value: records.value }).from(records).where(eq(records.companyId, "company-a"))).toEqual([{ value: 1 }]); + await expect(tx.transaction(async nested => { + await nested.update(records).set({ value: 99 }).where(eq(records.companyId, "company-a")); + throw new Error("rollback savepoint"); + })).rejects.toThrow("rollback savepoint"); + expect(await tx.select({ value: records.value }).from(records).where(eq(records.companyId, "company-a"))).toEqual([{ value: 1 }]); + await tx.insert(records).values({ id: "committed", companyId: "company-b", value: 3 }); + const names = await tx.execute(sql`select statement from pg_prepared_statements`); + expect(names.some(row => String(row.statement).includes('"prepared_query_fixture"'))).toBe(true); + }, { isolationLevel: "serializable" }); + expect(await db.select({ value: records.value }).from(records).where(eq(records.id, "committed"))).toEqual([{ value: 3 }]); + }); + + it("rolls back failed outer transactions without replay", async () => { + const db = fixture({ prepare: true }); + await expect(db.transaction(async tx => { + await tx.update(records).set({ value: 42 }).where(eq(records.id, "a")); + throw new Error("rollback outer transaction"); + })).rejects.toThrow("rollback outer transaction"); + expect(await db.select({ value: records.value }).from(records).where(eq(records.id, "a"))).toEqual([{ value: 1 }]); + }); + + it("prepares reserved-client queries and keeps their bound values fresh", async () => { + const db = fixture({ prepare: true }); + const reserved = await db.$client.reserve(); + try { + for (const marker of [1, 2, 3]) { + expect((await reserved.unsafe("select $1::integer as reserved_prepared_marker", [marker]))[0].reserved_prepared_marker).toBe(marker); + } + const names = await reserved.unsafe("select statement from pg_prepared_statements where statement like $1", ["%reserved_prepared_marker%"]); + expect(names).toHaveLength(1); + } finally { reserved.release(); } + }); + + it("preserves object and array results for the same prepared shape", async () => { + const db = fixture({ prepare: true }); + expect(await db.$client.unsafe("select $1::integer as result_mode_marker", [41]).values()).toEqual([[41]]); + expect(await db.$client.unsafe("select $1::integer as result_mode_marker", [42])).toEqual([{ result_mode_marker: 42 }]); + }); + + it("keeps custom stored-payload decoding fresh", async () => { + const db = fixture({ prepare: true }); + const first = { text: "encoded first payload🙂".repeat(2048) }; + const second = { text: "encoded second payload🙂".repeat(2048) }; + await setup.insert(encoded).values({ id: "encoded", payload: first }); + expect((await db.select().from(encoded).where(eq(encoded.id, "encoded")))[0].payload).toEqual(first); + await setup.update(encoded).set({ payload: second }).where(eq(encoded.id, "encoded")); + expect((await db.select().from(encoded).where(eq(encoded.id, "encoded")))[0].payload).toEqual(second); + expect(await statements(db, '%"prepared_encoded_fixture"%')).toHaveLength(1); + }); + + it("inherits the opt-in in dedicated and liveness clients", async () => { + const db = fixture({ prepare: true }); + await withDedicatedDbConnection(db, async dedicated => { + expect(await dedicated.select({ value: records.value }).from(records).where(eq(records.companyId, "company-b"))).toEqual([{ value: 2 }]); + expect(await statements(dedicated)).toHaveLength(1); + }); + const live = livenessDb(db); + expect(await live.select({ value: records.value }).from(records).where(eq(records.companyId, "company-a"))).toEqual([{ value: 1 }]); + expect(await statements(live)).toHaveLength(1); + }); + + it("refreshes prepared result metadata after a schema change", async () => { + const db = fixture({ prepare: true }); + const query = "select * from prepared_schema_change_fixture where id=$1"; + expect(await db.$client.unsafe(query, ["retained"])).toEqual([{ id: "retained" }]); + try { + await setup.execute(sql`alter table prepared_schema_change_fixture add column extra text`); + expect(await db.$client.unsafe(query, ["retained"])).toEqual([{ id: "retained", extra: null }]); + } finally { + await setup.execute(sql`alter table prepared_schema_change_fixture drop column if exists extra`); + } + }); +}); diff --git a/packages/db/src/prepared-queries.ts b/packages/db/src/prepared-queries.ts new file mode 100644 index 00000000000..7622722027a --- /dev/null +++ b/packages/db/src/prepared-queries.ts @@ -0,0 +1,47 @@ +type Method = (this: unknown, ...args: unknown[]) => unknown; + +export function installPreparedQueries(client: unknown): void { + const visited = new WeakSet(); + + function visit(value: unknown): void { + if ((typeof value !== "object" || value === null) && typeof value !== "function") return; + if (visited.has(value)) return; + visited.add(value); + const target = value as Record; + const unsafe = target.unsafe; + if (typeof unsafe === "function") { + target.unsafe = function (this: unknown, ...args: unknown[]) { + const index = args.length === 2 && !Array.isArray(args[1]) ? 1 : 2; + const options = args[index]; + if (options !== undefined && (options === null || typeof options !== "object")) { + return Reflect.apply(unsafe, this, args); + } + if (index === 2 && args.length < 2) args[1] = []; + args[index] = { prepare: true, ...options }; + return Reflect.apply(unsafe, this, args); + }; + } + for (const method of ["begin", "savepoint"]) { + const original = target[method]; + if (typeof original !== "function") continue; + target[method] = function (this: unknown, ...args: unknown[]) { + const wrapped = args.map(arg => typeof arg !== "function" ? arg : function (this: unknown, ...inner: unknown[]) { + visit(inner[0]); + return Reflect.apply(arg, this, inner); + }); + return Reflect.apply(original, this, wrapped); + } satisfies Method; + } + const reserve = target.reserve; + if (typeof reserve === "function") { + target.reserve = function (this: unknown, ...args: unknown[]) { + return Promise.resolve(Reflect.apply(reserve, this, args)).then(reserved => { + visit(reserved); + return reserved; + }); + } satisfies Method; + } + } + + visit(client); +} diff --git a/packages/db/src/response-capture-postgres.test.ts b/packages/db/src/response-capture-postgres.test.ts index 96de91ff913..42d5d805a0c 100644 --- a/packages/db/src/response-capture-postgres.test.ts +++ b/packages/db/src/response-capture-postgres.test.ts @@ -31,11 +31,11 @@ describe.skipIf(!support.supported)("database response capture PostgreSQL behavi vi.restoreAllMocks(); vi.unstubAllEnvs(); }); afterAll(async () => { await temporary?.cleanup(); }); - function fixture() { + function fixture(prepare?: boolean) { const events: Array> = []; const now = Date.now(); const capture = createDatabaseResponseCapture({ start: now - 100, end: now + 59_000 }, event => events.push(event)); - const db = createDb(temporary.connectionString); + const db = createDb(temporary.connectionString, { prepare }); clients.push(db); captures.push(capture); expect(installDatabaseResponseCapture(db, { table, encodedTable }, capture)).toBe(db); expect(installDatabaseResponseCapture(db, { table, encodedTable }, capture)).toBe(db); @@ -43,6 +43,18 @@ describe.skipIf(!support.supported)("database response capture PostgreSQL behavi } function summary(events: Array>) { return events.find(event => event.event === "paperclip.db_response_capture.summary.v1")!; } + it("retains capture counts when ORM preparation is enabled", async () => { + const f = fixture(true), id = randomUUID(), payload = "prepared-capture-value"; + await f.db.insert(table).values({ id, payload }); + await f.db.transaction(async tx => { + expect(await tx.select().from(table).where(eq(table.id, id))).toEqual([{ id, payload }]); + }); + f.capture.flush(); + expect(summary(f.events)).toMatchObject({ startedQueries: 2, completedQueries: 2 }); + expect(JSON.stringify(f.events)).not.toContain(payload); + expect(JSON.stringify(f.events)).not.toContain(id); + }); + it("keeps rows and builder identity through lazy select and write returning", async () => { const f = fixture(), id = randomUUID(), payload = "sentinel-result-and-parameter-value🙂\n\""; const insert = f.db.insert(table).values({ id, payload }); @@ -173,11 +185,11 @@ describe.skipIf(!support.supported)("database response capture PostgreSQL behavi f.capture.flush(); expect(summary(f.events).completedQueries).toBeGreaterThan(4); }); - it("covers automatically created dedicated pools and retains owner identity", async () => { + it.each([undefined, true])("covers automatically created dedicated pools with prepare %s and retains owner identity", async prepare => { const logs: string[] = [], now = Date.now(); vi.spyOn(console, "info").mockImplementation(value => logs.push(String(value))); vi.stubEnv("PAPERCLIP_DB_RESPONSE_CAPTURE_WINDOW", `${new Date(now - 100).toISOString()}/${new Date(now + 59_000).toISOString()}`); - const db = createDb(temporary.connectionString); clients.push(db); + const db = createDb(temporary.connectionString, { prepare }); clients.push(db); const capture = databaseResponseCaptureFromEnv()!; captures.push(capture); expect(await withDedicatedDbConnection(db, async dedicated => dedicated.execute(sql`select ${"dedicated-value"}::text as value`))) .toEqual([{ value: "dedicated-value" }]);