Sanitize reused PostgreSQL connections at checkout

postgres/pg-cache/README.md

@@ -23,6 +23,7 @@ npm install pg-cache
 ## Features
 
 - LRU cache for PostgreSQL connection pools
+- Checkout sanitation for reused node-postgres clients
 - Automatic pool cleanup and disposal
 - Extensible cleanup callback system
 - Service cache for general use
@@ -126,6 +127,11 @@ The main PostgreSQL pool cache instance.
 ### getPgPool(config: Partial<PgConfig>): Pool
 
 Get or create a cached PostgreSQL pool using the provided configuration.
+Clients from the default node-postgres factory run `DISCARD ALL` before every
+checkout, and stale client-side prepared-statement bookkeeping is cleared to
+match the server. If sanitation fails, the client is destroyed and the checkout
+fails. Alternate registered pool factories retain ownership of backend-specific
+checkout sanitation.
 
 ### svcCache
 
@@ -138,3 +144,7 @@ Gracefully close all cached pools and wait for disposal.
 ## Integration with Other Packages
 
 This package is designed to be extended. For example, `graphile-cache` uses the cleanup callback system to automatically clean up PostGraphile instances when their associated pools are disposed.
+
+### Checkout sanitation performance
+
+The default sanitizer adds a database round trip and invalidates prepared statements on every checkout. See the [reproducible benchmark and measured tradeoff](../pg-query-context/benchmarks/README.md) before setting a production throughput budget. The benchmark does not weaken the default sanitation contract.

postgres/pg-cache/src/__tests__/driver.test.ts

