From f2bcd121cd566bd5c53d9c12571d69cb62ee14e2 Mon Sep 17 00:00:00 2001 From: zetazzz Date: Mon, 17 Aug 2026 16:54:43 +0800 Subject: [PATCH] Add exact PostgreSQL pool identities and leases --- pnpm-lock.yaml | 3 - postgres/pg-cache/package.json | 1 - .../pg-cache/src/__tests__/driver.test.ts | 265 ++++++++- postgres/pg-cache/src/__tests__/lru.test.ts | 240 +++++++- postgres/pg-cache/src/driver.ts | 19 +- postgres/pg-cache/src/index.ts | 28 +- postgres/pg-cache/src/lru.ts | 554 +++++++++++++++--- postgres/pg-cache/src/pg.ts | 373 +++++++++++- postgres/pg-env/src/pg-config.ts | 31 +- 9 files changed, 1369 insertions(+), 145 deletions(-) diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 40e3cef7e9..477e15ef0f 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -3482,9 +3482,6 @@ importers: '@pgpmjs/types': specifier: workspace:^ version: link:../../pgpm/types/dist - lru-cache: - specifier: ^11.2.7 - version: 11.2.7 pg: specifier: ^8.21.0 version: 8.21.0 diff --git a/postgres/pg-cache/package.json b/postgres/pg-cache/package.json index 8d5e80788f..2b849b3e01 100644 --- a/postgres/pg-cache/package.json +++ b/postgres/pg-cache/package.json @@ -32,7 +32,6 @@ "12factor-env": "workspace:^", "@pgpmjs/logger": "workspace:^", "@pgpmjs/types": "workspace:^", - "lru-cache": "^11.2.7", "pg": "^8.21.0", "pg-env": "workspace:^" }, diff --git a/postgres/pg-cache/src/__tests__/driver.test.ts b/postgres/pg-cache/src/__tests__/driver.test.ts index 105e3a669a..6613c2311e 100644 --- a/postgres/pg-cache/src/__tests__/driver.test.ts +++ b/postgres/pg-cache/src/__tests__/driver.test.ts @@ -3,25 +3,39 @@ // lets an alternate backend (e.g. PGlite) plug in without any change to pgpm / // pgsql-* — and guarantees the default path is untouched when nothing registers. -import { randomUUID } from 'crypto'; +import { createHash, randomUUID } from 'crypto'; import pg from 'pg'; import { + acquirePgPool, defaultPgPoolFactory, getActivePgPoolFactory, getPgPool, + getPgPoolConfig, + getPgPoolDriverIdentity, + getPgPoolIdentity, hasPgPoolFactory, PgPoolFactory, - registerPgPoolFactory + registerPgPoolFactory, } from '../index'; import { pgCache } from '../lru'; const createMockPool = (): pg.Pool => - ({ query: jest.fn(), connect: jest.fn(), end: jest.fn(async () => {}) } as unknown as pg.Pool); + ({ + query: jest.fn(), + connect: jest.fn(), + end: jest.fn(async () => {}), + }) as unknown as pg.Pool; const freshConfig = () => { const database = `seam_${randomUUID()}`; - return { database, host: 'localhost', port: 5432, user: 'postgres', password: 'x' }; + return { + database, + host: 'localhost', + port: 5432, + user: 'postgres', + password: 'x', + }; }; describe('pg-cache pool-factory seam', () => { @@ -57,10 +71,10 @@ describe('pg-cache pool-factory seam', () => { expect(pool).toBe(mock); expect(pool.connect).toBe(alternateConnect); - pgCache.delete(cfg.database); + pgCache.delete(getPgPoolIdentity(cfg)); }); - it('caches by database: a second call reuses the pool and does not re-invoke the factory', () => { + it('caches by exact identity: an identical call reuses the pool', () => { const cfg = freshConfig(); const factory = jest.fn(() => createMockPool()); registerPgPoolFactory(factory); @@ -71,7 +85,210 @@ describe('pg-cache pool-factory seam', () => { expect(first).toBe(second); expect(factory).toHaveBeenCalledTimes(1); - pgCache.delete(cfg.database); + pgCache.delete(getPgPoolIdentity(cfg)); + }); + + it('leases the exact identity with idempotent release', () => { + const cfg = freshConfig(); + const mock = createMockPool(); + const factory = jest.fn(() => mock); + registerPgPoolFactory(factory); + + const first = acquirePgPool(cfg, { purpose: 'runtime' }); + const second = acquirePgPool(cfg, { purpose: 'runtime' }); + + expect(first.identity).toBe(getPgPoolIdentity(cfg, { purpose: 'runtime' })); + expect(first.pool).toBe(mock); + expect(second.pool).toBe(mock); + expect(factory).toHaveBeenCalledTimes(1); + expect(pgCache.getStats().activeLeases).toBeGreaterThanOrEqual(2); + + first.release(); + first.release(); + second.release(); + pgCache.delete(first.identity); + }); + + it('separates credentials and purpose for one physical database', () => { + const cfg = freshConfig(); + const factory = jest.fn(() => createMockPool()); + registerPgPoolFactory(factory); + + const control = getPgPool(cfg, { purpose: 'control' }); + const runtime = getPgPool( + { ...cfg, user: 'runtime', password: 'runtime-secret' }, + { purpose: 'runtime' } + ); + const notification = getPgPool( + { ...cfg, user: 'runtime', password: 'runtime-secret' }, + { purpose: 'notification' } + ); + + expect(control).not.toBe(runtime); + expect(runtime).not.toBe(notification); + expect(factory).toHaveBeenCalledTimes(3); + + pgCache.delete(getPgPoolIdentity(cfg, { purpose: 'control' })); + pgCache.delete( + getPgPoolIdentity( + { ...cfg, user: 'runtime', password: 'runtime-secret' }, + { purpose: 'runtime' } + ) + ); + pgCache.delete( + getPgPoolIdentity( + { ...cfg, user: 'runtime', password: 'runtime-secret' }, + { purpose: 'notification' } + ) + ); + }); + + it('normalizes maxUses into the exact pool identity', () => { + const cfg = freshConfig(); + const unlimited = getPgPoolIdentity(cfg); + const explicitUnlimited = getPgPoolIdentity({ + ...cfg, + pool: { maxUses: 0 }, + }); + const singleUse = getPgPoolIdentity({ ...cfg, pool: { maxUses: 1 } }); + + expect(explicitUnlimited).toBe(unlimited); + expect(singleUse).not.toBe(unlimited); + }); + + it('uses a process-keyed identity instead of an offline password verifier', () => { + const cfg = { + ...freshConfig(), + pool: { + max: 3, + idleTimeoutMillis: 1234, + connectionTimeoutMillis: 5678, + allowExitOnIdle: true, + }, + }; + const identity = getPgPoolIdentity(cfg, { purpose: 'runtime' }); + const unkeyedInput = JSON.stringify({ + version: 1, + driver: getPgPoolDriverIdentity(), + host: cfg.host, + port: cfg.port, + database: cfg.database, + user: cfg.user, + password: cfg.password, + ssl: null, + pool: { + max: 3, + maxUses: null, + idleTimeoutMillis: 1234, + connectionTimeoutMillis: 5678, + allowExitOnIdle: true, + }, + purpose: 'runtime', + checkout: 'registered-factory-owned-v1', + }); + const offlineDigest = `pg:v1:${createHash('sha256') + .update(unkeyedInput) + .digest('hex')}`; + + expect(getPgPoolIdentity(cfg, { purpose: 'runtime' })).toBe(identity); + expect(identity).toMatch(/^pg:v1:[a-f0-9]{64}$/); + expect(identity).not.toBe(offlineDigest); + }); + + it('rejects credential callbacks and noncanonical identity inputs', () => { + const cfg = freshConfig(); + const accessorSsl = {} as Record; + Object.defineProperty(accessorSsl, 'ca', { get: () => 'dynamic-ca' }); + + expect(() => + getPgPoolIdentity({ + ...cfg, + password: (async () => 'secret') as unknown as string, + }) + ).toThrow('pg.password must be a string'); + expect(() => + getPgPoolIdentity({ + ...cfg, + port: '5432' as unknown as number, + }) + ).toThrow('pg.port must be a safe integer'); + expect(() => + getPgPoolIdentity(cfg, { + purpose: {} as unknown as string, + }) + ).toThrow('pg pool purpose must be a non-empty string'); + expect(() => + getPgPoolIdentity({ + ...cfg, + ssl: accessorSsl as never, + }) + ).toThrow('pg.ssl.ca must be a data property'); + }); + + it('canonicalizes TLS data but separates distinct trust contracts', () => { + const cfg = freshConfig(); + const verified = getPgPoolIdentity({ + ...cfg, + ssl: { + ca: 'tenant-ca', + rejectUnauthorized: true, + servername: 'db.internal', + }, + }); + const reordered = getPgPoolIdentity({ + ...cfg, + ssl: { + servername: 'db.internal', + rejectUnauthorized: true, + ca: 'tenant-ca', + }, + }); + const insecure = getPgPoolIdentity({ + ...cfg, + ssl: { + ca: 'tenant-ca', + rejectUnauthorized: false, + servername: 'db.internal', + }, + }); + + expect(reordered).toBe(verified); + expect(insecure).not.toBe(verified); + expect(() => + getPgPoolIdentity({ + ...cfg, + ssl: { checkServerIdentity: (): undefined => undefined } as any, + }) + ).toThrow('must contain only deterministic data values'); + }); + + it('never discloses credentials in an identity', () => { + const cfg = { ...freshConfig(), password: 'top-secret-password' }; + const identity = getPgPoolIdentity(cfg, { purpose: 'runtime' }); + + expect(identity).toMatch(/^pg:v1:[a-f0-9]{64}$/); + expect(identity).not.toContain(cfg.user); + expect(identity).not.toContain(cfg.password); + }); + + it('parses and validates maxUses before constructing a pool', () => { + const previous = process.env.PG_POOL_MAX_USES; + try { + process.env.PG_POOL_MAX_USES = '0'; + expect(getPgPoolConfig().maxUses).toBeUndefined(); + process.env.PG_POOL_MAX_USES = '17'; + expect(getPgPoolConfig().maxUses).toBe(17); + process.env.PG_POOL_MAX_USES = '1e2'; + expect(() => getPgPoolConfig()).toThrow( + 'PG_POOL_MAX_USES must be 0 or a positive safe integer' + ); + expect(() => getPgPoolConfig({ maxUses: -1 })).toThrow( + 'pool.maxUses must be 0 or a positive safe integer' + ); + } finally { + if (previous === undefined) delete process.env.PG_POOL_MAX_USES; + else process.env.PG_POOL_MAX_USES = previous; + } }); it('falls back to defaultPgPoolFactory when nothing is registered', () => { @@ -80,7 +297,7 @@ describe('pg-cache pool-factory seam', () => { // query runs, so this is safe without a live server. const pool = getPgPool(cfg); expect(pool).toBeInstanceOf(pg.Pool); - pgCache.delete(cfg.database); + pgCache.delete(getPgPoolIdentity(cfg)); }); it('defaultPgPoolFactory returns a pg.Pool', () => { @@ -88,4 +305,36 @@ describe('pg-cache pool-factory seam', () => { expect(pool).toBeInstanceOf(pg.Pool); return pool.end(); }); + + it('passes exact credentials, pool limits, and TLS fields to node-postgres', async () => { + const ssl = { + ca: 'tenant-ca', + cert: 'runtime-cert', + key: 'runtime-key', + rejectUnauthorized: true, + servername: 'db.internal', + minVersion: 'TLSv1.2' as const, + }; + const cfg = { + ...freshConfig(), + user: 'runtime@tenant', + password: 'x@evil.example/other?sslmode=require', + database: 'tenant/database', + ssl, + pool: { maxUses: 1 }, + }; + const pool = defaultPgPoolFactory(cfg); + const options = (pool as pg.Pool & { options: pg.PoolConfig }).options; + + expect(options).toMatchObject({ + host: cfg.host, + port: cfg.port, + database: cfg.database, + user: cfg.user, + password: cfg.password, + maxUses: 1, + ssl, + }); + await pool.end(); + }); }); diff --git a/postgres/pg-cache/src/__tests__/lru.test.ts b/postgres/pg-cache/src/__tests__/lru.test.ts index a68afa616f..c853641424 100644 --- a/postgres/pg-cache/src/__tests__/lru.test.ts +++ b/postgres/pg-cache/src/__tests__/lru.test.ts @@ -1,22 +1,37 @@ -// Guards against the pg-cache close() resource leak fixed in feat/observability. -// -// Previously, close() reset this.closed = false after shutdown, allowing -// set() to silently accept new pools that were never cleaned up. The module- -// level closePromise also reset to null, enabling double-shutdown. -// -// These tests lock the fix: close() is final, set() rejects, and repeated -// close() calls are idempotent. See pg-cache-close-leak.md for full details. - import pg from 'pg'; -import { PgPoolCacheManager } from '../lru'; +import { + DEFAULT_PG_CACHE_MAX, + PG_CACHE_GRAPHILE_CONTRACT_CAPACITY, + PG_CACHE_OPERATIONAL_RESERVE, + PgPoolCacheManager, + PgPoolCapacityError, +} from '../lru'; + +describe('process lifecycle ownership', () => { + it('does not install process signal handlers from a library import', () => { + const beforeSigterm = process.listenerCount('SIGTERM'); + const beforeSigint = process.listenerCount('SIGINT'); + + jest.isolateModules(() => { + jest.requireActual('../lru'); + }); + + expect(process.listenerCount('SIGTERM')).toBe(beforeSigterm); + expect(process.listenerCount('SIGINT')).toBe(beforeSigint); + }); +}); // Minimal mock — we only need pool.end() and pool.ended const createMockPool = (): pg.Pool => { let ended = false; return { - get ended() { return ended; }, - end: jest.fn(async () => { ended = true; }), + get ended() { + return ended; + }, + end: jest.fn(async () => { + ended = true; + }), } as unknown as pg.Pool; }; @@ -29,7 +44,11 @@ describe('PgPoolCacheManager', () => { afterEach(async () => { // Ensure all pools are cleaned up even if a test fails mid-way - try { await cache.close(); } catch { /* already closed */ } + try { + await cache.close(); + } catch { + /* already closed */ + } }); it('stores and retrieves a pool', () => { @@ -45,8 +64,11 @@ describe('PgPoolCacheManager', () => { }); describe('configuration', () => { - it('uses env-var defaults (max=50) when no overrides given', () => { - expect(cache.config.max).toBe(50); + it('reserves two identities per supported Graphile contract plus operations', () => { + expect(DEFAULT_PG_CACHE_MAX).toBe( + PG_CACHE_GRAPHILE_CONTRACT_CAPACITY * 2 + PG_CACHE_OPERATIONAL_RESERVE + ); + expect(cache.config.max).toBe(2064); }); it('accepts constructor overrides', () => { @@ -90,6 +112,171 @@ describe('PgPoolCacheManager', () => { }); }); + describe('leases and fail-closed admission', () => { + it('keeps database-name deletion as a migration alias for opaque keys', async () => { + const pool = createMockPool(); + cache.set('pg:v1:opaque', pool); + cache.registerAlias('tenant_a', 'pg:v1:opaque'); + + cache.delete('tenant_a'); + await cache.waitForDisposals(); + + expect(cache.has('pg:v1:opaque')).toBe(false); + expect(pool.end).toHaveBeenCalledTimes(1); + }); + + it('counts an existing exact identity as zero new slots', async () => { + const small = new PgPoolCacheManager({ max: 1 }); + const pool = createMockPool(); + const factory = jest.fn(() => pool); + + const first = small.acquire('runtime-a', factory); + const second = small.acquire('runtime-a', factory); + + expect(first.pool).toBe(pool); + expect(second.pool).toBe(pool); + expect(factory).toHaveBeenCalledTimes(1); + expect(small.getStats()).toMatchObject({ + size: 1, + leasedPools: 1, + activeLeases: 2, + leasesAcquired: 2, + }); + + first.release(); + first.release(); + expect(small.getStats().activeLeases).toBe(1); + second.release(); + await small.close(); + }); + + it('refuses before constructing or ending when every slot is leased', async () => { + const small = new PgPoolCacheManager({ max: 1 }); + const firstPool = createMockPool(); + const first = small.acquire('runtime-a', () => firstPool); + const rejectedFactory = jest.fn(() => createMockPool()); + + let capacityError: PgPoolCapacityError | undefined; + try { + small.acquire('runtime-b', rejectedFactory); + } catch (error) { + capacityError = error as PgPoolCapacityError; + } + + expect(capacityError).toBeInstanceOf(PgPoolCapacityError); + expect(capacityError).toMatchObject({ + code: 'PG_POOL_CAPACITY', + retryAfterSeconds: 15, + max: 1, + size: 1, + leased: 1, + }); + expect(rejectedFactory).not.toHaveBeenCalled(); + expect(firstPool.end).not.toHaveBeenCalled(); + expect(small.getStats().capacityRefusals).toBe(1); + + first.release(); + await small.close(); + }); + + it('evicts only the least-recent zero-lease identity', async () => { + const small = new PgPoolCacheManager({ max: 2 }); + const leasedPool = createMockPool(); + const idlePool = createMockPool(); + const replacementPool = createMockPool(); + const lease = small.acquire('leased', () => leasedPool); + small.set('idle', idlePool); + + small.set('replacement', replacementPool); + await small.waitForDisposals(); + + expect(small.has('leased')).toBe(true); + expect(leasedPool.end).not.toHaveBeenCalled(); + expect(small.has('idle')).toBe(false); + expect(idlePool.end).toHaveBeenCalledTimes(1); + expect(small.has('replacement')).toBe(true); + + lease.release(); + await small.close(); + }); + + it('keeps an expired leased identity until release', async () => { + jest.useFakeTimers(); + const small = new PgPoolCacheManager({ max: 1, ttl: 50 }); + const pool = createMockPool(); + const lease = small.acquire('runtime', () => pool); + try { + jest.advanceTimersByTime(51); + expect(small.has('runtime')).toBe(true); + expect(pool.end).not.toHaveBeenCalled(); + + lease.release(); + await small.waitForDisposals(); + + expect(small.has('runtime')).toBe(false); + expect(pool.end).toHaveBeenCalledTimes(1); + expect(small.getStats().ttlExpirations).toBe(1); + } finally { + jest.useRealTimers(); + await small.close(); + } + }); + + it('deterministically gives the final slot to the first synchronous acquisition', async () => { + const small = new PgPoolCacheManager({ max: 1 }); + const firstFactory = jest.fn(() => createMockPool()); + const secondFactory = jest.fn(() => createMockPool()); + + const outcomes = await Promise.allSettled([ + Promise.resolve().then(() => small.acquire('first', firstFactory)), + Promise.resolve().then(() => small.acquire('second', secondFactory)), + ]); + + expect(outcomes[0].status).toBe('fulfilled'); + expect(outcomes[1].status).toBe('rejected'); + expect((outcomes[1] as PromiseRejectedResult).reason).toBeInstanceOf( + PgPoolCapacityError + ); + expect(firstFactory).toHaveBeenCalledTimes(1); + expect(secondFactory).not.toHaveBeenCalled(); + + if (outcomes[0].status === 'fulfilled') outcomes[0].value.release(); + await small.close(); + }); + + it('rolls back its reservation if pool construction fails', async () => { + const small = new PgPoolCacheManager({ max: 1 }); + const retained = createMockPool(); + small.set('retained', retained); + + expect(() => + small.acquire('broken', () => { + throw new Error('factory failed'); + }) + ).toThrow('factory failed'); + + expect(small.has('retained')).toBe(true); + expect(retained.end).not.toHaveBeenCalled(); + expect(small.getStats()).toMatchObject({ size: 1, reservations: 0 }); + await small.close(); + }); + + it('does not end a physical pool retained under another exact identity', async () => { + const small = new PgPoolCacheManager({ max: 1 }); + const sharedPool = createMockPool(); + + small.set('identity-a', sharedPool); + small.set('identity-b', sharedPool); + await small.waitForDisposals(); + expect(sharedPool.end).not.toHaveBeenCalled(); + + small.delete('identity-b'); + await small.waitForDisposals(); + expect(sharedPool.end).toHaveBeenCalledTimes(1); + await small.close(); + }); + }); + describe('close() lifecycle', () => { it('set() after close() succeeds (cache re-opens for restart)', async () => { const pool1 = createMockPool(); @@ -122,6 +309,29 @@ describe('PgPoolCacheManager', () => { expect(pool.end).toHaveBeenCalledTimes(1); }); + it('makes concurrent close callers await the same pool teardown', async () => { + let finishEnd!: () => void; + const ended = new Promise((resolve) => { + finishEnd = resolve; + }); + const pool = createMockPool(); + (pool.end as jest.Mock).mockImplementation(() => ended); + cache.set('key1', pool); + + const first = cache.close(); + const second = cache.close(); + let secondSettled = false; + void second.then(() => { + secondSettled = true; + }); + await Promise.resolve(); + + expect(secondSettled).toBe(false); + finishEnd(); + await Promise.all([first, second]); + expect(pool.end).toHaveBeenCalledTimes(1); + }); + it('close() disposes all pools', async () => { const pool1 = createMockPool(); const pool2 = createMockPool(); diff --git a/postgres/pg-cache/src/driver.ts b/postgres/pg-cache/src/driver.ts index 9a6c22ffb7..b110a1bcab 100644 --- a/postgres/pg-cache/src/driver.ts +++ b/postgres/pg-cache/src/driver.ts @@ -14,10 +14,14 @@ import type { PgConfig, PgPoolConfig } from 'pg-env'; * `end()` (plus an `ended` flag for disposal), so a factory may return anything * implementing that subset — `QueryablePool`. A real `pg.Pool` structurally * satisfies it, so the default path is unchanged and fully backward-compatible. + * The default TCP factory owns the checkout sanitation contract. Registered + * factories retain their existing behavior and advertise a separate driver + * identity because they own their own checkout semantics. */ export interface QueryableClient { query(text: string, values?: any[]): Promise; - release(...args: any[]): void; + /** A truthy error argument must permanently discard this client. */ + release(error?: Error | boolean): void; } export interface QueryablePool { @@ -27,10 +31,16 @@ export interface QueryablePool { } export type PgPoolFactory = ( - config: Partial & { pool?: PgPoolConfig } + config: Partial & { pool?: PgPoolConfig }, + options?: PgPoolFactoryOptions ) => pg.Pool | QueryablePool; +export interface PgPoolFactoryOptions { + purpose: string; +} + let activeFactory: PgPoolFactory | undefined; +let driverGeneration = 0; /** * Register the factory `getPgPool` uses to build new pools. Pass `undefined` @@ -42,6 +52,7 @@ let activeFactory: PgPoolFactory | undefined; */ export const registerPgPoolFactory = (factory: PgPoolFactory | undefined): void => { activeFactory = factory; + driverGeneration++; }; /** The currently-registered factory, or `undefined` when using the default. */ @@ -49,3 +60,7 @@ export const getActivePgPoolFactory = (): PgPoolFactory | undefined => activeFac /** Whether a non-default pool factory is currently registered. */ export const hasPgPoolFactory = (): boolean => activeFactory !== undefined; + +/** Stable until the active factory registration changes. */ +export const getPgPoolDriverIdentity = (): string => + activeFactory ? `registered:${driverGeneration}` : 'node-postgres'; diff --git a/postgres/pg-cache/src/index.ts b/postgres/pg-cache/src/index.ts index 286748297a..369a4c039d 100644 --- a/postgres/pg-cache/src/index.ts +++ b/postgres/pg-cache/src/index.ts @@ -1,23 +1,39 @@ // Main exports from pg-cache package export { getActivePgPoolFactory, + getPgPoolDriverIdentity, hasPgPoolFactory, registerPgPoolFactory } from './driver'; export { close, + DEFAULT_PG_CACHE_MAX, getPgCacheConfig, - pgCache, - PgPoolCacheManager, + getPgCacheStats, + PG_CACHE_GRAPHILE_CONTRACT_CAPACITY, + PG_CACHE_OPERATIONAL_RESERVE, + PG_POOL_CAPACITY_ERROR_CODE, + pgCache, + PgPoolCacheManager, + PgPoolCapacityError, teardownPgPools } from './lru'; export { + acquirePgPool, buildConnectionString, defaultPgPoolFactory, + getPgDatabaseTargetIdentity, getPgPool, - getPgPoolConfig + getPgPoolConfig, + getPgPoolIdentity } from './pg'; - // Re-export types -export type { PgPoolFactory, QueryableClient, QueryablePool } from './driver'; -export type { PgCacheConfig, PoolCleanupCallback } from './lru'; \ No newline at end of file +export type { PgPoolFactory, PgPoolFactoryOptions, QueryableClient, QueryablePool } from './driver'; +export type { + PgCacheConfig, + PgPoolCacheStats, + PgPoolDisposalReason, + PgPoolLease, + PoolCleanupCallback, +} from './lru'; +export type { GetPgPoolOptions } from './pg'; diff --git a/postgres/pg-cache/src/lru.ts b/postgres/pg-cache/src/lru.ts index 633dc6388e..38fe32d0e5 100644 --- a/postgres/pg-cache/src/lru.ts +++ b/postgres/pg-cache/src/lru.ts @@ -1,6 +1,5 @@ import { Logger } from '@pgpmjs/logger'; import { parseEnvNumber } from '12factor-env'; -import { LRUCache } from 'lru-cache'; import pg from 'pg'; const log = new Logger('pg-cache'); @@ -9,58 +8,136 @@ const ONE_HOUR_IN_MS = 1000 * 60 * 60; const ONE_DAY = ONE_HOUR_IN_MS * 24; const ONE_YEAR = ONE_DAY * 366; -// Kubernetes sends only SIGTERM on pod shutdown -const SYS_EVENTS = ['SIGTERM']; +// One runtime and one control identity per database-per-tenant Graphile +// contract, plus room for routing, diagnostics, listeners, and build overlap. +export const PG_CACHE_GRAPHILE_CONTRACT_CAPACITY = 1024; +export const PG_CACHE_OPERATIONAL_RESERVE = 16; +export const DEFAULT_PG_CACHE_MAX = + PG_CACHE_GRAPHILE_CONTRACT_CAPACITY * 2 + PG_CACHE_OPERATIONAL_RESERVE; type PgPoolKey = string; +type PoolFactory = () => pg.Pool; -// Cleanup callback type - called when a pg pool is disposed +export type PgPoolDisposalReason = + 'capacity' | 'ttl' | 'delete' | 'clear' | 'close' | 'replace'; + +// Called only when an identity is actually removed from the registry. export type PoolCleanupCallback = (pgPoolKey: string) => void; -// --- Cache Configuration --- +export interface PgPoolLease { + pool: pg.Pool; + identity: string; + /** Idempotently release this exact ownership claim. */ + release(): void; +} export interface PgCacheConfig { - /** Maximum number of pools in the LRU cache (env: PG_CACHE_MAX, default: 50) */ + /** Maximum number of lazy pool identities retained by this process. */ max: number; - /** TTL for cached pools in ms (default: ONE_YEAR) */ + /** Idle identity TTL in milliseconds. Leased identities never expire. */ ttl: number; } -/** - * Read cache configuration from environment variables. - * - * Supports: - * - PG_CACHE_MAX: Maximum number of pools (default: 50) - * - PG_CACHE_TTL_MS: TTL in milliseconds (default: ONE_YEAR) - */ +export interface PgPoolCacheStats { + size: number; + max: number; + ttl: number; + leasedPools: number; + idlePools: number; + activeLeases: number; + reservations: number; + pendingDisposals: number; + hits: number; + misses: number; + poolsCreated: number; + leasesAcquired: number; + leasesReleased: number; + capacityEvictions: number; + ttlExpirations: number; + capacityRefusals: number; + disposalsStarted: number; + disposalsCompleted: number; + disposalFailures: number; +} + +interface PgPoolCacheCounters { + hits: number; + misses: number; + poolsCreated: number; + leasesAcquired: number; + leasesReleased: number; + capacityEvictions: number; + ttlExpirations: number; + capacityRefusals: number; + disposalsStarted: number; + disposalsCompleted: number; + disposalFailures: number; +} + +interface SlotReservation { + key: PgPoolKey; + victims: ManagedPgPool[]; +} + +export const PG_POOL_CAPACITY_ERROR_CODE = 'PG_POOL_CAPACITY'; + +/** Fail-closed pool admission error suitable for a stable HTTP 503 mapping. */ +export class PgPoolCapacityError extends Error { + readonly code = PG_POOL_CAPACITY_ERROR_CODE; + readonly retryAfterSeconds = 15; + + constructor( + readonly max: number, + readonly size: number, + readonly leased: number + ) { + super( + `PostgreSQL pool capacity exhausted: ${size}/${max} identities are retained ` + + `and ${leased} are leased` + ); + this.name = 'PgPoolCapacityError'; + } +} + +/** Read cache configuration without allocating any pools or connections. */ export function getPgCacheConfig(): PgCacheConfig { return { - max: parseEnvNumber(process.env.PG_CACHE_MAX) ?? 50, + max: parseEnvNumber(process.env.PG_CACHE_MAX) ?? DEFAULT_PG_CACHE_MAX, ttl: parseEnvNumber(process.env.PG_CACHE_TTL_MS) ?? ONE_YEAR, }; } class ManagedPgPool { public isDisposed = false; + public leaseCount = 0; + public lastAccessOrder = 0; + public expiresAt = 0; private disposePromise: Promise | null = null; - constructor(public readonly pool: pg.Pool, public readonly key: string) {} + constructor( + public readonly pool: pg.Pool, + public readonly key: string + ) {} + + touch(order: number, now: number, ttl: number): void { + this.lastAccessOrder = order; + this.expiresAt = now + ttl; + } + + isExpired(now: number): boolean { + return now >= this.expiresAt; + } async dispose(): Promise { if (this.isDisposed) return this.disposePromise; this.isDisposed = true; this.disposePromise = (async () => { - try { - if (!this.pool.ended) { - await this.pool.end(); - log.success(`pg.Pool ${this.key} ended.`); - } else { - log.info(`pg.Pool ${this.key} already ended.`); - } - } catch (err) { - log.error(`Error ending pg.Pool ${this.key}: ${(err as Error).message}`); - throw err; + if (!this.pool.ended) { + await this.pool.end(); + log.success(`pg.Pool ${this.key} ended.`); + } else { + log.info(`pg.Pool ${this.key} already ended.`); } })(); @@ -68,37 +145,75 @@ class ManagedPgPool { } } +/** + * A lease-aware, lazy pool registry. + * + * JavaScript executes acquisition synchronously, including slot reservation and + * factory invocation. Two callers therefore cannot both claim the final slot. + * Pools may finish ending asynchronously after a zero-lease identity is removed. + */ export class PgPoolCacheManager { - private cleanupTasks: Promise[] = []; + private readonly records = new Map(); + private readonly cleanupTasks = new Set>(); + private readonly cleanupCallbacks = new Set(); + private readonly aliasKeys = new Map>(); + private readonly keyAliases = new Map>(); + private readonly reservedKeys = new Set(); + private reservations = 0; + private accessOrder = 0; private closed = false; - private cleanupCallbacks: Set = new Set(); + private closePromise: Promise | null = null; readonly config: PgCacheConfig; - private readonly pgCache: LRUCache; + private readonly counters: PgPoolCacheCounters = { + hits: 0, + misses: 0, + poolsCreated: 0, + leasesAcquired: 0, + leasesReleased: 0, + capacityEvictions: 0, + ttlExpirations: 0, + capacityRefusals: 0, + disposalsStarted: 0, + disposalsCompleted: 0, + disposalFailures: 0, + }; constructor(config?: Partial) { const defaults = getPgCacheConfig(); this.config = { ...defaults, ...config }; + if (!Number.isSafeInteger(this.config.max) || this.config.max <= 0) { + throw new Error('pg-cache max must be a positive safe integer'); + } + if (!Number.isFinite(this.config.ttl) || this.config.ttl <= 0) { + throw new Error('pg-cache ttl must be a positive number'); + } + } - this.pgCache = new LRUCache({ - max: this.config.max, - ttl: this.config.ttl, - updateAgeOnGet: true, - dispose: (managedPool, key, reason) => { - log.debug(`Disposing pg pool [${key}] (${reason})`); - this.notifyCleanup(key); - this.disposePool(managedPool); - } - }); + get size(): number { + return this.records.size; } - // Register a cleanup callback to be called when pools are disposed registerCleanupCallback(callback: PoolCleanupCallback): () => void { this.cleanupCallbacks.add(callback); - // Return unregister function - return () => { - this.cleanupCallbacks.delete(callback); - }; + return () => this.cleanupCallbacks.delete(callback); + } + + /** Preserve database-name cleanup for callers migrating to opaque keys. */ + registerAlias(alias: string, key: PgPoolKey): void { + if (!this.records.has(key)) return; + let keys = this.aliasKeys.get(alias); + if (!keys) { + keys = new Set(); + this.aliasKeys.set(alias, keys); + } + keys.add(key); + let aliases = this.keyAliases.get(key); + if (!aliases) { + aliases = new Set(); + this.keyAliases.set(key, aliases); + } + aliases.add(alias); } get(key: PgPoolKey): pg.Pool | undefined { @@ -106,51 +221,335 @@ export class PgPoolCacheManager { log.warn(`Cache is closed, ignoring get(${key})`); return undefined; } - return this.pgCache.get(key)?.pool; + const managedPool = this.getLiveRecord(key, true); + if (!managedPool) { + this.counters.misses++; + return undefined; + } + this.counters.hits++; + return managedPool.pool; } has(key: PgPoolKey): boolean { - return this.pgCache.has(key); + if (this.closed) return false; + return Boolean(this.getLiveRecord(key, false)); } + /** + * Legacy direct insertion. Prefer getOrCreate/acquire so capacity is checked + * before the caller constructs a pool. + */ set(key: PgPoolKey, pool: pg.Pool): void { - if (this.closed) throw new Error(`Cannot add to cache after it has been closed (key: ${key})`); - this.pgCache.set(key, new ManagedPgPool(pool, key)); + this.assertOpen(key); + const existing = this.records.get(key); + if (existing?.pool === pool) { + this.touch(existing); + return; + } + if (existing?.leaseCount) { + throw new Error(`Cannot replace leased pg pool identity ${key}`); + } + if (existing) this.removeRecord(existing, 'replace'); + + const reservation = this.reserveSlot(key); + this.commitReservation(reservation, pool, 0); } + /** Atomically capacity-check, synchronously construct, and cache an idle pool. */ + getOrCreate(key: PgPoolKey, factory: PoolFactory): pg.Pool { + this.assertOpen(key); + const existing = this.getLiveRecord(key, true); + if (existing) { + this.counters.hits++; + return existing.pool; + } + + this.counters.misses++; + return this.createWithReservation(key, factory, 0).pool; + } + + /** + * Atomically get/create and lease an exact identity. A leased identity cannot + * be selected by capacity or TTL eviction until every lease is released. + */ + acquire(key: PgPoolKey, factory: PoolFactory): PgPoolLease { + this.assertOpen(key); + let managedPool = this.getLiveRecord(key, true); + if (managedPool) { + this.counters.hits++; + managedPool.leaseCount++; + } else { + this.counters.misses++; + managedPool = this.createWithReservation(key, factory, 1); + } + this.counters.leasesAcquired++; + return this.makeLease(managedPool); + } + + /** Explicit deletion never interrupts a lease; callers may retry after release. */ delete(key: PgPoolKey): void { - const managedPool = this.pgCache.get(key); - const existed = this.pgCache.delete(key); - if (!existed && managedPool) { - this.notifyCleanup(key); - this.disposePool(managedPool); + const managedPool = this.records.get(key); + if (managedPool) { + if (managedPool.leaseCount === 0) + this.removeRecord(managedPool, 'delete'); + return; + } + for (const aliasedKey of [...(this.aliasKeys.get(key) ?? [])]) { + const aliasedPool = this.records.get(aliasedKey); + if (aliasedPool && aliasedPool.leaseCount === 0) { + this.removeRecord(aliasedPool, 'delete'); + } } } + /** Clear every currently unleased identity. */ clear(): void { - const entries = [...this.pgCache.entries()]; - this.pgCache.clear(); - for (const [key, managedPool] of entries) { - this.notifyCleanup(key); - this.disposePool(managedPool); + for (const managedPool of [...this.records.values()]) { + if (managedPool.leaseCount === 0) this.removeRecord(managedPool, 'clear'); } } async close(): Promise { - if (this.closed) return; + if (this.closePromise) return this.closePromise; this.closed = true; - this.clear(); - await this.waitForDisposals(); - // Re-open the cache so it can accept new entries if the process - // survives the shutdown signal (e.g. during provisioning or restart). - this.closed = false; + this.closePromise = (async () => { + try { + // Explicit process teardown is the only operation that may override leases. + for (const managedPool of [...this.records.values()]) { + this.removeRecord(managedPool, 'close'); + } + await this.waitForDisposals(); + } finally { + // Preserve the established restart/provisioning behavior. + this.closed = false; + this.closePromise = null; + } + })(); + return this.closePromise; } async waitForDisposals(): Promise { - if (this.cleanupTasks.length === 0) return; - const tasks = [...this.cleanupTasks]; - this.cleanupTasks = []; - await Promise.allSettled(tasks); + while (this.cleanupTasks.size > 0) { + await Promise.allSettled([...this.cleanupTasks]); + } + } + + getStats(): PgPoolCacheStats { + let leasedPools = 0; + let activeLeases = 0; + for (const managedPool of this.records.values()) { + if (managedPool.leaseCount > 0) leasedPools++; + activeLeases += managedPool.leaseCount; + } + return { + size: this.records.size, + max: this.config.max, + ttl: this.config.ttl, + leasedPools, + idlePools: this.records.size - leasedPools, + activeLeases, + reservations: this.reservations, + pendingDisposals: this.cleanupTasks.size, + ...this.counters, + }; + } + + private assertOpen(key: PgPoolKey): void { + if (this.closed) { + throw new Error( + `Cannot access pg cache while it is closed (key: ${key})` + ); + } + } + + private touch(managedPool: ManagedPgPool): void { + managedPool.touch(++this.accessOrder, Date.now(), this.config.ttl); + } + + private getLiveRecord( + key: PgPoolKey, + updateAge: boolean + ): ManagedPgPool | undefined { + const managedPool = this.records.get(key); + if (!managedPool) return undefined; + if (managedPool.leaseCount === 0 && managedPool.isExpired(Date.now())) { + this.removeRecord(managedPool, 'ttl'); + return undefined; + } + if (updateAge) this.touch(managedPool); + return managedPool; + } + + private idleRecordsByAge(): ManagedPgPool[] { + return [...this.records.values()] + .filter((managedPool) => managedPool.leaseCount === 0) + .sort((a, b) => a.lastAccessOrder - b.lastAccessOrder); + } + + private reserveSlot(key: PgPoolKey): SlotReservation { + if (this.reservedKeys.has(key)) { + throw new Error(`Re-entrant pg pool acquisition for identity ${key}`); + } + + const overflow = Math.max( + 0, + this.records.size + this.reservations + 1 - this.config.max + ); + const candidates = this.idleRecordsByAge(); + if (candidates.length < overflow) { + this.counters.capacityRefusals++; + throw new PgPoolCapacityError( + this.config.max, + this.records.size + this.reservations, + this.countLeasedPools() + ); + } + + const victims = candidates.slice(0, overflow); + for (const victim of victims) this.records.delete(victim.key); + this.reservations++; + this.reservedKeys.add(key); + return { key, victims }; + } + + private rollbackReservation(reservation: SlotReservation): void { + this.reservations = Math.max(0, this.reservations - 1); + this.reservedKeys.delete(reservation.key); + for (const victim of reservation.victims) { + this.records.set(victim.key, victim); + } + } + + private commitReservation( + reservation: SlotReservation, + pool: pg.Pool, + leaseCount: number + ): ManagedPgPool { + const managedPool = new ManagedPgPool(pool, reservation.key); + managedPool.leaseCount = leaseCount; + this.touch(managedPool); + this.records.set(reservation.key, managedPool); + this.reservations = Math.max(0, this.reservations - 1); + this.reservedKeys.delete(reservation.key); + this.counters.poolsCreated++; + + for (const victim of reservation.victims) { + this.counters.capacityEvictions++; + this.disposeRemovedRecord(victim); + } + return managedPool; + } + + private createWithReservation( + key: PgPoolKey, + factory: PoolFactory, + leaseCount: number + ): ManagedPgPool { + const reservation = this.reserveSlot(key); + let pool: pg.Pool; + try { + pool = factory(); + } catch (error) { + this.rollbackReservation(reservation); + throw error; + } + return this.commitReservation(reservation, pool, leaseCount); + } + + private makeLease(managedPool: ManagedPgPool): PgPoolLease { + let released = false; + return { + pool: managedPool.pool, + identity: managedPool.key, + release: () => { + if (released) return; + released = true; + this.counters.leasesReleased++; + managedPool.leaseCount = Math.max(0, managedPool.leaseCount - 1); + + // close() may already have detached this record. + if (this.records.get(managedPool.key) !== managedPool) return; + if (managedPool.leaseCount > 0) return; + if (managedPool.isExpired(Date.now())) { + this.removeRecord(managedPool, 'ttl'); + return; + } + this.enforceCapacity(); + }, + }; + } + + private enforceCapacity(): void { + while (this.records.size > this.config.max) { + const victim = this.idleRecordsByAge()[0]; + if (!victim) return; + this.removeRecord(victim, 'capacity'); + } + } + + private countLeasedPools(): number { + let leased = 0; + for (const managedPool of this.records.values()) { + if (managedPool.leaseCount > 0) leased++; + } + return leased; + } + + private removeRecord( + managedPool: ManagedPgPool, + reason: PgPoolDisposalReason + ): void { + if (this.records.get(managedPool.key) !== managedPool) return; + this.records.delete(managedPool.key); + if (reason === 'capacity') this.counters.capacityEvictions++; + if (reason === 'ttl') this.counters.ttlExpirations++; + this.disposeRemovedRecord(managedPool); + } + + private disposeRemovedRecord(managedPool: ManagedPgPool): void { + const cleanupKeys = [ + managedPool.key, + ...this.unregisterAliases(managedPool.key), + ]; + for (const cleanupKey of cleanupKeys) this.notifyCleanup(cleanupKey); + + // Alternate drivers may intentionally return one physical pool for multiple + // exact identities. Never end it while another retained identity owns it. + if ( + [...this.records.values()].some( + (entry) => entry.pool === managedPool.pool + ) + ) { + return; + } + if (managedPool.isDisposed) return; + + this.counters.disposalsStarted++; + let task: Promise; + task = managedPool + .dispose() + .then(() => { + this.counters.disposalsCompleted++; + }) + .catch((error) => { + this.counters.disposalFailures++; + log.error( + `Error ending pg.Pool ${managedPool.key}: ${(error as Error).message}` + ); + }) + .finally(() => this.cleanupTasks.delete(task)); + this.cleanupTasks.add(task); + } + + private unregisterAliases(key: PgPoolKey): string[] { + const aliases = [...(this.keyAliases.get(key) ?? [])]; + this.keyAliases.delete(key); + for (const alias of aliases) { + const keys = this.aliasKeys.get(alias); + keys?.delete(key); + if (keys?.size === 0) this.aliasKeys.delete(alias); + } + return aliases; } private notifyCleanup(pgPoolKey: string): void { @@ -162,17 +561,14 @@ export class PgPoolCacheManager { } }); } - - private disposePool(managedPool: ManagedPgPool): void { - if (managedPool.isDisposed) return; - const task = managedPool.dispose(); - this.cleanupTasks.push(task); - } } -// Create the singleton instance +// Process-wide registry. Its large capacity is only a key limit; pools and +// PostgreSQL connections remain lazily allocated on first use. export const pgCache = new PgPoolCacheManager(); +export const getPgCacheStats = (): PgPoolCacheStats => pgCache.getStats(); + // --- Graceful Shutdown --- const closePromise: { promise: Promise | null } = { promise: null }; @@ -185,7 +581,6 @@ export const close = async (verbose = false): Promise => { await pgCache.close(); if (verbose) log.success('PG cache disposed.'); } finally { - // Reset so close() can be called again if the process survives. closePromise.promise = null; } })(); @@ -193,13 +588,6 @@ export const close = async (verbose = false): Promise => { return closePromise.promise; }; -SYS_EVENTS.forEach(event => { - process.on(event, () => { - log.info(`Received ${event}`); - close(); - }); -}); - export const teardownPgPools = async (verbose = false): Promise => { return close(verbose); }; diff --git a/postgres/pg-cache/src/pg.ts b/postgres/pg-cache/src/pg.ts index 920a900ddd..67d1102dbd 100644 --- a/postgres/pg-cache/src/pg.ts +++ b/postgres/pg-cache/src/pg.ts @@ -1,50 +1,329 @@ +import { createHash, createHmac, randomBytes } from 'node:crypto'; + import { Logger } from '@pgpmjs/logger'; import { parseEnvNumber } from '12factor-env'; import pg from 'pg'; import { getPgEnvOptions, PgConfig, PgPoolConfig } from 'pg-env'; -import { getActivePgPoolFactory, PgPoolFactory } from './driver'; -import { pgCache } from './lru'; +import { + getActivePgPoolFactory, + getPgPoolDriverIdentity, + PgPoolFactory, +} from './driver'; +import { pgCache, type PgPoolLease } from './lru'; import { installCheckoutSanitizer } from './sanitizer'; const log = new Logger('pg-cache'); +export interface GetPgPoolOptions { + /** Separates pools used by different trust boundaries. */ + purpose?: string; +} + +const normalizePoolOptions = ( + options: GetPgPoolOptions = {} +): Required => { + const purpose = options.purpose ?? 'default'; + if (typeof purpose !== 'string' || purpose.length === 0) { + throw new TypeError('pg pool purpose must be a non-empty string'); + } + return { purpose }; +}; + +// Pool identities may appear in diagnostics and cache lifecycle logs. A plain +// digest over a known connection shape could act as an offline password +// verifier, so identities are keyed and intentionally process-local. +const pgIdentityHmacKey = randomBytes(32); + +const hmacIdentity = (prefix: string, identity: string): string => + `${prefix}:${createHmac('sha256', pgIdentityHmacKey) + .update(identity) + .digest('hex')}`; + +const requireIdentityString = (value: unknown, path: string): string => { + if (typeof value !== 'string') { + throw new TypeError(`${path} must be a string`); + } + return value; +}; + +const requireIdentityInteger = ( + value: unknown, + path: string, + minimum: number, + maximum = Number.MAX_SAFE_INTEGER +): number => { + if ( + typeof value !== 'number' || + !Number.isSafeInteger(value) || + value < minimum || + value > maximum + ) { + throw new TypeError( + `${path} must be a safe integer between ${minimum} and ${maximum}` + ); + } + return value; +}; + +const canonicalizeIdentityValue = ( + value: unknown, + path: string, + ancestors = new Set() +): unknown => { + if ( + value === null || + typeof value === 'string' || + typeof value === 'boolean' + ) { + return value; + } + if (typeof value === 'number') { + if (!Number.isFinite(value)) { + throw new TypeError(`${path} must not contain a non-finite number`); + } + return Object.is(value, -0) ? ['number', '-0'] : value; + } + if (Buffer.isBuffer(value)) { + return ['buffer-sha256', createHash('sha256').update(value).digest('hex')]; + } + if (Array.isArray(value)) { + if (ancestors.has(value)) throw new TypeError(`${path} must not be cyclic`); + const ownKeys = Reflect.ownKeys(value); + if ( + ownKeys.some((key) => typeof key !== 'string') || + ownKeys.some( + (key) => key !== 'length' && !/^(?:0|[1-9]\d*)$/.test(key as string) + ) || + value.some( + (_entry, index) => !Object.prototype.hasOwnProperty.call(value, index) + ) || + Object.keys(value).length !== value.length + ) { + throw new TypeError( + `${path} must be a dense array without custom properties` + ); + } + ancestors.add(value); + const result = value.map((entry, index) => { + if (entry === undefined) { + throw new TypeError(`${path}[${index}] must not be undefined`); + } + return canonicalizeIdentityValue(entry, `${path}[${index}]`, ancestors); + }); + ancestors.delete(value); + return ['array', result]; + } + if (typeof value === 'object') { + const record = value as Record; + const prototype = Object.getPrototypeOf(record); + if (prototype !== Object.prototype && prototype !== null) { + throw new TypeError(`${path} must contain only data values`); + } + if (ancestors.has(record)) + throw new TypeError(`${path} must not be cyclic`); + ancestors.add(record); + const result: Array<[string, unknown]> = []; + const ownKeys = Reflect.ownKeys(record); + if (ownKeys.some((key) => typeof key !== 'string')) { + throw new TypeError(`${path} must not contain symbol properties`); + } + for (const key of (ownKeys as string[]).sort()) { + const descriptor = Object.getOwnPropertyDescriptor(record, key); + if (!descriptor || !('value' in descriptor)) { + throw new TypeError(`${path}.${key} must be a data property`); + } + const entry = descriptor.value; + if (entry === undefined) { + throw new TypeError(`${path}.${key} must not be undefined`); + } + result.push([ + key, + canonicalizeIdentityValue(entry, `${path}.${key}`, ancestors), + ]); + } + ancestors.delete(record); + return ['object', result]; + } + throw new TypeError(`${path} must contain only deterministic data values`); +}; + export const buildConnectionString = ( user: string, password: string, host: string, port: string | number, database: string -): string => - `postgres://${user}:${password}@${host}:${port}/${database}`; +): string => { + const encodedHost = + host.includes(':') && !host.startsWith('[') + ? `[${host}]` + : encodeURIComponent(host); + return ( + `postgres://${encodeURIComponent(user)}:${encodeURIComponent(password)}` + + `@${encodedHost}:${port}/${encodeURIComponent(database)}` + ); +}; /** * Read per-pool configuration from environment variables. * * Supports: * - PG_POOL_MAX: Maximum clients per pool (default: 5) + * - PG_POOL_MAX_USES: Retire a client after this many checkouts (0/unset: unlimited) * - PG_POOL_IDLE_TIMEOUT_MS: Close idle clients after ms (default: 30000) * - PG_POOL_CONNECTION_TIMEOUT_MS: Fail connect() after ms (default: 5000) */ +const normalizeMaxUses = ( + value: number | string | undefined, + source: 'pool.maxUses' | 'PG_POOL_MAX_USES' +): number | undefined => { + if (value === undefined || value === '') return undefined; + if (typeof value !== 'number' && typeof value !== 'string') { + throw new TypeError(`${source} must be 0 or a positive safe integer`); + } + if (typeof value === 'string' && !/^(?:0|[1-9]\d*)$/.test(value)) { + throw new TypeError(`${source} must be 0 or a positive safe integer`); + } + const parsed = typeof value === 'number' ? value : Number(value); + if (!Number.isSafeInteger(parsed) || parsed < 0) { + throw new TypeError(`${source} must be 0 or a positive safe integer`); + } + return parsed === 0 ? undefined : parsed; +}; + export function getPgPoolConfig(overrides?: PgPoolConfig): pg.PoolConfig { - return { + const maxUses = + overrides?.maxUses !== undefined + ? normalizeMaxUses(overrides.maxUses, 'pool.maxUses') + : normalizeMaxUses(process.env.PG_POOL_MAX_USES, 'PG_POOL_MAX_USES'); + const pool = { max: overrides?.max ?? parseEnvNumber(process.env.PG_POOL_MAX) ?? 5, - idleTimeoutMillis: overrides?.idleTimeoutMillis ?? parseEnvNumber(process.env.PG_POOL_IDLE_TIMEOUT_MS) ?? 30000, - connectionTimeoutMillis: overrides?.connectionTimeoutMillis ?? parseEnvNumber(process.env.PG_POOL_CONNECTION_TIMEOUT_MS) ?? 5000, - ...(overrides?.allowExitOnIdle !== undefined && { allowExitOnIdle: overrides.allowExitOnIdle }), + ...(maxUses !== undefined && { maxUses }), + idleTimeoutMillis: + overrides?.idleTimeoutMillis ?? + parseEnvNumber(process.env.PG_POOL_IDLE_TIMEOUT_MS) ?? + 30000, + connectionTimeoutMillis: + overrides?.connectionTimeoutMillis ?? + parseEnvNumber(process.env.PG_POOL_CONNECTION_TIMEOUT_MS) ?? + 5000, + ...(overrides?.allowExitOnIdle !== undefined && { + allowExitOnIdle: overrides.allowExitOnIdle, + }), + }; + requireIdentityInteger(pool.max, 'pool.max', 1); + if (pool.maxUses !== undefined) { + requireIdentityInteger(pool.maxUses, 'pool.maxUses', 1); + } + requireIdentityInteger(pool.idleTimeoutMillis, 'pool.idleTimeoutMillis', 0); + requireIdentityInteger( + pool.connectionTimeoutMillis, + 'pool.connectionTimeoutMillis', + 0 + ); + if ( + pool.allowExitOnIdle !== undefined && + typeof pool.allowExitOnIdle !== 'boolean' + ) { + throw new TypeError('pool.allowExitOnIdle must be a boolean'); + } + return pool; +} + +const normalizeIdentityConfig = ( + pgConfig: Partial & { pool?: PgPoolConfig } +): { config: PgConfig; ssl: unknown } => { + const config = getPgEnvOptions(pgConfig); + requireIdentityString(config.host, 'pg.host'); + requireIdentityInteger(config.port, 'pg.port', 1, 65_535); + requireIdentityString(config.database, 'pg.database'); + requireIdentityString(config.user, 'pg.user'); + // node-postgres accepts password callbacks at runtime; captured credentials + // cannot be represented exactly, so the shared cache rejects them. + requireIdentityString(config.password, 'pg.password'); + return { + config, + ssl: canonicalizeIdentityValue(config.ssl ?? null, 'pg.ssl'), }; +}; + +/** Opaque identity for the complete connection and reuse contract. */ +export function getPgPoolIdentity( + pgConfig: Partial & { pool?: PgPoolConfig }, + options: GetPgPoolOptions = {} +): string { + const { config, ssl } = normalizeIdentityConfig(pgConfig); + const pool = getPgPoolConfig(pgConfig.pool); + const normalizedOptions = normalizePoolOptions(options); + const driver = requireIdentityString( + getPgPoolDriverIdentity(), + 'pg driver identity' + ); + const identity = JSON.stringify({ + version: 1, + driver, + host: config.host, + port: config.port, + database: config.database, + user: config.user, + password: config.password, + ssl, + pool: { + max: pool.max, + maxUses: pool.maxUses ?? null, + idleTimeoutMillis: pool.idleTimeoutMillis, + connectionTimeoutMillis: pool.connectionTimeoutMillis, + allowExitOnIdle: pool.allowExitOnIdle ?? false, + }, + purpose: normalizedOptions.purpose, + checkout: getActivePgPoolFactory() + ? 'registered-factory-owned-v1' + : 'discard-all-v1', + }); + return hmacIdentity('pg:v1', identity); +} + +/** Opaque identity for a physical database, excluding login and pool policy. */ +export function getPgDatabaseTargetIdentity( + pgConfig: Partial +): string { + const { config } = normalizeIdentityConfig(pgConfig); + const driver = requireIdentityString( + getPgPoolDriverIdentity(), + 'pg driver identity' + ); + const identity = JSON.stringify({ + version: 1, + driver, + host: config.host, + port: config.port, + database: config.database, + }); + return hmacIdentity('pg-target:v1', identity); } /** * Default pool factory: builds a real `pg.Pool` over TCP. This is the behavior * used whenever no alternate driver is registered (see `./driver`). */ -export const defaultPgPoolFactory: PgPoolFactory = (pgConfig): pg.Pool => { - const config = getPgEnvOptions(pgConfig); - const { user, password, host, port, database } = config; - const connectionString = buildConnectionString(user, password, host, port, database); +export const defaultPgPoolFactory: PgPoolFactory = ( + pgConfig, + options +): pg.Pool => { + const { config } = normalizeIdentityConfig(pgConfig); + normalizePoolOptions(options); + const { user, password, host, port, database, ssl } = config; const poolConfig = getPgPoolConfig(pgConfig.pool); - const pgPool = new pg.Pool({ connectionString, ...poolConfig }); + const pgPool = new pg.Pool({ + host, + port, + database, + user, + password, + ...(ssl !== undefined && { ssl }), + ...poolConfig, + }); /** * IMPORTANT: Pool-level error handler for idle connection errors. @@ -90,31 +369,73 @@ export const defaultPgPoolFactory: PgPoolFactory = (pgConfig): pg.Pool => { pgPool.on('error', (err: Error & { code?: string }) => { if (err.code === '57P01') { // Expected during database cleanup - log at debug level - log.debug(`Pool ${database} connection terminated (expected during cleanup): ${err.message}`); + log.debug( + `Pool ${database} connection terminated (expected during cleanup): ${err.message}` + ); } else { // Unexpected pool error - log at error level for visibility // Note: This does NOT swallow query errors - those still throw via Promise rejection - log.error(`Pool ${database} unexpected idle connection error [${err.code || 'unknown'}]: ${err.message}`); + log.error( + `Pool ${database} unexpected idle connection error [${err.code || 'unknown'}]: ${err.message}` + ); } }); return installCheckoutSanitizer(pgPool); }; -export const getPgPool = (pgConfig: Partial & { pool?: PgPoolConfig }): pg.Pool => { - const config = getPgEnvOptions(pgConfig); - const { database } = config; - if (pgCache.has(database)) { - const cached = pgCache.get(database); - if (cached) return cached; - } - +const createPgPool = ( + pgConfig: Partial & { pool?: PgPoolConfig }, + options: Required +): pg.Pool => { // Route through the registered driver (default = pg.Pool over TCP). A custom // factory may return any QueryablePool (e.g. an in-process PGlite pool); it is // treated as a pg.Pool since that is the only surface consumers use. const factory = getActivePgPoolFactory() ?? defaultPgPoolFactory; - const pgPool = factory(pgConfig) as pg.Pool; + return factory(pgConfig, options) as pg.Pool; +}; + +const getPgPoolWithOptions = ( + pgConfig: Partial & { pool?: PgPoolConfig }, + options: GetPgPoolOptions = {} +): pg.Pool => { + const normalizedOptions = normalizePoolOptions(options); + const identity = getPgPoolIdentity(pgConfig, normalizedOptions); + const pool = pgCache.getOrCreate(identity, () => + createPgPool(pgConfig, normalizedOptions) + ); + pgCache.registerAlias( + normalizeIdentityConfig(pgConfig).config.database, + identity + ); + return pool; +}; + +/** Compatibility API with an optional exact-purpose reuse boundary. */ +export function getPgPool( + pgConfig: Partial & { pool?: PgPoolConfig }, + options?: GetPgPoolOptions +): pg.Pool; +export function getPgPool( + pgConfig: Partial & { pool?: PgPoolConfig }, + options: GetPgPoolOptions = {} +): pg.Pool { + return getPgPoolWithOptions(pgConfig, options); +} - pgCache.set(database, pgPool); - return pgPool; +/** Acquire an idempotently releasable ownership claim over one exact pool. */ +export const acquirePgPool = ( + pgConfig: Partial & { pool?: PgPoolConfig }, + options: GetPgPoolOptions = {} +): PgPoolLease => { + const normalizedOptions = normalizePoolOptions(options); + const identity = getPgPoolIdentity(pgConfig, normalizedOptions); + const lease = pgCache.acquire(identity, () => + createPgPool(pgConfig, normalizedOptions) + ); + pgCache.registerAlias( + normalizeIdentityConfig(pgConfig).config.database, + identity + ); + return lease; }; diff --git a/postgres/pg-env/src/pg-config.ts b/postgres/pg-env/src/pg-config.ts index 7ed78ce5cd..673a89cde3 100644 --- a/postgres/pg-env/src/pg-config.ts +++ b/postgres/pg-env/src/pg-config.ts @@ -1,9 +1,36 @@ +import type { SecureVersion } from 'node:tls'; + +/** + * Serializable TLS options supported by the shared PostgreSQL connection + * contract. Keeping this surface data-only is intentional: pool identities + * must account for every TLS input, which callback and socket objects cannot + * do deterministically. + */ +export interface PgSslOptions { + ca?: string | Buffer | Array; + cert?: string | Buffer | Array; + key?: + | string + | Buffer + | Array; + passphrase?: string; + rejectUnauthorized?: boolean; + servername?: string; + minVersion?: SecureVersion; + maxVersion?: SecureVersion; + ciphers?: string; +} + +export type PgSslConfig = boolean | PgSslOptions; + export interface PgConfig { host: string; port: number; user: string; password: string; database: string; + /** TLS settings passed directly to node-postgres. */ + ssl?: PgSslConfig; } /** @@ -15,6 +42,8 @@ export interface PgConfig { export interface PgPoolConfig { /** Maximum number of clients in the pool (env: PG_POOL_MAX, default: 5) */ max?: number; + /** Retire a client after this many checkouts (env: PG_POOL_MAX_USES, 0/unset: unlimited) */ + maxUses?: number; /** Close idle clients after this many ms (env: PG_POOL_IDLE_TIMEOUT_MS, default: 30000) */ idleTimeoutMillis?: number; /** Reject pool.connect() after this many ms (env: PG_POOL_CONNECTION_TIMEOUT_MS, default: 5000) */ @@ -29,4 +58,4 @@ export const defaultPgConfig: PgConfig = { user: 'postgres', password: 'password', database: 'postgres' -}; \ No newline at end of file +};