diff --git a/package-lock.json b/package-lock.json index db16b8d..a6967c0 100644 --- a/package-lock.json +++ b/package-lock.json @@ -1,12 +1,12 @@ { "name": "@askrjs/orm", - "version": "0.0.1", + "version": "0.0.2", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "@askrjs/orm", - "version": "0.0.1", + "version": "0.0.2", "license": "Apache-2.0", "dependencies": { "tsx": "^4.23.11" diff --git a/package.json b/package.json index 846bac5..304bc88 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "@askrjs/orm", - "version": "0.0.1", + "version": "0.0.2", "description": "Postgres-first, SQL-shaped micro-ORM for Askr", "keywords": [ "askr", diff --git a/src/client.test.ts b/src/client.test.ts index 194b35e..5165366 100644 --- a/src/client.test.ts +++ b/src/client.test.ts @@ -35,6 +35,22 @@ const users = table("users", { groupId: uuid().notNull(), }); +const wideColumns = Object.fromEntries( + Array.from({ length: 70 }, (_, index) => [ + `value${index}`, + index === 0 ? text().primaryKey() : text().notNull(), + ]), +); +const wide = table("wide", wideColumns); + +function wideRows(prefix: string): Record[] { + return Array.from({ length: 1000 }, (_, row) => + Object.fromEntries( + Array.from({ length: 70 }, (_, column) => [`value${column}`, `${prefix}-${row}-${column}`]), + ), + ); +} + describe("database client", () => { it("should use status-first CRUD and explicit returning", async () => { const adapter = new RecordingAdapter(); @@ -69,6 +85,20 @@ describe("database client", () => { expect(adapter.transactions).toBe(1); }); + it("should keep insertMany and upsertMany chunks within the PostgreSQL parameter limit", async () => { + const adapter = new RecordingAdapter(); + const db = createDatabaseClient({ wide }, adapter); + + await db.wide.insertMany(wideRows("insert")); + await db.wide.upsertMany(wideRows("upsert")); + + expect(adapter.queries).toHaveLength(4); + expect(adapter.queries.every((query) => query.values.length <= 65_535)).toBe(true); + expect(adapter.queries.map((query) => query.values.length)).toEqual([ + 65_520, 4480, 65_520, 4480, + ]); + }); + it("should require explicit join projection and compile null-safe left joins", () => { const adapter = new RecordingAdapter(); const db = createDatabaseClient({ users, groups }, adapter); diff --git a/src/client.ts b/src/client.ts index a821e99..b0f15e8 100644 --- a/src/client.ts +++ b/src/client.ts @@ -34,6 +34,27 @@ export interface ReturningStatus { readonly returning?: "status"; } +const POSTGRES_PARAMETER_LIMIT = 65_535; + +function batchProperties(inputs: readonly Record[]): string[] { + return [...new Set(inputs.flatMap((input) => Object.keys(input)))]; +} + +function effectiveBatchChunkSize( + configuredChunkSize: number, + propertyCount: number, + operation: "insertMany" | "upsertMany", +): number { + if (propertyCount === 0) throw new Error(`${operation} rows require at least one value.`); + const parameterBound = Math.floor(POSTGRES_PARAMETER_LIMIT / propertyCount); + if (parameterBound < 1) { + throw new Error( + `${operation} rows contain ${propertyCount} values, exceeding PostgreSQL's ${POSTGRES_PARAMETER_LIMIT}-parameter statement limit.`, + ); + } + return Math.min(configuredChunkSize, parameterBound); +} + type PrimaryKeyName = { [K in keyof T["$columns"]]: K extends keyof InferKey ? K : never; }[keyof T["$columns"]]; @@ -262,8 +283,9 @@ export class TableClient { } /** - * Inserts many rows in chunks (default 1000 per statement, via `options.chunkSize`). - * Pass `{ returning: "rows" }` to get all inserted rows back. + * Inserts many rows in chunks (default 1000 per statement, via `options.chunkSize`), capped by + * row width so no statement exceeds PostgreSQL's 65,535 bind-parameter limit. Pass + * `{ returning: "rows" }` to get all inserted rows back. */ async insertMany( inputs: readonly InferInsert[], @@ -280,14 +302,17 @@ export class TableClient { if (inputs.length === 0) return options.returning === "rows" ? [] : { rowsAffected: 0 }; const chunkSize = options.chunkSize ?? 1000; if (!Number.isSafeInteger(chunkSize) || chunkSize < 1) throw new Error("Invalid chunkSize."); + const allProperties = batchProperties(inputs as readonly Record[]); + const effectiveChunkSize = effectiveBatchChunkSize( + chunkSize, + allProperties.length, + "insertMany", + ); let rowsAffected = 0; const rows: InferRow[] = []; - for (let index = 0; index < inputs.length; index += chunkSize) { - const chunk = inputs.slice(index, index + chunkSize); - const properties = [ - ...new Set(chunk.flatMap((input) => Object.keys(input as Record))), - ]; - if (properties.length === 0) throw new Error("insertMany rows require at least one value."); + for (let index = 0; index < inputs.length; index += effectiveChunkSize) { + const chunk = inputs.slice(index, index + effectiveChunkSize); + const properties = batchProperties(chunk as readonly Record[]); const columns = properties.map((property) => { const column = this.definition.$columns[property]; if (!column) throw new Error(`Unknown ${this.definition.$name} property ${property}.`); @@ -396,7 +421,8 @@ export class TableClient { /** * Inserts many rows, updating non-primary-key columns on conflict (`ON CONFLICT ... DO UPDATE`), - * in chunks (default 1000 per statement, via `options.chunkSize`). Requires the table to have a + * in chunks (default 1000 per statement, via `options.chunkSize`) capped by row width so no + * statement exceeds PostgreSQL's 65,535 bind-parameter limit. Requires the table to have a * primary key. */ async upsertMany( @@ -422,11 +448,15 @@ export class TableClient { const returned: InferRow[] = []; const chunkSize = options.chunkSize ?? 1000; if (!Number.isSafeInteger(chunkSize) || chunkSize < 1) throw new Error("Invalid chunkSize."); - for (let index = 0; index < inputs.length; index += chunkSize) { - const chunk = inputs.slice(index, index + chunkSize); - const properties = [ - ...new Set(chunk.flatMap((input) => Object.keys(input as Record))), - ]; + const allProperties = batchProperties(inputs as readonly Record[]); + const effectiveChunkSize = effectiveBatchChunkSize( + chunkSize, + allProperties.length, + "upsertMany", + ); + for (let index = 0; index < inputs.length; index += effectiveChunkSize) { + const chunk = inputs.slice(index, index + effectiveChunkSize); + const properties = batchProperties(chunk as readonly Record[]); const columns = properties.map((property) => { const column = this.definition.$columns[property]; if (!column) throw new Error(`Unknown ${this.definition.$name} property ${property}.`); diff --git a/src/integration.test.ts b/src/integration.test.ts index cbea663..d8037a9 100644 --- a/src/integration.test.ts +++ b/src/integration.test.ts @@ -1,4 +1,7 @@ import { randomUUID } from "node:crypto"; +import { execFile } from "node:child_process"; +import path from "node:path"; +import { promisify } from "node:util"; import { Pool } from "pg"; import { afterAll, beforeAll, describe, expect, it } from "vitest"; import type { DatabaseAdapter } from "./adapter"; @@ -8,6 +11,7 @@ import { createMigrationsApi, type MigrationManifest } from "./migrations"; const databaseUrl = process.env.ASKR_ORM_TEST_DATABASE_URL; const integration = databaseUrl ? describe : describe.skip; +const execFileAsync = promisify(execFile); const groups = table("orm_groups", { id: uuid().primaryKey(), @@ -21,6 +25,21 @@ const users = table("orm_users", { .references(() => groups.id), createdAt: timestampTz().notNull().defaultNow(), }); +const wideColumns = Object.fromEntries( + Array.from({ length: 70 }, (_, index) => [ + `value${index}`, + index === 0 ? text().primaryKey() : text().notNull(), + ]), +); +const wide = table("orm_wide", wideColumns); + +function createWideRows(prefix: string): Record[] { + return Array.from({ length: 1000 }, (_, row) => + Object.fromEntries( + Array.from({ length: 70 }, (_, column) => [`value${column}`, `${prefix}-${row}-${column}`]), + ), + ); +} integration("PostgreSQL adapter conformance", () => { const pool = new Pool({ connectionString: databaseUrl, max: 4 }); @@ -36,11 +55,15 @@ integration("PostgreSQL adapter conformance", () => { await pool.query( 'CREATE TABLE "orm_users" ("id" uuid PRIMARY KEY DEFAULT gen_random_uuid(), "email" text NOT NULL UNIQUE, "group_id" uuid NOT NULL REFERENCES "orm_groups" ("id"), "created_at" timestamptz NOT NULL DEFAULT CURRENT_TIMESTAMP)', ); + await pool.query('DROP TABLE IF EXISTS "orm_wide"'); + await pool.query( + `CREATE TABLE "orm_wide" (${Array.from({ length: 70 }, (_, index) => `"value${index}" text ${index === 0 ? "PRIMARY KEY" : "NOT NULL"}`).join(", ")})`, + ); }); afterAll(async () => { await pool.query( - 'DROP TABLE IF EXISTS "_askr_migrations", "orm_migration_probe", "orm_users", "orm_groups" CASCADE', + 'DROP TABLE IF EXISTS "_askr_migrations", "orm_migration_probe", "orm_wide", "orm_users", "orm_groups" CASCADE', ); await pool.end(); await adapter.close?.(); @@ -98,6 +121,25 @@ integration("PostgreSQL adapter conformance", () => { ).toBeNull(); }); + it("should reject a checked-out connection terminated while idle in a transaction", async () => { + const fixture = path.resolve("tests/fixtures/terminated-postgres-client.ts"); + const result = await execFileAsync(process.execPath, ["--import", "tsx", fixture], { + env: { ...process.env, ASKR_ORM_TEST_DATABASE_URL: databaseUrl! }, + }); + expect(result.stderr).toBe(""); + expect(result.stdout).toMatch(/^caught:connection:57P01\n$/); + }); + + it("should insert and upsert real wide-table batches without overflowing bind parameters", async () => { + const wideDb = createDatabaseClient({ wide }, adapter); + await expect(wideDb.wide.insertMany(createWideRows("insert"))).resolves.toEqual({ + rowsAffected: 1000, + }); + await expect(wideDb.wide.upsertMany(createWideRows("upsert"))).resolves.toEqual({ + rowsAffected: 1000, + }); + }); + it("should isolate nested savepoints and cancel cursor streams", async () => { const groupId = randomUUID(); await db.groups.insert({ id: groupId, name: "Savepoints" }); diff --git a/src/postgres.test.ts b/src/postgres.test.ts index 1dbb13b..bb45905 100644 --- a/src/postgres.test.ts +++ b/src/postgres.test.ts @@ -4,6 +4,7 @@ const state = vi.hoisted(() => ({ statements: [] as string[], ends: 0, releases: 0, + clientErrorListeners: new Set<(error: Error) => void>(), })); vi.mock("pg", () => { @@ -16,6 +17,12 @@ vi.mock("pg", () => { async connect() { return { query: (config: string | { text?: string }) => this.query(config), + on: (event: string, listener: (error: Error) => void) => { + if (event === "error") state.clientErrorListeners.add(listener); + }, + off: (event: string, listener: (error: Error) => void) => { + if (event === "error") state.clientErrorListeners.delete(listener); + }, release: () => { state.releases += 1; }, @@ -43,6 +50,7 @@ describe("PostgreSQL adapter", () => { state.statements.length = 0; state.ends = 0; state.releases = 0; + state.clientErrorListeners.clear(); }); it("should deallocate described statements and support repeated description", async () => { @@ -67,4 +75,21 @@ describe("PostgreSQL adapter", () => { await adapter.close?.(); expect(state.ends).toBe(1); }); + + it("should normalize a checked-out client error and remove its listener on release", async () => { + const adapter = await postgres({ + url: "postgres://target", + shadowUrl: "postgres://shadow", + }).open(); + + await expect( + adapter.transaction(async (transaction) => { + const error = Object.assign(new Error("connection terminated"), { code: "57P01" }); + for (const listener of state.clientErrorListeners) listener(error); + await transaction.execute({ text: "SELECT 1", values: [] }); + }), + ).rejects.toMatchObject({ category: "connection", code: "57P01" }); + expect(state.clientErrorListeners).toHaveLength(0); + expect(state.releases).toBe(1); + }); }); diff --git a/src/postgres.ts b/src/postgres.ts index 17ba64e..377bfbe 100644 --- a/src/postgres.ts +++ b/src/postgres.ts @@ -8,6 +8,7 @@ import type { } from "./adapter"; import type { DatabaseToolingAdapter } from "./definition"; import type { SqlQuery } from "./sql"; +import { normalizeDatabaseError, type DatabaseError } from "./errors"; export { jsonb, postgresEnum, postgresType, timestampTz, bytea } from "./schema"; const MIGRATION_LOCK_KEY = "4707438161740729"; @@ -25,6 +26,43 @@ function lazy(value: string | (() => string) | undefined, environment: string): return result; } +class CheckedOutClient { + private failure: DatabaseError | undefined; + + private readonly onError = (error: Error): void => { + this.failure ??= normalizeDatabaseError(error); + }; + + constructor(private readonly client: PoolClient) { + client.on("error", this.onError); + } + + assertHealthy(): void { + if (this.failure) throw this.failure; + } + + async query(config: unknown): Promise { + this.assertHealthy(); + try { + const result = await this.client.query(config as never); + this.assertHealthy(); + return result; + } catch (error) { + throw normalizeDatabaseError(error); + } + } + + startStream(query: T): T { + this.assertHealthy(); + return this.client.query(query as never) as T; + } + + release(): void { + this.client.off("error", this.onError); + this.client.release(this.failure); + } +} + class PgAdapter implements DatabaseAdapter { readonly identity: string; private closed = false; @@ -32,7 +70,7 @@ class PgAdapter implements DatabaseAdapter { private readonly executor: { query(config: unknown): Promise }, identity: string, private readonly pool?: PoolType, - private readonly client?: PoolClient, + private readonly client?: CheckedOutClient, private readonly transactionDepth = 0, ) { this.identity = identity; @@ -52,19 +90,21 @@ class PgAdapter implements DatabaseAdapter { async *stream(query: SqlQuery, options: QueryOptions = {}): AsyncIterable { if (options.signal?.aborted) throw options.signal.reason; - const owned = this.client ? undefined : await this.pool?.connect(); + if (!this.client && !this.pool) throw new Error("PostgreSQL streaming requires a pool."); + const owned = this.client ? undefined : new CheckedOutClient(await this.pool!.connect()); const client = this.client ?? owned; if (!client) throw new Error("PostgreSQL streaming requires a pinned client."); try { if (options.signal?.aborted) throw options.signal.reason; const { default: QueryStream } = await import("pg-query-stream"); if (options.signal?.aborted) throw options.signal.reason; - const stream = client.query(new QueryStream(query.text, [...query.values])); + const stream = client.startStream(new QueryStream(query.text, [...query.values])); const abort = () => stream.destroy(options.signal?.reason); options.signal?.addEventListener("abort", abort, { once: true }); if (options.signal?.aborted) abort(); try { for await (const row of stream) yield row as Row; + client.assertHealthy(); } finally { options.signal?.removeEventListener("abort", abort); stream.destroy(); @@ -84,7 +124,7 @@ class PgAdapter implements DatabaseAdapter { try { const value = await callback( new PgAdapter( - this.client as never, + this.client, this.identity, undefined, this.client, @@ -94,13 +134,13 @@ class PgAdapter implements DatabaseAdapter { await this.client.query(`RELEASE SAVEPOINT ${savepoint}`); return value; } catch (error) { - await this.client.query(`ROLLBACK TO SAVEPOINT ${savepoint}`); - await this.client.query(`RELEASE SAVEPOINT ${savepoint}`); + await this.client.query(`ROLLBACK TO SAVEPOINT ${savepoint}`).catch(() => undefined); + await this.client.query(`RELEASE SAVEPOINT ${savepoint}`).catch(() => undefined); throw error; } } if (!this.client && !this.pool) throw new Error("PostgreSQL transactions require a pool."); - const owned = this.client ? undefined : await this.pool!.connect(); + const owned = this.client ? undefined : new CheckedOutClient(await this.pool!.connect()); const client = this.client ?? owned!; try { const clauses = [ @@ -110,13 +150,11 @@ class PgAdapter implements DatabaseAdapter { .filter(Boolean) .join(" "); await client.query(`BEGIN${clauses ? ` ${clauses}` : ""}`); - const value = await callback( - new PgAdapter(client as never, this.identity, undefined, client, 1), - ); + const value = await callback(new PgAdapter(client, this.identity, undefined, client, 1)); await client.query("COMMIT"); return value; } catch (error) { - await client.query("ROLLBACK"); + await client.query("ROLLBACK").catch(() => undefined); throw error; } finally { owned?.release(); @@ -126,9 +164,11 @@ class PgAdapter implements DatabaseAdapter { async session(callback: (adapter: DatabaseAdapter) => Promise): Promise { if (this.client) return callback(this); if (!this.pool) throw new Error("PostgreSQL sessions require a pool."); - const client = await this.pool.connect(); + const client = new CheckedOutClient(await this.pool.connect()); try { - return await callback(new PgAdapter(client as never, this.identity, undefined, client)); + const value = await callback(new PgAdapter(client, this.identity, undefined, client)); + client.assertHealthy(); + return value; } finally { client.release(); } @@ -194,7 +234,7 @@ async function pgTooling( .rows; }, async describe(sql, parameterNames) { - const client = await pool.connect(); + const client = new CheckedOutClient(await pool.connect()); let prepared = false; try { await client.query(`PREPARE askr_describe AS ${sql}`); diff --git a/tests/fixtures/terminated-postgres-client.ts b/tests/fixtures/terminated-postgres-client.ts new file mode 100644 index 0000000..a643167 --- /dev/null +++ b/tests/fixtures/terminated-postgres-client.ts @@ -0,0 +1,28 @@ +import { Pool } from "pg"; +import { DatabaseError } from "../../src/errors"; +import { postgres } from "../../src/postgres"; + +const databaseUrl = process.env.ASKR_ORM_TEST_DATABASE_URL; +if (!databaseUrl) throw new Error("ASKR_ORM_TEST_DATABASE_URL is required."); + +const adapter = await postgres({ url: databaseUrl }).open(); +const killer = new Pool({ connectionString: databaseUrl }); + +try { + await adapter.transaction(async (transaction) => { + const result = await transaction.execute<{ pid: number }>({ + text: "SELECT pg_backend_pid() AS pid", + values: [], + }); + await killer.query("SELECT pg_terminate_backend($1)", [result.rows[0]!.pid]); + await new Promise((resolve) => setTimeout(resolve, 200)); + await transaction.execute({ text: "SELECT 1", values: [] }); + }); + throw new Error("Terminated transaction unexpectedly succeeded."); +} catch (error) { + if (!(error instanceof DatabaseError) || error.category !== "connection") throw error; + process.stdout.write(`caught:${error.category}:${error.code ?? "unknown"}\n`); +} finally { + await killer.end(); + await adapter.close?.(); +}