fix: harden PostgreSQL failure boundaries

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",

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<string, string>[] {
+  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);

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, unknown>[]): 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<T extends AnyTable> = {
   [K in keyof T["$columns"]]: K extends keyof InferKey<T> ? K : never;
 }[keyof T["$columns"]];
@@ -262,8 +283,9 @@ export class TableClient<T extends AnyTable> {
   }
 
   /**
-   * 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<T>[],
@@ -280,14 +302,17 @@ export class TableClient<T extends AnyTable> {
     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<string, unknown>[]);
+    const effectiveChunkSize = effectiveBatchChunkSize(
+      chunkSize,
+      allProperties.length,
+      "insertMany",
+    );
     let rowsAffected = 0;
     const rows: InferRow<T>[] = [];
-    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<string, unknown>))),
-      ];
-      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<string, unknown>[]);
       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<T extends AnyTable> {
 
   /**
    * 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<T extends AnyTable> {
     const returned: InferRow<T>[] = [];
     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<string, unknown>))),
-      ];
+    const allProperties = batchProperties(inputs as readonly Record<string, unknown>[]);
+    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<string, unknown>[]);
       const columns = properties.map((property) => {
         const column = this.definition.$columns[property];
         if (!column) throw new Error(`Unknown ${this.definition.$name} property ${property}.`);

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<string, string>[] {
+  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" });

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);
+  });
 });