@@ -47,13 +47,15 @@ describe('pg-cache pool-factory seam', () => {
   it('getPgPool builds via the registered factory (no real pg connection)', () => {
     const cfg = freshConfig();
     const mock = createMockPool();
+    const alternateConnect = mock.connect;
     const factory = jest.fn<pg.Pool, [any]>(() => mock);
     registerPgPoolFactory(factory);
 
     const pool = getPgPool(cfg);
 
     expect(factory).toHaveBeenCalledTimes(1);
     expect(pool).toBe(mock);
+    expect(pool.connect).toBe(alternateConnect);
 
     pgCache.delete(cfg.database);
   });

postgres/pg-cache/src/__tests__/sanitizer.test.ts

@@ -0,0 +1,129 @@
+import type { Pool, PoolClient } from 'pg';
+
+import {
+  installCheckoutSanitizer,
+  sanitizePgClient,
+} from '../sanitizer';
+
+type PreparedConnection = {
+  parsedStatements: Record<string, string>;
+  _graphilePreparedStatementCache?: { dispose: jest.Mock };
+};
+
+const createClient = (
+  connection: PreparedConnection,
+  query = jest.fn().mockResolvedValue({ rows: [] })
+): PoolClient =>
+  ({
+    connection,
+    query,
+    release: jest.fn(),
+  }) as unknown as PoolClient;
+
+const createPool = (connect: jest.Mock): Pool =>
+  ({
+    connect,
+  }) as unknown as Pool;
+
+describe('PostgreSQL checkout sanitation', () => {
+  it('discards session state and clears client-side prepared-statement bookkeeping', async () => {
+    const graphileCache = { dispose: jest.fn() };
+    const connection: PreparedConnection = {
+      parsedStatements: {
+        tenantLookup: 'select 1',
+        tenantMutation: 'select 2',
+      },
+      _graphilePreparedStatementCache: graphileCache,
+    };
+    const client = createClient(connection);
+
+    await expect(sanitizePgClient(client)).resolves.toBe(client);
+
+    expect(client.query).toHaveBeenCalledWith('DISCARD ALL');
+    expect(connection.parsedStatements).toEqual({});
+    expect(connection).not.toHaveProperty('_graphilePreparedStatementCache');
+    expect(graphileCache.dispose).not.toHaveBeenCalled();
+    expect(client.release).not.toHaveBeenCalled();
+  });
+
+  it('destroys a client and preserves the original sanitation error', async () => {
+    const error = new Error('DISCARD ALL failed');
+    const connection: PreparedConnection = {
+      parsedStatements: { tenantLookup: 'select 1' },
+    };
+    const client = createClient(
+      connection,
+      jest.fn().mockRejectedValue(error)
+    );
+
+    await expect(sanitizePgClient(client)).rejects.toBe(error);
+
+    expect(client.release).toHaveBeenCalledWith(true);
+    expect(connection.parsedStatements).toEqual({
+      tenantLookup: 'select 1',
+    });
+  });
+
+  it('sanitizes every promise-based checkout and installs only once', async () => {
+    const connection: PreparedConnection = { parsedStatements: {} };
+    const client = createClient(connection);
+    const connect = jest.fn().mockResolvedValue(client);
+    const pool = createPool(connect);
+
+    expect(installCheckoutSanitizer(pool)).toBe(pool);
+    expect(installCheckoutSanitizer(pool)).toBe(pool);
+
+    await expect(pool.connect()).resolves.toBe(client);
+    await expect(pool.connect()).resolves.toBe(client);
+
+    expect(connect).toHaveBeenCalledTimes(2);
+    expect(client.query).toHaveBeenNthCalledWith(1, 'DISCARD ALL');
+    expect(client.query).toHaveBeenNthCalledWith(2, 'DISCARD ALL');
+  });
+
+  it('preserves node-postgres callback checkout semantics', async () => {
+    const connection: PreparedConnection = { parsedStatements: {} };
+    const client = createClient(connection);
+    const pool = createPool(jest.fn().mockResolvedValue(client));
+    installCheckoutSanitizer(pool);
+
+    await new Promise<void>((resolve, reject) => {
+      pool.connect((error, checkedOutClient, done) => {
+        try {
+          expect(error).toBeUndefined();
+          expect(checkedOutClient).toBe(client);
+          expect(done).toEqual(expect.any(Function));
+          done();
+          expect(client.release).toHaveBeenCalledWith();
+          resolve();
+        } catch (assertionError) {
+          reject(assertionError);
+        }
+      });
+    });
+  });
+
+  it('reports callback checkout failures only after destroying the client', async () => {
+    const error = new Error('cannot sanitize client');
+    const connection: PreparedConnection = { parsedStatements: {} };
+    const client = createClient(
+      connection,
+      jest.fn().mockRejectedValue(error)
+    );
+    const pool = createPool(jest.fn().mockResolvedValue(client));
+    installCheckoutSanitizer(pool);
+
+    await new Promise<void>((resolve, reject) => {
+      pool.connect((checkoutError, checkedOutClient) => {
+        try {
+          expect(checkoutError).toBe(error);
+          expect(checkedOutClient).toBeUndefined();
+          expect(client.release).toHaveBeenCalledWith(true);
+          resolve();
+        } catch (assertionError) {
+          reject(assertionError);
+        }
+      });
+    });
+  });
+});

postgres/pg-cache/src/pg.ts

@@ -5,6 +5,7 @@ import { getPgEnvOptions, PgConfig, PgPoolConfig } from 'pg-env';
 
 import { getActivePgPoolFactory, PgPoolFactory } from './driver';
 import { pgCache } from './lru';
+import { installCheckoutSanitizer } from './sanitizer';
 
 const log = new Logger('pg-cache');
 
@@ -97,7 +98,7 @@ export const defaultPgPoolFactory: PgPoolFactory = (pgConfig): pg.Pool => {
     }
   });
 
-  return pgPool;
+  return installCheckoutSanitizer(pgPool);
 };
 
 export const getPgPool = (pgConfig: Partial<PgConfig> & { pool?: PgPoolConfig }): pg.Pool => {

postgres/pg-cache/src/sanitizer.ts

@@ -0,0 +1,79 @@
+import type pg from 'pg';
+
+type PgConnectionWithPreparedState = {
+  parsedStatements?: Record<string, string>;
+  _graphilePreparedStatementCache?: unknown;
+};
+
+type PgClientWithPreparedState = pg.PoolClient & {
+  connection?: PgConnectionWithPreparedState;
+};
+
+const checkoutSanitizedPools = new WeakSet<pg.Pool>();
+
+/**
+ * Forget prepared statements that PostgreSQL removed during `DISCARD ALL`.
+ *
+ * Graphile's cache is deliberately deleted rather than disposed: its disposer
+ * issues asynchronous `DEALLOCATE` queries, which would duplicate `DISCARD ALL`
+ * and could race with the next owner of the checked-out client.
+ */
+export function clearPreparedStatementBookkeeping(client: pg.PoolClient): void {
+  const connection = (client as PgClientWithPreparedState).connection;
+  if (!connection) return;
+
+  if (connection.parsedStatements) {
+    for (const statementName of Object.keys(connection.parsedStatements)) {
+      delete connection.parsedStatements[statementName];
+    }
+  }
+
+  delete connection._graphilePreparedStatementCache;
+}
+
+/**
+ * Restore a checked-out PostgreSQL client to server defaults before reuse.
+ * A client that cannot be sanitized is destroyed instead of being returned to
+ * application code with unknown session state.
+ */
+export async function sanitizePgClient(client: pg.PoolClient): Promise<pg.PoolClient> {
+  try {
+    await client.query('DISCARD ALL');
+    clearPreparedStatementBookkeeping(client);
+    return client;
+  } catch (error) {
+    client.release(true);
+    throw error;
+  }
+}
+
+/**
+ * Sanitize every client obtained from a node-postgres pool. `pool.query()` also
+ * goes through `connect()`, so both checkout APIs share the same boundary.
+ */
+export function installCheckoutSanitizer(pool: pg.Pool): pg.Pool {
+  if (checkoutSanitizedPools.has(pool)) return pool;
+
+  const connect = pool.connect.bind(pool);
+  const sanitizedConnect = async (): Promise<pg.PoolClient> => {
+    const client = await connect();
+    return sanitizePgClient(client);
+  };
+
+  pool.connect = ((callback?: (
+    err: Error | undefined,
+    client: pg.PoolClient | undefined,
+    done: (release?: boolean | Error) => void
+  ) => void): Promise<pg.PoolClient> | void => {
+    const pendingClient = sanitizedConnect();
+    if (!callback) return pendingClient;
+
+    pendingClient.then(
+      (client) => callback(undefined, client, client.release.bind(client)),
+      (error: Error) => callback(error, undefined, () => undefined)
+    );
+  }) as typeof pool.connect;
+
+  checkoutSanitizedPools.add(pool);
+  return pool;
+}

postgres/pg-query-context/benchmarks/README.md

@@ -0,0 +1,25 @@
+# Checkout sanitation cost
+
+Run after building `pg-cache`, `pg-query-context`, and the `pgsql-test` dependencies, against an isolated PostgreSQL instance with the pgpm test users bootstrapped:
+
+```sh
+node postgres/pg-query-context/benchmarks/checkout-sanitation.cjs /tmp/checkout-sanitation.json 1000
+```
+
+Connection options come through the existing `pgsql-test` environment provider. The harness creates and drops its own database. The baseline uses a harness-owned unsanitized pool; the comparison uses the actual default `pg-cache` factory. Both use the application login, one client, and concurrency one. Each arm warms up for 50 operations; three rounds alternate arm order, with 1,000 measured operations per arm/workload/round. Results contain latency percentiles and throughput.
+
+## Local result, 2026-09-09
+
+PostgreSQL 18.6, Node 24.20.0, localhost TCP. Values below are the median of each metric across the three rounds; the raw result is `checkout-sanitation.pg18-local.json`.
+
+| Workload | p50 ms, baseline → sanitized | p95 ms, baseline → sanitized | Operations/s, baseline → sanitized |
+| --- | --- | --- | --- |
+| Checkout and release | 0.013 → 0.322 | 0.017 → 1.947 | 33,818 → 1,519 |
+| Named prepared SELECT | 0.383 → 0.825 | 1.981 → 3.378 | 1,749 → 632 |
+| Request-context transaction + named SELECT | 1.884 → 2.143 | 5.845 → 5.293 | 401 → 371 |
+
+This shared host had unrelated CPU load. These measurements establish local cost, not production percentiles or a performance gate. In particular, the lower transaction p95 in the sanitized arm is noise, not evidence that sanitation improves tail latency. Repeat on representative deployment hardware, network latency, pool sizes, and query mixes before adopting a throughput budget.
+
+`DISCARD ALL` adds a server round trip to every checkout and removes prepared statements: the final baseline checkout retained one named prepared statement, while the sanitized checkout retained none. The prepared-query arm lost about 64% throughput locally; the complete transaction arm lost about 7%. A cheap query-heavy workload therefore needs particular scrutiny.
+
+The PR retains the fail-closed default while making this cost reviewable. A future cheaper reset must prove equivalent removal of roles/GUCs, temporary state, LISTEN state, advisory locks, and prepared-statement bookkeeping before replacing it. Merely using `RESET ALL`, or skipping cleanup based on assumptions about callers, does not provide that equivalence. This benchmark does not justify such a replacement or an opt-out.

postgres/pg-query-context/benchmarks/checkout-sanitation.cjs

@@ -0,0 +1,78 @@
+/* Non-gating microbenchmark. Run after building this package and its test dependencies. */
+const { performance } = require('node:perf_hooks');
+const { writeFileSync } = require('node:fs');
+const { getConnections } = require('pgsql-test');
+const { defaultPgPoolFactory } = require('pg-cache');
+const { withPgClient } = require('../dist');
+
+async function run() {
+  const samples = Number(process.argv[3] ?? 500);
+  if (!Number.isSafeInteger(samples) || samples < 100) {
+    throw new Error('samples must be an integer >= 100');
+  }
+  const fixture = await getConnections({}, []);
+  let sanitized;
+  try {
+    const config = { ...fixture.db.config, max: 1 };
+    const baseline = fixture.manager.getPool(config);
+    // This factory is the behavior under measurement; the fixture owns the DB.
+    sanitized = defaultPgPoolFactory({ ...fixture.db.config, pool: { max: 1 } });
+    const server = await baseline.query('SHOW server_version');
+    const prepared = { name: 'checkout_cost', text: 'SELECT 1 AS value' };
+    const workloads = {
+      checkout: async pool => {
+        const client = await pool.connect();
+        client.release();
+      },
+      prepared_select: async pool => {
+        const client = await pool.connect();
+        try { await client.query(prepared); }
+        finally { client.release(); }
+      },
+      transaction: async pool => withPgClient(pool, {
+        role: fixture.db.config.user,
+        row_security: 'on',
+        search_path: 'pg_catalog',
+        'jwt.claims.user_id': '',
+      }, client => client.query(prepared)),
+    };
+    const results = [];
+    for (let round = 0; round < 3; round++) {
+      for (const [workload, operation] of Object.entries(workloads)) {
+        const arms = round % 2 ? [['sanitized', sanitized], ['baseline', baseline]]
+          : [['baseline', baseline], ['sanitized', sanitized]];
+        for (const [arm, pool] of arms) {
+          for (let i = 0; i < 50; i++) await operation(pool);
+          const latencies = [];
+          const started = performance.now();
+          for (let i = 0; i < samples; i++) {
+            const before = performance.now();
+            await operation(pool);
+            latencies.push(performance.now() - before);
+          }
+          const elapsedMs = performance.now() - started;
+          latencies.sort((a, b) => a - b);
+          const percentile = p => latencies[Math.ceil(samples * p) - 1];
+          results.push({ round: round + 1, workload, arm, samples, elapsedMs,
+            operationsPerSecond: samples * 1000 / elapsedMs,
+            p50Ms: percentile(0.5), p95Ms: percentile(0.95), p99Ms: percentile(0.99) });
+        }
+      }
+    }
+    const preparedState = {};
+    for (const [arm, pool] of [['baseline', baseline], ['sanitized', sanitized]]) {
+      const result = await pool.query("SELECT count(*)::int AS count FROM pg_prepared_statements WHERE name = 'checkout_cost'");
+      preparedState[arm] = result.rows[0].count;
+    }
+    const report = { timestamp: new Date().toISOString(), node: process.version,
+      postgres: server.rows[0].server_version, clients: 1, concurrency: 1,
+      warmupPerArm: 50, order: 'alternating by round', preparedState, results };
+    const output = JSON.stringify(report, null, 2) + '\n';
+    if (process.argv[2]) writeFileSync(process.argv[2], output);
+    else process.stdout.write(output);
+  } finally {
+    try { if (sanitized) await sanitized.end(); }
+    finally { await fixture.teardown(); }
+  }
+}
+run().catch(error => { console.error(error); process.exitCode = 1; });