From 43303496f586c09e260cbde7e8fd608f3fedab23 Mon Sep 17 00:00:00 2001 From: zetazzz Date: Mon, 17 Aug 2026 16:55:13 +0800 Subject: [PATCH 1/4] Add notification listener role attestation --- .../src/__tests__/notification-role.test.ts | 309 +++++++++++++ postgres/pg-cache/src/index.ts | 19 + postgres/pg-cache/src/notification-role.ts | 436 ++++++++++++++++++ 3 files changed, 764 insertions(+) create mode 100644 postgres/pg-cache/src/__tests__/notification-role.test.ts create mode 100644 postgres/pg-cache/src/notification-role.ts diff --git a/postgres/pg-cache/src/__tests__/notification-role.test.ts b/postgres/pg-cache/src/__tests__/notification-role.test.ts new file mode 100644 index 0000000000..27541defd6 --- /dev/null +++ b/postgres/pg-cache/src/__tests__/notification-role.test.ts @@ -0,0 +1,309 @@ +import type { Pool } from 'pg'; + +import { + assertPgNotificationRole, + assertPgNotificationRoleClient, + auditPgNotificationRole, + auditPgNotificationRoleClient, + normalizePgNotificationRoleContracts, + PG_NOTIFICATION_ROLE_AUDIT_SQL, + PG_NOTIFICATION_ROLE_AUDIT_VERSION, + type PgNotificationRoleClient, + PgNotificationRoleContractError, + type PgNotificationRoleViolationCode, + UnsafePgNotificationRoleError, +} from '../notification-role'; + +const contract = { + role: 'tenant_001_notification', + database: 'tenant_001', +}; + +const safeRow = { + expected_role: contract.role, + session_role: contract.role, + active_role: contract.role, + active_database: contract.database, + rolcanlogin: true, + rolinherit: false, + rolsuper: false, + rolbypassrls: false, + rolcreaterole: false, + rolcreatedb: false, + rolreplication: false, + membership_count: 0, + target_database_exists: true, + target_connect: true, + other_database_connect_count: 0, + target_database_owner: false, + target_database_create: false, + target_database_temp: false, + schema_owner_count: 0, + schema_create_count: 0, + schema_usage_count: 0, + relation_privilege_count: 0, + function_privilege_count: 0, + sequence_privilege_count: 0, +}; + +const poolWithRow = (row: Record | undefined) => { + const client = { + query: jest.fn(async (query: string) => + query === PG_NOTIFICATION_ROLE_AUDIT_SQL + ? { rows: row ? [row] : [] } + : { rows: [] } + ), + release: jest.fn(), + }; + return { + pool: { connect: jest.fn(async () => client) } as unknown as Pool, + client, + }; +}; + +describe('PostgreSQL notification-role audit', () => { + it('returns a frozen credential-free attestation for an exact safe login', async () => { + const { pool, client } = poolWithRow(safeRow); + const audit = await assertPgNotificationRole(pool, { + ...contract, + password: 'must-not-escape', + } as typeof contract); + + expect(audit).toEqual({ + version: PG_NOTIFICATION_ROLE_AUDIT_VERSION, + ...contract, + safe: true, + violations: [], + }); + expect(Object.isFrozen(audit)).toBe(true); + expect(Object.isFrozen(audit.violations)).toBe(true); + expect(JSON.stringify(audit)).not.toContain('must-not-escape'); + expect(client.query).toHaveBeenNthCalledWith(1, 'BEGIN READ ONLY'); + expect(client.query).toHaveBeenNthCalledWith(2, 'SET LOCAL jit TO off'); + expect(client.query).toHaveBeenNthCalledWith( + 3, + PG_NOTIFICATION_ROLE_AUDIT_SQL, + [contract.role, contract.database] + ); + expect(client.query).toHaveBeenNthCalledWith(4, 'COMMIT'); + expect(client.release).toHaveBeenCalledWith(false); + }); + + it('audits an already-owned listener client without releasing it', async () => { + const { client } = poolWithRow(safeRow); + + await expect( + assertPgNotificationRoleClient( + client as unknown as PgNotificationRoleClient, + contract + ) + ).resolves.toMatchObject({ ...contract, safe: true }); + expect(client.release).not.toHaveBeenCalled(); + }); + + it('rolls back a failed pinned-client audit without taking ownership of release', async () => { + const failure = new Error('catalog unavailable'); + const client = { + query: jest + .fn() + .mockResolvedValueOnce({ rows: [] }) + .mockResolvedValueOnce({ rows: [] }) + .mockRejectedValueOnce(failure) + .mockResolvedValueOnce({ rows: [] }), + release: jest.fn(), + }; + + await expect(auditPgNotificationRoleClient(client, contract)).rejects.toBe( + failure + ); + expect(client.query).toHaveBeenNthCalledWith(4, 'ROLLBACK'); + expect(client.release).not.toHaveBeenCalled(); + }); + + it('reports both audit and rollback failures without releasing the pinned client', async () => { + const auditFailure = new Error('catalog unavailable'); + const rollbackFailure = new Error('rollback unavailable'); + const client = { + query: jest + .fn() + .mockResolvedValueOnce({ rows: [] }) + .mockResolvedValueOnce({ rows: [] }) + .mockRejectedValueOnce(auditFailure) + .mockRejectedValueOnce(rollbackFailure), + release: jest.fn(), + }; + + const rejected = auditPgNotificationRoleClient(client, contract); + await expect(rejected).rejects.toBeInstanceOf(AggregateError); + await expect(rejected).rejects.toMatchObject({ + cause: auditFailure, + errors: [auditFailure, rollbackFailure], + }); + expect(client.query).toHaveBeenNthCalledWith(4, 'ROLLBACK'); + expect(client.release).not.toHaveBeenCalled(); + }); + + it.each<[keyof typeof safeRow, unknown, PgNotificationRoleViolationCode]>([ + ['session_role', 'different_login', 'LOGIN_ROLE_MISMATCH'], + ['active_role', 'set_role_target', 'CURRENT_ROLE_MISMATCH'], + ['active_database', 'different_database', 'DATABASE_MISMATCH'], + ['rolcanlogin', false, 'LOGIN_REQUIRED'], + ['rolinherit', true, 'NOINHERIT_REQUIRED'], + ['rolsuper', true, 'SUPERUSER'], + ['rolbypassrls', true, 'BYPASSRLS'], + ['rolcreaterole', true, 'CREATEROLE'], + ['rolcreatedb', true, 'CREATEDB'], + ['rolreplication', true, 'REPLICATION'], + ['membership_count', 1, 'ROLE_MEMBERSHIP'], + ['target_database_exists', false, 'TARGET_DATABASE_MISSING'], + ['target_connect', false, 'TARGET_CONNECT_REQUIRED'], + ['other_database_connect_count', 1, 'CROSS_DATABASE_CONNECT'], + ['target_database_owner', true, 'DATABASE_OWNER'], + ['target_database_create', true, 'DATABASE_CREATE'], + ['target_database_temp', true, 'DATABASE_TEMP'], + ['schema_owner_count', 1, 'SCHEMA_OWNER'], + ['schema_create_count', 1, 'SCHEMA_CREATE'], + ['schema_usage_count', 1, 'SCHEMA_USAGE'], + ['relation_privilege_count', 1, 'RELATION_PRIVILEGE'], + ['function_privilege_count', 1, 'FUNCTION_PRIVILEGE'], + ['sequence_privilege_count', 1, 'SEQUENCE_PRIVILEGE'], + ])( + 'maps %s to its stable violation code', + async (field, unsafeValue, code) => { + const { pool } = poolWithRow({ ...safeRow, [field]: unsafeValue }); + const audit = await auditPgNotificationRole(pool, contract); + + expect(audit.safe).toBe(false); + expect(audit.violations).toContain(code); + await expect( + assertPgNotificationRole( + poolWithRow({ ...safeRow, [field]: unsafeValue }).pool, + contract + ) + ).rejects.toMatchObject({ + code: 'PG_NOTIFICATION_ROLE_UNSAFE', + audit: expect.objectContaining({ + violations: expect.arrayContaining([code]), + }), + }); + } + ); + + it('fails closed when the catalog audit returns no role row', async () => { + const { pool } = poolWithRow(undefined); + const audit = await auditPgNotificationRole(pool, contract); + + expect(audit).toMatchObject({ + safe: false, + violations: ['AUDIT_NO_RESULT'], + }); + await expect( + assertPgNotificationRole(poolWithRow(undefined).pool, contract) + ).rejects.toBeInstanceOf(UnsafePgNotificationRoleError); + }); + + it('rolls back and destroys the client when the catalog query fails', async () => { + const failure = new Error('catalog unavailable'); + const client = { + query: jest + .fn() + .mockResolvedValueOnce({ rows: [] }) + .mockResolvedValueOnce({ rows: [] }) + .mockRejectedValueOnce(failure) + .mockResolvedValueOnce({ rows: [] }), + release: jest.fn(), + }; + const pool = { connect: jest.fn(async () => client) } as unknown as Pool; + + await expect(auditPgNotificationRole(pool, contract)).rejects.toBe(failure); + expect(client.query).toHaveBeenNthCalledWith(4, 'ROLLBACK'); + expect(client.release).toHaveBeenCalledWith(true); + }); + + it('audits exact database scope, membership edges, and every prohibited ACL class', () => { + expect(PG_NOTIFICATION_ROLE_AUDIT_SQL).toContain( + 'membership.member = r.oid OR membership.roleid = r.oid' + ); + expect(PG_NOTIFICATION_ROLE_AUDIT_SQL).toContain( + 'database_record.datname <> $2::text' + ); + expect(PG_NOTIFICATION_ROLE_AUDIT_SQL).toContain("'CONNECT'"); + expect(PG_NOTIFICATION_ROLE_AUDIT_SQL).toContain("'CREATE'"); + expect(PG_NOTIFICATION_ROLE_AUDIT_SQL).toContain("'TEMP'"); + expect(PG_NOTIFICATION_ROLE_AUDIT_SQL).toContain( + 'schema_record.nspowner = r.oid' + ); + expect(PG_NOTIFICATION_ROLE_AUDIT_SQL).toContain("'USAGE'"); + expect(PG_NOTIFICATION_ROLE_AUDIT_SQL).toContain( + 'pg_catalog.has_table_privilege' + ); + expect(PG_NOTIFICATION_ROLE_AUDIT_SQL).toContain( + 'pg_catalog.has_any_column_privilege' + ); + expect(PG_NOTIFICATION_ROLE_AUDIT_SQL).toContain( + 'pg_catalog.has_function_privilege' + ); + expect(PG_NOTIFICATION_ROLE_AUDIT_SQL).toContain( + 'pg_catalog.has_sequence_privilege' + ); + expect(PG_NOTIFICATION_ROLE_AUDIT_SQL).toContain("n.nspname !~ '^pg_'"); + }); +}); + +describe('notification-role fleet contract', () => { + it('collapses exact generation duplicates and returns a deterministic frozen mapping', () => { + const normalized = normalizePgNotificationRoleContracts([ + { role: 'notify_b', database: 'tenant_b' }, + { ...contract }, + { ...contract }, + ]); + + expect(normalized).toEqual([ + contract, + { role: 'notify_b', database: 'tenant_b' }, + ]); + expect(Object.isFrozen(normalized)).toBe(true); + expect(normalized.every(Object.isFrozen)).toBe(true); + }); + + it('rejects multiple logins for one database and one login spanning databases', () => { + expect(() => + normalizePgNotificationRoleContracts([ + contract, + { role: 'another_notification', database: contract.database }, + ]) + ).toThrow('maps to multiple login roles'); + expect(() => + normalizePgNotificationRoleContracts([ + contract, + { role: contract.role, database: 'tenant_002' }, + ]) + ).toThrow('maps to multiple databases'); + }); + + const malformedContracts: Array<{ + contracts: readonly { role: string; database: string }[]; + }> = [ + { contracts: [] }, + { contracts: [{ role: '', database: 'tenant_001' }] }, + { contracts: [{ role: 'notify', database: '' }] }, + { contracts: [{ role: 'n'.repeat(64), database: 'tenant_001' }] }, + { + contracts: [ + { + role: 'notify', + database: `bad${String.fromCharCode(0xd800)}`, + }, + ], + }, + ]; + + it.each(malformedContracts)( + 'rejects malformed contract input', + ({ contracts }) => { + expect(() => normalizePgNotificationRoleContracts(contracts)).toThrow( + PgNotificationRoleContractError + ); + } + ); +}); diff --git a/postgres/pg-cache/src/index.ts b/postgres/pg-cache/src/index.ts index 369a4c039d..77bbc13908 100644 --- a/postgres/pg-cache/src/index.ts +++ b/postgres/pg-cache/src/index.ts @@ -18,6 +18,19 @@ export { PgPoolCapacityError, teardownPgPools } from './lru'; +export { + assertPgNotificationRole, + assertPgNotificationRoleClient, + auditPgNotificationRole, + auditPgNotificationRoleClient, + normalizePgNotificationRoleContracts, + PG_NOTIFICATION_ROLE_AUDIT_SQL, + PG_NOTIFICATION_ROLE_AUDIT_VERSION, + PG_NOTIFICATION_ROLE_CONTRACT_ERROR_CODE, + PG_NOTIFICATION_ROLE_UNSAFE_ERROR_CODE, + PgNotificationRoleContractError, + UnsafePgNotificationRoleError, +} from './notification-role'; export { acquirePgPool, buildConnectionString, @@ -36,4 +49,10 @@ export type { PgPoolLease, PoolCleanupCallback, } from './lru'; +export type { + PgNotificationRoleAudit, + PgNotificationRoleClient, + PgNotificationRoleContract, + PgNotificationRoleViolationCode, +} from './notification-role'; export type { GetPgPoolOptions } from './pg'; diff --git a/postgres/pg-cache/src/notification-role.ts b/postgres/pg-cache/src/notification-role.ts new file mode 100644 index 0000000000..5468c20c80 --- /dev/null +++ b/postgres/pg-cache/src/notification-role.ts @@ -0,0 +1,436 @@ +import type { Pool, PoolClient, QueryResult } from 'pg'; + +export const PG_NOTIFICATION_ROLE_AUDIT_VERSION = 'pg-notification-role:v1'; +export const PG_NOTIFICATION_ROLE_UNSAFE_ERROR_CODE = + 'PG_NOTIFICATION_ROLE_UNSAFE'; +export const PG_NOTIFICATION_ROLE_CONTRACT_ERROR_CODE = + 'PG_NOTIFICATION_ROLE_CONTRACT_INVALID'; + +export type PgNotificationRoleViolationCode = + | 'LOGIN_ROLE_MISMATCH' + | 'CURRENT_ROLE_MISMATCH' + | 'DATABASE_MISMATCH' + | 'LOGIN_REQUIRED' + | 'NOINHERIT_REQUIRED' + | 'SUPERUSER' + | 'BYPASSRLS' + | 'CREATEROLE' + | 'CREATEDB' + | 'REPLICATION' + | 'ROLE_MEMBERSHIP' + | 'TARGET_DATABASE_MISSING' + | 'TARGET_CONNECT_REQUIRED' + | 'CROSS_DATABASE_CONNECT' + | 'DATABASE_OWNER' + | 'DATABASE_CREATE' + | 'DATABASE_TEMP' + | 'SCHEMA_OWNER' + | 'SCHEMA_CREATE' + | 'SCHEMA_USAGE' + | 'RELATION_PRIVILEGE' + | 'FUNCTION_PRIVILEGE' + | 'SEQUENCE_PRIVILEGE' + | 'AUDIT_NO_RESULT'; + +/** Credential-free identity expected from one dedicated listener login. */ +export interface PgNotificationRoleContract { + role: string; + database: string; +} + +/** Safe to persist in diagnostics: connection secrets/config are never copied. */ +export interface PgNotificationRoleAudit { + version: typeof PG_NOTIFICATION_ROLE_AUDIT_VERSION; + role: string; + database: string; + safe: boolean; + violations: readonly PgNotificationRoleViolationCode[]; +} + +/** Catalog-query capability used by the broker's pinned LISTEN client. */ +export type PgNotificationRoleClient = Pick; + +interface PgNotificationRoleAuditRow { + expected_role: string; + session_role: string; + active_role: string; + active_database: string; + rolcanlogin: boolean; + rolinherit: boolean; + rolsuper: boolean; + rolbypassrls: boolean; + rolcreaterole: boolean; + rolcreatedb: boolean; + rolreplication: boolean; + membership_count: number; + target_database_exists: boolean; + target_connect: boolean; + other_database_connect_count: number; + target_database_owner: boolean; + target_database_create: boolean; + target_database_temp: boolean; + schema_owner_count: number; + schema_create_count: number; + schema_usage_count: number; + relation_privilege_count: number; + function_privilege_count: number; + sequence_privilege_count: number; +} + +/** + * Audit only the session login's effective privileges. PostgreSQL system + * schemas/objects are excluded because ordinary logins necessarily use the + * catalog; every non-system schema and object remains in scope. + */ +export const PG_NOTIFICATION_ROLE_AUDIT_SQL = ` +WITH login_role AS MATERIALIZED ( + SELECT r.oid, r.rolname, r.rolcanlogin, r.rolinherit, r.rolsuper, + r.rolbypassrls, r.rolcreaterole, r.rolcreatedb, r.rolreplication + FROM pg_catalog.pg_roles r + WHERE r.rolname = session_user +), target_database AS MATERIALIZED ( + SELECT d.oid, d.datname, d.datdba + FROM pg_catalog.pg_database d + WHERE d.datname = $2::text +), application_schemas AS MATERIALIZED ( + SELECT n.oid, n.nspname, n.nspowner + FROM pg_catalog.pg_namespace n + WHERE n.nspname <> 'information_schema' + AND n.nspname !~ '^pg_' +) +SELECT $1::text AS expected_role, + session_user AS session_role, + current_user AS active_role, + pg_catalog.current_database() AS active_database, + r.rolcanlogin, + r.rolinherit, + r.rolsuper, + r.rolbypassrls, + r.rolcreaterole, + r.rolcreatedb, + r.rolreplication, + ( + SELECT count(*)::int + FROM pg_catalog.pg_auth_members membership + WHERE membership.member = r.oid OR membership.roleid = r.oid + ) AS membership_count, + (target.oid IS NOT NULL) AS target_database_exists, + COALESCE( + pg_catalog.has_database_privilege(r.rolname, target.oid, 'CONNECT'), + false + ) AS target_connect, + ( + SELECT count(*)::int + FROM pg_catalog.pg_database database_record + WHERE database_record.datname <> $2::text + AND pg_catalog.has_database_privilege( + r.rolname, + database_record.oid, + 'CONNECT' + ) + ) AS other_database_connect_count, + COALESCE(target.datdba = r.oid, false) AS target_database_owner, + COALESCE( + pg_catalog.has_database_privilege(r.rolname, target.oid, 'CREATE'), + false + ) AS target_database_create, + COALESCE( + pg_catalog.has_database_privilege(r.rolname, target.oid, 'TEMP'), + false + ) AS target_database_temp, + ( + SELECT count(*)::int + FROM application_schemas schema_record + WHERE schema_record.nspowner = r.oid + ) AS schema_owner_count, + ( + SELECT count(*)::int + FROM application_schemas schema_record + WHERE pg_catalog.has_schema_privilege( + r.rolname, + schema_record.oid, + 'CREATE' + ) + ) AS schema_create_count, + ( + SELECT count(*)::int + FROM application_schemas schema_record + WHERE pg_catalog.has_schema_privilege( + r.rolname, + schema_record.oid, + 'USAGE' + ) + ) AS schema_usage_count, + ( + SELECT count(*)::int + FROM application_schemas schema_record + INNER JOIN pg_catalog.pg_class relation + ON relation.relnamespace = schema_record.oid + WHERE CASE WHEN relation.relkind IN ('r', 'p', 'v', 'm', 'f') + THEN pg_catalog.has_table_privilege( + r.rolname, + relation.oid, + 'SELECT,INSERT,UPDATE,DELETE,TRUNCATE,REFERENCES,TRIGGER' + ) + OR pg_catalog.has_any_column_privilege( + r.rolname, + relation.oid, + 'SELECT,INSERT,UPDATE,REFERENCES' + ) + ELSE false + END + ) AS relation_privilege_count, + ( + SELECT count(*)::int + FROM application_schemas schema_record + INNER JOIN pg_catalog.pg_proc routine + ON routine.pronamespace = schema_record.oid + WHERE pg_catalog.has_function_privilege( + r.rolname, + routine.oid, + 'EXECUTE' + ) + ) AS function_privilege_count, + ( + SELECT count(*)::int + FROM application_schemas schema_record + INNER JOIN pg_catalog.pg_class sequence_record + ON sequence_record.relnamespace = schema_record.oid + WHERE CASE WHEN sequence_record.relkind = 'S' + THEN pg_catalog.has_sequence_privilege( + r.rolname, + sequence_record.oid, + 'USAGE,SELECT,UPDATE' + ) + ELSE false + END + ) AS sequence_privilege_count +FROM login_role r +LEFT JOIN target_database target ON true +`; + +const containsUnpairedSurrogate = (value: string): boolean => { + for (let index = 0; index < value.length; index++) { + const code = value.charCodeAt(index); + if (code >= 0xd800 && code <= 0xdbff) { + const next = value.charCodeAt(index + 1); + if (!(next >= 0xdc00 && next <= 0xdfff)) return true; + index++; + } else if (code >= 0xdc00 && code <= 0xdfff) { + return true; + } + } + return false; +}; + +const assertIdentifier = ( + kind: 'role' | 'database', + value: unknown +): string => { + if (typeof value !== 'string' || value.length === 0) { + throw new PgNotificationRoleContractError( + `${kind} must be a non-empty string` + ); + } + if (value.includes('\0') || containsUnpairedSurrogate(value)) { + throw new PgNotificationRoleContractError( + `${kind} is not a valid PostgreSQL name` + ); + } + const bytes = Buffer.byteLength(value, 'utf8'); + if (bytes > 63) { + throw new PgNotificationRoleContractError( + `${kind} is ${bytes} UTF-8 bytes; PostgreSQL allows at most 63` + ); + } + return value; +}; + +const normalizeContract = ( + contract: PgNotificationRoleContract +): Readonly => + Object.freeze({ + role: assertIdentifier('role', contract?.role), + database: assertIdentifier('database', contract?.database), + }); + +export class PgNotificationRoleContractError extends Error { + readonly code = PG_NOTIFICATION_ROLE_CONTRACT_ERROR_CODE; + + constructor(reason: string) { + super(`Invalid PostgreSQL notification-role contract: ${reason}`); + this.name = 'PgNotificationRoleContractError'; + } +} + +export class UnsafePgNotificationRoleError extends Error { + readonly code = PG_NOTIFICATION_ROLE_UNSAFE_ERROR_CODE; + + constructor(readonly audit: PgNotificationRoleAudit) { + super( + `PostgreSQL notification role ${JSON.stringify(audit.role)} for database ` + + `${JSON.stringify(audit.database)} is unsafe: ${audit.violations.join(',')}` + ); + this.name = 'UnsafePgNotificationRoleError'; + } +} + +/** + * Enforce a one-to-one role/database mapping without accepting connection + * config. Exact duplicate pairs are collapsed for multi-generation reuse. + */ +export const normalizePgNotificationRoleContracts = ( + contracts: readonly PgNotificationRoleContract[] +): readonly Readonly[] => { + if (!Array.isArray(contracts) || contracts.length === 0) { + throw new PgNotificationRoleContractError( + 'at least one role/database pair is required' + ); + } + const byDatabase = new Map(); + const byRole = new Map(); + const unique = new Map>(); + for (const candidate of contracts) { + const contract = normalizeContract(candidate); + const databaseRole = byDatabase.get(contract.database); + if (databaseRole && databaseRole !== contract.role) { + throw new PgNotificationRoleContractError( + `database ${JSON.stringify(contract.database)} maps to multiple login roles` + ); + } + const roleDatabase = byRole.get(contract.role); + if (roleDatabase && roleDatabase !== contract.database) { + throw new PgNotificationRoleContractError( + `login role ${JSON.stringify(contract.role)} maps to multiple databases` + ); + } + byDatabase.set(contract.database, contract.role); + byRole.set(contract.role, contract.database); + unique.set(`${contract.database}\0${contract.role}`, contract); + } + return Object.freeze( + [...unique.values()].sort((left, right) => { + if (left.database !== right.database) { + return left.database < right.database ? -1 : 1; + } + if (left.role === right.role) return 0; + return left.role < right.role ? -1 : 1; + }) + ); +}; + +const violationCodes = ( + row: PgNotificationRoleAuditRow | undefined, + contract: Readonly +): PgNotificationRoleViolationCode[] => { + if (!row) return ['AUDIT_NO_RESULT']; + const violations: PgNotificationRoleViolationCode[] = []; + if (row.session_role !== contract.role) + violations.push('LOGIN_ROLE_MISMATCH'); + if (row.active_role !== row.session_role) + violations.push('CURRENT_ROLE_MISMATCH'); + if (row.active_database !== contract.database) + violations.push('DATABASE_MISMATCH'); + if (!row.rolcanlogin) violations.push('LOGIN_REQUIRED'); + if (row.rolinherit) violations.push('NOINHERIT_REQUIRED'); + if (row.rolsuper) violations.push('SUPERUSER'); + if (row.rolbypassrls) violations.push('BYPASSRLS'); + if (row.rolcreaterole) violations.push('CREATEROLE'); + if (row.rolcreatedb) violations.push('CREATEDB'); + if (row.rolreplication) violations.push('REPLICATION'); + if (row.membership_count > 0) violations.push('ROLE_MEMBERSHIP'); + if (!row.target_database_exists) violations.push('TARGET_DATABASE_MISSING'); + if (!row.target_connect) violations.push('TARGET_CONNECT_REQUIRED'); + if (row.other_database_connect_count > 0) + violations.push('CROSS_DATABASE_CONNECT'); + if (row.target_database_owner) violations.push('DATABASE_OWNER'); + if (row.target_database_create) violations.push('DATABASE_CREATE'); + if (row.target_database_temp) violations.push('DATABASE_TEMP'); + if (row.schema_owner_count > 0) violations.push('SCHEMA_OWNER'); + if (row.schema_create_count > 0) violations.push('SCHEMA_CREATE'); + if (row.schema_usage_count > 0) violations.push('SCHEMA_USAGE'); + if (row.relation_privilege_count > 0) violations.push('RELATION_PRIVILEGE'); + if (row.function_privilege_count > 0) violations.push('FUNCTION_PRIVILEGE'); + if (row.sequence_privilege_count > 0) violations.push('SEQUENCE_PRIVILEGE'); + return violations; +}; + +/** Execute one fresh audit on an already-owned client without releasing it. */ +export const auditPgNotificationRoleClient = async ( + client: PgNotificationRoleClient, + candidate: PgNotificationRoleContract +): Promise => { + const contract = normalizeContract(candidate); + let inTransaction = false; + let result: QueryResult; + try { + await client.query('BEGIN READ ONLY'); + inTransaction = true; + await client.query('SET LOCAL jit TO off'); + result = await client.query( + PG_NOTIFICATION_ROLE_AUDIT_SQL, + [contract.role, contract.database] + ); + await client.query('COMMIT'); + inTransaction = false; + } catch (error) { + if (inTransaction) { + try { + await client.query('ROLLBACK'); + } catch (rollbackError) { + throw new AggregateError( + [error, rollbackError], + 'PostgreSQL notification role audit and rollback both failed', + { cause: error } + ); + } + } + throw error; + } + + const violations = Object.freeze(violationCodes(result.rows[0], contract)); + return Object.freeze({ + version: PG_NOTIFICATION_ROLE_AUDIT_VERSION, + role: contract.role, + database: contract.database, + safe: violations.length === 0, + violations, + }); +}; + +/** Execute one fresh, read-only catalog audit. Successful results are not cached. */ +export const auditPgNotificationRole = async ( + pool: Pool, + candidate: PgNotificationRoleContract +): Promise => { + const client: PoolClient = await pool.connect(); + let destroyClient = false; + try { + return await auditPgNotificationRoleClient(client, candidate); + } catch (error) { + destroyClient = true; + throw error; + } finally { + client.release(destroyClient); + } +}; + +/** Fail closed on a pinned client without exposing general query access. */ +export const assertPgNotificationRoleClient = async ( + client: PgNotificationRoleClient, + contract: PgNotificationRoleContract +): Promise => { + const audit = await auditPgNotificationRoleClient(client, contract); + if (!audit.safe) throw new UnsafePgNotificationRoleError(audit); + return audit; +}; + +/** Fail closed with a stable code while retaining a credential-free audit. */ +export const assertPgNotificationRole = async ( + pool: Pool, + contract: PgNotificationRoleContract +): Promise => { + const audit = await auditPgNotificationRole(pool, contract); + if (!audit.safe) throw new UnsafePgNotificationRoleError(audit); + return audit; +}; From 5a78d6fcef676e862b20ee4fef95738a566e4258 Mon Sep 17 00:00:00 2001 From: zetazzz Date: Mon, 17 Aug 2026 16:55:30 +0800 Subject: [PATCH 2/4] Add the PostgreSQL notification broker lifecycle --- .../src/__tests__/notification-broker.test.ts | 970 ++++++++++++++ postgres/pg-cache/src/index.ts | 29 + postgres/pg-cache/src/notification-broker.ts | 1178 +++++++++++++++++ 3 files changed, 2177 insertions(+) create mode 100644 postgres/pg-cache/src/__tests__/notification-broker.test.ts create mode 100644 postgres/pg-cache/src/notification-broker.ts diff --git a/postgres/pg-cache/src/__tests__/notification-broker.test.ts b/postgres/pg-cache/src/__tests__/notification-broker.test.ts new file mode 100644 index 0000000000..3d26f5d962 --- /dev/null +++ b/postgres/pg-cache/src/__tests__/notification-broker.test.ts @@ -0,0 +1,970 @@ +import { EventEmitter } from 'node:events'; + +import { + DEFAULT_PG_NOTIFICATION_OPERATION_TIMEOUT_MS, + getPgNotificationBrokerIdentity, + getPgNotificationDatabaseIdentity, + PgNotificationBrokerFailedError, + PgNotificationBrokerRegistry, + PgNotificationConnectionSource, + PgNotificationOperationTimeoutError, + PgNotificationQueueOverflowError, + PgNotificationTopicError, +} from '../notification-broker'; +import { + PG_NOTIFICATION_ROLE_AUDIT_SQL, + UnsafePgNotificationRoleError, +} from '../notification-role'; + +const roleContract = { + role: 'tenant_a_notify', + database: 'tenant_a', +}; + +const safeRoleAuditRow = { + expected_role: roleContract.role, + session_role: roleContract.role, + active_role: roleContract.role, + active_database: roleContract.database, + rolcanlogin: true, + rolinherit: false, + rolsuper: false, + rolbypassrls: false, + rolcreaterole: false, + rolcreatedb: false, + rolreplication: false, + membership_count: 0, + target_database_exists: true, + target_connect: true, + other_database_connect_count: 0, + target_database_owner: false, + target_database_create: false, + target_database_temp: false, + schema_owner_count: 0, + schema_create_count: 0, + schema_usage_count: 0, + relation_privilege_count: 0, + function_privilege_count: 0, + sequence_privilege_count: 0, +}; + +class MockNotificationClient extends EventEmitter { + readonly queries: string[] = []; + roleAuditRow: Record | undefined = safeRoleAuditRow; + readonly query = jest.fn( + async (text: string, _values?: readonly unknown[]): Promise => { + this.queries.push(text); + if (text === PG_NOTIFICATION_ROLE_AUDIT_SQL) { + return { rows: this.roleAuditRow ? [this.roleAuditRow] : [] }; + } + return { rows: [] }; + } + ); + readonly release = jest.fn( + async (_error?: Error | boolean): Promise => {} + ); + + notification(channel: string, payload?: string): void { + this.emit('notification', { channel, payload }); + } +} + +const createSource = (client = new MockNotificationClient()) => { + const source: PgNotificationConnectionSource = { + connect: jest.fn(async () => client), + release: jest.fn(async () => {}), + }; + return { client, source }; +}; + +const deferred = () => { + let resolve!: (value: T | PromiseLike) => void; + let reject!: (reason?: unknown) => void; + const promise = new Promise((resolvePromise, rejectPromise) => { + resolve = resolvePromise; + reject = rejectPromise; + }); + return { promise, resolve, reject }; +}; + +const flushMicrotasks = async (): Promise => { + for (let index = 0; index < 20; index++) await Promise.resolve(); +}; + +describe('PgNotificationBrokerRegistry', () => { + it('shares one dedicated listener and reference-counts exact topics', async () => { + const registry = new PgNotificationBrokerRegistry(4); + const { client, source } = createSource(); + const sourceFactory = jest.fn(() => source); + + const [first, second] = await Promise.all([ + registry.acquireForTests('opaque-a', sourceFactory, [ + 'tenant.a', + 'shared', + ]), + registry.acquireForTests('opaque-a', sourceFactory, [ + 'shared', + 'tenant.b', + ]), + ]); + + expect(sourceFactory).toHaveBeenCalledTimes(1); + expect(source.connect).toHaveBeenCalledTimes(1); + expect(client.queries).toEqual([ + 'LISTEN "tenant.a"', + 'LISTEN "shared"', + 'LISTEN "tenant.b"', + ]); + expect(registry.stats()).toMatchObject({ + brokers: 1, + listenerConnections: 1, + leases: 2, + topics: 3, + }); + + await first.release(); + expect(client.queries).toContain('UNLISTEN "tenant.a"'); + expect(client.queries).not.toContain('UNLISTEN "shared"'); + expect(client.release).not.toHaveBeenCalled(); + + await second.release(); + expect(client.queries.slice(-2)).toEqual([ + 'UNLISTEN "shared"', + 'UNLISTEN "tenant.b"', + ]); + expect(client.release).toHaveBeenCalledWith(true); + expect(source.release).toHaveBeenCalledTimes(1); + expect(registry.stats()).toMatchObject({ + brokers: 0, + leases: 0, + topics: 0, + }); + }); + + it('audits three generations on the one pinned listener before admission', async () => { + const registry = new PgNotificationBrokerRegistry(4); + const { client, source } = createSource(); + const sourceFactory = jest.fn(() => source); + + const first = await registry.acquireAttestedForTests( + 'opaque-attested', + sourceFactory, + ['tenant.a'], + roleContract + ); + const second = await registry.acquireAttestedForTests( + 'opaque-attested', + sourceFactory, + ['tenant.b'], + roleContract + ); + const third = await registry.acquireAttestedForTests( + 'opaque-attested', + sourceFactory, + ['tenant.c'], + roleContract + ); + + expect(sourceFactory).toHaveBeenCalledTimes(1); + expect(source.connect).toHaveBeenCalledTimes(1); + expect( + client.queries.filter((query) => query === PG_NOTIFICATION_ROLE_AUDIT_SQL) + ).toHaveLength(3); + expect(client.queries).toEqual([ + 'BEGIN READ ONLY', + 'SET LOCAL jit TO off', + PG_NOTIFICATION_ROLE_AUDIT_SQL, + 'COMMIT', + 'LISTEN "tenant.a"', + 'BEGIN READ ONLY', + 'SET LOCAL jit TO off', + PG_NOTIFICATION_ROLE_AUDIT_SQL, + 'COMMIT', + 'LISTEN "tenant.b"', + 'BEGIN READ ONLY', + 'SET LOCAL jit TO off', + PG_NOTIFICATION_ROLE_AUDIT_SQL, + 'COMMIT', + 'LISTEN "tenant.c"', + ]); + expect(first.roleAudit).toMatchObject({ ...roleContract, safe: true }); + expect(second.roleAudit).toMatchObject({ ...roleContract, safe: true }); + expect(third.roleAudit).toMatchObject({ ...roleContract, safe: true }); + expect(registry.stats()).toMatchObject({ + listenerConnections: 1, + leases: 3, + roleAuditAttempts: 3, + roleAuditFailures: 0, + }); + + await Promise.all([first.release(), second.release(), third.release()]); + }); + + it('serializes concurrent admission audits without another connection', async () => { + const registry = new PgNotificationBrokerRegistry(4); + const { client, source } = createSource(); + const firstCatalogAudit = deferred(); + let catalogAuditsStarted = 0; + let activeCatalogAudits = 0; + let peakCatalogAudits = 0; + client.query.mockImplementation(async (text: string) => { + client.queries.push(text); + if (text === PG_NOTIFICATION_ROLE_AUDIT_SQL) { + catalogAuditsStarted++; + activeCatalogAudits++; + peakCatalogAudits = Math.max(peakCatalogAudits, activeCatalogAudits); + if (catalogAuditsStarted === 1) await firstCatalogAudit.promise; + activeCatalogAudits--; + return { rows: [safeRoleAuditRow] }; + } + return { rows: [] }; + }); + + const acquisitions = [ + registry.acquireAttestedForTests( + 'opaque-attested', + () => source, + ['a'], + roleContract + ), + registry.acquireAttestedForTests( + 'opaque-attested', + () => source, + ['b'], + roleContract + ), + registry.acquireAttestedForTests( + 'opaque-attested', + () => source, + ['c'], + roleContract + ), + ]; + await flushMicrotasks(); + expect(catalogAuditsStarted).toBe(1); + expect(source.connect).toHaveBeenCalledTimes(1); + + firstCatalogAudit.resolve(); + const leases = await Promise.all(acquisitions); + expect(catalogAuditsStarted).toBe(3); + expect(peakCatalogAudits).toBe(1); + expect(source.connect).toHaveBeenCalledTimes(1); + await Promise.all(leases.map((lease) => lease.release())); + }); + + it('bounds a never-resolving admission audit and destroys its client', async () => { + jest.useFakeTimers(); + try { + const registry = new PgNotificationBrokerRegistry(4, 25); + const { client, source } = createSource(); + client.query.mockImplementation((text: string) => { + client.queries.push(text); + if (text === 'BEGIN READ ONLY') return new Promise(() => undefined); + return Promise.resolve({ rows: [] }); + }); + + const acquiring = registry.acquireAttestedForTests( + 'opaque-timeout', + () => source, + ['a'], + roleContract + ); + const rejected = expect(acquiring).rejects.toBeInstanceOf( + PgNotificationOperationTimeoutError + ); + await flushMicrotasks(); + await jest.advanceTimersByTimeAsync(25); + await rejected; + + expect(client.queries).toEqual(['BEGIN READ ONLY']); + expect(client.release).toHaveBeenCalledWith( + expect.any(PgNotificationBrokerFailedError) + ); + expect(source.release).toHaveBeenCalledTimes(1); + expect(registry.stats()).toMatchObject({ + brokers: 0, + listenerConnections: 0, + leases: 0, + fatalFailures: 1, + roleAuditAttempts: 1, + roleAuditFailures: 1, + }); + } finally { + jest.useRealTimers(); + } + }); + + it('revalidates on the pinned listener and fails every lease closed on drift', async () => { + const registry = new PgNotificationBrokerRegistry(4); + const { client, source } = createSource(); + const first = await registry.acquireAttestedForTests( + 'opaque-attested', + () => source, + ['a'], + roleContract + ); + const second = await registry.acquireAttestedForTests( + 'opaque-attested', + () => source, + ['b'], + roleContract + ); + + await expect(first.revalidateRole()).resolves.toMatchObject({ safe: true }); + expect(source.connect).toHaveBeenCalledTimes(1); + expect(registry.stats()).toMatchObject({ + roleAuditAttempts: 3, + roleAuditFailures: 0, + }); + + const firstNext = first.subscribe('a').next(); + const secondNext = second.subscribe('b').next(); + client.roleAuditRow = { ...safeRoleAuditRow, rolsuper: true }; + await expect(second.revalidateRole()).rejects.toBeInstanceOf( + UnsafePgNotificationRoleError + ); + await expect(firstNext).rejects.toBeInstanceOf( + PgNotificationBrokerFailedError + ); + await expect(secondNext).rejects.toBeInstanceOf( + PgNotificationBrokerFailedError + ); + await expect(first.terminated).resolves.toBeInstanceOf( + PgNotificationBrokerFailedError + ); + await expect(second.terminated).resolves.toBeInstanceOf( + PgNotificationBrokerFailedError + ); + expect(source.connect).toHaveBeenCalledTimes(1); + expect(client.release).toHaveBeenCalledWith( + expect.any(PgNotificationBrokerFailedError) + ); + expect(registry.stats()).toMatchObject({ + listenerConnections: 0, + leases: 2, + fatalFailures: 1, + roleAuditAttempts: 4, + roleAuditFailures: 1, + }); + + await Promise.all([first.release(), second.release()]); + }); + + it('bounds a never-resolving TTL role refresh on the pinned listener', async () => { + jest.useFakeTimers(); + try { + const registry = new PgNotificationBrokerRegistry(4, 25); + const { client, source } = createSource(); + const lease = await registry.acquireAttestedForTests( + 'opaque-refresh-timeout', + () => source, + ['a'], + roleContract + ); + client.query.mockImplementation((text: string) => { + client.queries.push(text); + if (text === 'BEGIN READ ONLY') return new Promise(() => undefined); + return Promise.resolve({ rows: [] }); + }); + + const refreshing = lease.revalidateRole(); + const rejected = expect(refreshing).rejects.toBeInstanceOf( + PgNotificationOperationTimeoutError + ); + await flushMicrotasks(); + await jest.advanceTimersByTimeAsync(25); + await rejected; + await expect(lease.terminated).resolves.toBeInstanceOf( + PgNotificationBrokerFailedError + ); + + expect(client.release).toHaveBeenCalledWith( + expect.any(PgNotificationBrokerFailedError) + ); + expect(registry.stats()).toMatchObject({ + listenerConnections: 0, + leases: 1, + fatalFailures: 1, + roleAuditAttempts: 2, + roleAuditFailures: 1, + }); + await lease.release(); + } finally { + jest.useRealTimers(); + } + }); + + it('uses exact topic equality for prefix and quoted-identifier channels', async () => { + const registry = new PgNotificationBrokerRegistry(4); + const { client, source } = createSource(); + const hostileButValid = 'tenant"; UNLISTEN *;--'; + const lease = await registry.acquireForTests('opaque-a', () => source, [ + 'tenant', + 'tenant.longer', + hostileButValid, + ]); + const exact = lease.subscribe('tenant'); + const longer = lease.subscribe('tenant.longer'); + const hostile = lease.subscribe(hostileButValid); + + expect(() => lease.subscribe('ten')).toThrow(PgNotificationTopicError); + expect(client.queries).toContain('LISTEN "tenant""; UNLISTEN *;--"'); + + client.notification('ten', 'wrong-prefix'); + client.notification('tenant.longer', 'longer'); + client.notification(hostileButValid, 'quoted'); + client.notification('tenant', 'exact'); + + await expect(exact.next()).resolves.toEqual({ + done: false, + value: 'exact', + }); + await expect(longer.next()).resolves.toEqual({ + done: false, + value: 'longer', + }); + await expect(hostile.next()).resolves.toEqual({ + done: false, + value: 'quoted', + }); + expect(registry.stats().ignoredNotifications).toBe(1); + await lease.release(); + }); + + it('rejects channels PostgreSQL would truncate, including multi-byte Unicode', async () => { + const registry = new PgNotificationBrokerRegistry(); + const { source } = createSource(); + + const ascii63 = 'a'.repeat(63); + const unicode63 = '界'.repeat(21); + const lease = await registry.acquireForTests('opaque-a', () => source, [ + ascii63, + unicode63, + ]); + expect(lease.topics).toEqual([ascii63, unicode63]); + + await expect( + registry.acquireForTests('opaque-b', () => source, ['a'.repeat(64)]) + ).rejects.toBeInstanceOf(PgNotificationTopicError); + await expect( + registry.acquireForTests('opaque-b', () => source, ['界'.repeat(22)]) + ).rejects.toBeInstanceOf(PgNotificationTopicError); + await expect( + registry.acquireForTests('opaque-b', () => source, [ + `bad${String.fromCharCode(0xd800)}`, + ]) + ).rejects.toBeInstanceOf(PgNotificationTopicError); + await lease.release(); + }); + + it('does not normalize canonically equivalent Unicode topics', async () => { + const registry = new PgNotificationBrokerRegistry(4); + const { client, source } = createSource(); + const composed = 'réaltime'; + const decomposed = 're\u0301altime'; + const lease = await registry.acquireForTests('opaque-a', () => source, [ + composed, + decomposed, + ]); + const composedStream = lease.subscribe(composed); + const decomposedStream = lease.subscribe(decomposed); + + client.notification(composed, 'composed-only'); + client.notification(decomposed, 'decomposed-only'); + + await expect(composedStream.next()).resolves.toMatchObject({ + value: 'composed-only', + }); + await expect(decomposedStream.next()).resolves.toMatchObject({ + value: 'decomposed-only', + }); + await lease.release(); + }); + + it('fans out only to subscribers for the exact allowed topic', async () => { + const registry = new PgNotificationBrokerRegistry(4); + const { client, source } = createSource(); + const first = await registry.acquireForTests('opaque-a', () => source, [ + 'a', + ]); + const second = await registry.acquireForTests('opaque-a', () => source, [ + 'a', + 'b', + ]); + const firstA = first.subscribe('a'); + const secondA = second.subscribe('a'); + const secondB = second.subscribe('b'); + + client.notification('a', 'for-a'); + client.notification('b', 'for-b'); + + await expect(firstA.next()).resolves.toMatchObject({ value: 'for-a' }); + await expect(secondA.next()).resolves.toMatchObject({ value: 'for-a' }); + await expect(secondB.next()).resolves.toMatchObject({ value: 'for-b' }); + await Promise.all([first.release(), second.release()]); + }); + + it('fails only the slow subscriber when its bounded queue overflows', async () => { + const registry = new PgNotificationBrokerRegistry(1); + const { client, source } = createSource(); + const lease = await registry.acquireForTests('opaque-a', () => source, [ + 'events', + ]); + const slow = lease.subscribe('events'); + const fast = lease.subscribe('events'); + + const fastFirst = fast.next(); + client.notification('events', 'one'); + const fastSecond = fast.next(); + client.notification('events', 'two'); + + await expect(fastFirst).resolves.toMatchObject({ value: 'one' }); + await expect(fastSecond).resolves.toMatchObject({ value: 'two' }); + await expect(slow.next()).rejects.toBeInstanceOf( + PgNotificationQueueOverflowError + ); + expect(registry.stats()).toMatchObject({ + subscribers: 1, + queueOverflows: 1, + }); + await lease.release(); + }); + + it('fails every active subscriber and never silently reconnects', async () => { + const registry = new PgNotificationBrokerRegistry(4); + const { client, source } = createSource(); + const sourceFactory = jest.fn(() => source); + const first = await registry.acquireForTests('opaque-a', sourceFactory, [ + 'a', + ]); + const second = await registry.acquireForTests('opaque-a', sourceFactory, [ + 'b', + ]); + const firstNext = first.subscribe('a').next(); + const secondNext = second.subscribe('b').next(); + + client.emit('error', new Error('socket lost')); + + await expect(firstNext).rejects.toBeInstanceOf( + PgNotificationBrokerFailedError + ); + await expect(secondNext).rejects.toBeInstanceOf( + PgNotificationBrokerFailedError + ); + await expect(first.terminated).resolves.toBeInstanceOf( + PgNotificationBrokerFailedError + ); + await expect( + registry.acquireForTests('opaque-a', sourceFactory, ['a']) + ).rejects.toBeInstanceOf(PgNotificationBrokerFailedError); + expect(source.connect).toHaveBeenCalledTimes(1); + expect(client.release).toHaveBeenCalledWith( + expect.any(PgNotificationBrokerFailedError) + ); + + await Promise.all([first.release(), second.release()]); + const replacement = createSource(); + const explicitReplacement = await registry.acquireForTests( + 'opaque-a', + () => replacement.source, + ['a'] + ); + expect(replacement.source.connect).toHaveBeenCalledTimes(1); + await explicitReplacement.release(); + }); + + it('bounds a never-resolving LISTEN and fails admission closed', async () => { + jest.useFakeTimers(); + try { + const registry = new PgNotificationBrokerRegistry(4, 25); + const { client, source } = createSource(); + client.query.mockImplementation((text: string) => { + client.queries.push(text); + if (text.startsWith('LISTEN')) return new Promise(() => undefined); + return Promise.resolve({ rows: [] }); + }); + + const acquiring = registry.acquireForTests( + 'opaque-listen-timeout', + () => source, + ['a'] + ); + const rejected = expect(acquiring).rejects.toMatchObject({ + code: 'PG_NOTIFICATION_BROKER_FAILED', + cause: { code: 'PG_NOTIFICATION_OPERATION_TIMEOUT' }, + }); + await flushMicrotasks(); + await jest.advanceTimersByTimeAsync(25); + await rejected; + + expect(client.release).toHaveBeenCalledWith( + expect.any(PgNotificationBrokerFailedError) + ); + expect(source.release).toHaveBeenCalledTimes(1); + expect(registry.stats()).toMatchObject({ + brokers: 0, + listenerConnections: 0, + fatalFailures: 1, + }); + } finally { + jest.useRealTimers(); + } + }); + + it('fails closed when a listener emits a malformed notification', async () => { + const registry = new PgNotificationBrokerRegistry(4); + const { client, source } = createSource(); + const lease = await registry.acquireForTests('opaque-a', () => source, [ + 'a', + ]); + const next = lease.subscribe('a').next(); + + client.emit('notification', { channel: 'a', payload: { hostile: true } }); + + await expect(next).rejects.toBeInstanceOf(PgNotificationBrokerFailedError); + await expect(lease.terminated).resolves.toBeInstanceOf( + PgNotificationBrokerFailedError + ); + await lease.release(); + }); + + it('makes double release idempotent and awaits UNLISTEN plus both releases', async () => { + const registry = new PgNotificationBrokerRegistry(4); + const { client, source } = createSource(); + const unlisten = deferred(); + const clientReleased = deferred(); + const sourceReleased = deferred(); + client.query.mockImplementation(async (text: string) => { + client.queries.push(text); + if (text.startsWith('UNLISTEN')) await unlisten.promise; + return { rows: [] }; + }); + client.release.mockImplementation(async () => clientReleased.promise); + (source.release as jest.Mock).mockImplementation( + async () => sourceReleased.promise + ); + const lease = await registry.acquireForTests('opaque-a', () => source, [ + 'a', + ]); + + const firstRelease = lease.release(); + const secondRelease = lease.release(); + expect(firstRelease).toBe(secondRelease); + await flushMicrotasks(); + expect(client.queries).toContain('UNLISTEN "a"'); + + let settled = false; + void firstRelease.then(() => { + settled = true; + }); + unlisten.resolve(); + await flushMicrotasks(); + expect(settled).toBe(false); + clientReleased.resolve(); + await flushMicrotasks(); + expect(settled).toBe(false); + sourceReleased.resolve(); + await firstRelease; + await expect(lease.terminated).resolves.toBeNull(); + expect(settled).toBe(true); + expect(client.release).toHaveBeenCalledTimes(1); + expect(source.release).toHaveBeenCalledTimes(1); + }); + + it('serializes a final release against a concurrent new acquisition', async () => { + const registry = new PgNotificationBrokerRegistry(4); + const firstSource = createSource(); + const unlisten = deferred(); + firstSource.client.query.mockImplementation(async (text: string) => { + firstSource.client.queries.push(text); + if (text.startsWith('UNLISTEN')) await unlisten.promise; + return { rows: [] }; + }); + const first = await registry.acquireForTests( + 'opaque-a', + () => firstSource.source, + ['a'] + ); + const releasing = first.release(); + await flushMicrotasks(); + + const secondSource = createSource(); + const acquiring = registry.acquireForTests( + 'opaque-a', + () => secondSource.source, + ['a'] + ); + await flushMicrotasks(); + expect(secondSource.source.connect).not.toHaveBeenCalled(); + + unlisten.resolve(); + await releasing; + const second = await acquiring; + expect(secondSource.source.connect).toHaveBeenCalledTimes(1); + await second.release(); + }); + + it('makes concurrent registry close calls await the same teardown', async () => { + const registry = new PgNotificationBrokerRegistry(4); + const { source } = createSource(); + const sourceReleased = deferred(); + (source.release as jest.Mock).mockImplementation( + async () => sourceReleased.promise + ); + await registry.acquireForTests('opaque-a', () => source, ['a']); + + const firstClose = registry.close(); + const secondClose = registry.close(); + let firstSettled = false; + let secondSettled = false; + void firstClose.then(() => { + firstSettled = true; + }); + void secondClose.then(() => { + secondSettled = true; + }); + await flushMicrotasks(); + + expect(firstSettled).toBe(false); + expect(secondSettled).toBe(false); + sourceReleased.resolve(); + await Promise.all([firstClose, secondClose]); + expect(source.release).toHaveBeenCalledTimes(1); + expect(registry.stats()).toMatchObject({ brokers: 0, leases: 0 }); + }); + + it('bounds a never-resolving UNLISTEN so teardown cannot hang', async () => { + jest.useFakeTimers(); + try { + const registry = new PgNotificationBrokerRegistry(4, 25); + const { client, source } = createSource(); + await registry.acquireForTests('opaque-unlisten-timeout', () => source, [ + 'a', + ]); + client.query.mockImplementation((text: string) => { + client.queries.push(text); + if (text.startsWith('UNLISTEN')) return new Promise(() => undefined); + return Promise.resolve({ rows: [] }); + }); + + const closing = registry.close(); + const rejected = expect(closing).rejects.toMatchObject({ + code: 'PG_NOTIFICATION_BROKER_FAILED', + cause: { code: 'PG_NOTIFICATION_OPERATION_TIMEOUT' }, + }); + await flushMicrotasks(); + await jest.advanceTimersByTimeAsync(25); + await rejected; + + expect(client.release).toHaveBeenCalledWith( + expect.any(PgNotificationBrokerFailedError) + ); + expect(source.release).toHaveBeenCalledTimes(1); + expect(registry.stats()).toMatchObject({ + brokers: 0, + listenerConnections: 0, + leases: 0, + fatalFailures: 1, + }); + } finally { + jest.useRealTimers(); + } + }); + + it('drains an in-flight acquisition before registry close resolves', async () => { + const registry = new PgNotificationBrokerRegistry(4); + const { client, source } = createSource(); + const connected = deferred(); + const sourceReleased = deferred(); + (source.connect as jest.Mock).mockImplementation( + async () => connected.promise + ); + (source.release as jest.Mock).mockImplementation( + async () => sourceReleased.promise + ); + const acquiring = registry.acquireForTests('opaque-a', () => source, ['a']); + await flushMicrotasks(); + expect(source.connect).toHaveBeenCalledTimes(1); + + const closing = registry.close(); + let closeSettled = false; + void closing.then(() => { + closeSettled = true; + }); + connected.resolve(client); + await flushMicrotasks(); + const closeSettledBeforeSourceRelease = closeSettled; + const issuedListenDuringClose = client.queries.includes('LISTEN "a"'); + sourceReleased.resolve(); + await expect(acquiring).rejects.toThrow( + 'PostgreSQL notification broker registry is closed' + ); + await closing; + expect(closeSettledBeforeSourceRelease).toBe(false); + expect(issuedListenDuringClose).toBe(false); + expect(client.release).toHaveBeenCalledTimes(1); + expect(source.release).toHaveBeenCalledTimes(1); + expect(registry.stats()).toMatchObject({ brokers: 0, leases: 0 }); + }); + + it('UNLISTENs a provisional topic when close races an in-flight LISTEN', async () => { + const registry = new PgNotificationBrokerRegistry(4); + const { client, source } = createSource(); + const listened = deferred(); + client.query.mockImplementation(async (text: string) => { + client.queries.push(text); + if (text === 'LISTEN "a"') await listened.promise; + return { rows: [] }; + }); + const acquiring = registry.acquireForTests('opaque-a', () => source, ['a']); + await flushMicrotasks(); + expect(client.queries).toEqual(['LISTEN "a"']); + + const closing = registry.close(); + listened.resolve(); + await expect(acquiring).rejects.toThrow( + 'PostgreSQL notification broker registry is closed' + ); + await closing; + + expect(client.queries).toEqual(['LISTEN "a"', 'UNLISTEN *']); + expect(client.release).toHaveBeenCalledTimes(1); + expect(source.release).toHaveBeenCalledTimes(1); + }); + + it('reports a failed UNLISTEN only after finishing registry teardown', async () => { + const registry = new PgNotificationBrokerRegistry(4); + const { client, source } = createSource(); + client.query.mockImplementation(async (text: string) => { + client.queries.push(text); + if (text.startsWith('UNLISTEN')) throw new Error('unlisten failed'); + return { rows: [] }; + }); + await registry.acquireForTests('opaque-a', () => source, ['a']); + + await expect(registry.close()).rejects.toBeInstanceOf( + PgNotificationBrokerFailedError + ); + expect(client.release).toHaveBeenCalledTimes(1); + expect(source.release).toHaveBeenCalledTimes(1); + expect(registry.stats()).toMatchObject({ brokers: 0, leases: 0 }); + }); + + it('aggregates client and pool-lease cleanup failures after teardown attempts both', async () => { + const registry = new PgNotificationBrokerRegistry(4); + const { client, source } = createSource(); + client.release.mockRejectedValue(new Error('client release failed')); + (source.release as jest.Mock).mockRejectedValue( + new Error('source release failed') + ); + await registry.acquireForTests('opaque-cleanup', () => source, ['a']); + + const closing = registry.close(); + await expect(closing).rejects.toMatchObject({ + code: 'PG_NOTIFICATION_BROKER_FAILED', + cause: expect.any(AggregateError), + }); + expect(client.release).toHaveBeenCalledTimes(1); + expect(source.release).toHaveBeenCalledTimes(1); + expect(registry.stats()).toMatchObject({ brokers: 0, leases: 0 }); + }); +}); + +describe('getPgNotificationBrokerIdentity', () => { + const baseConfig = { + host: 'db.internal', + port: 5432, + database: 'customer', + user: 'listener', + password: 'secret', + }; + + it('is versioned, opaque, stable, and includes the canonical SSL contract', () => { + const first = getPgNotificationBrokerIdentity({ + ...baseConfig, + ssl: { rejectUnauthorized: true, ca: 'ca-one' }, + pool: { + connectionTimeoutMillis: DEFAULT_PG_NOTIFICATION_OPERATION_TIMEOUT_MS, + }, + }); + const reordered = getPgNotificationBrokerIdentity({ + ...baseConfig, + ssl: { ca: 'ca-one', rejectUnauthorized: true }, + pool: { + connectionTimeoutMillis: DEFAULT_PG_NOTIFICATION_OPERATION_TIMEOUT_MS, + }, + }); + const changedTls = getPgNotificationBrokerIdentity({ + ...baseConfig, + ssl: { rejectUnauthorized: false, ca: 'ca-one' }, + pool: { + connectionTimeoutMillis: DEFAULT_PG_NOTIFICATION_OPERATION_TIMEOUT_MS, + }, + }); + const changedDeadline = getPgNotificationBrokerIdentity({ + ...baseConfig, + ssl: { rejectUnauthorized: true, ca: 'ca-one' }, + pool: { + connectionTimeoutMillis: + DEFAULT_PG_NOTIFICATION_OPERATION_TIMEOUT_MS + 1, + }, + }); + + expect(first).toBe(reordered); + expect(first).not.toBe(changedTls); + expect(first).not.toBe(changedDeadline); + expect(first).toMatch(/^pg-notification-broker:v1:pg:v1:[a-f0-9]{64}$/); + expect(first).not.toContain('listener'); + expect(first).not.toContain('secret'); + }); + + it.each([0, -1, 1.5, 2_147_483_648])( + 'rejects invalid notification operation timeout %p before identity publication', + (connectionTimeoutMillis) => { + expect(() => + getPgNotificationBrokerIdentity({ + ...baseConfig, + pool: { connectionTimeoutMillis }, + }) + ).toThrow('notification operation timeout'); + } + ); + + it('uses one credential-free identity for the same physical database target', () => { + const first = getPgNotificationDatabaseIdentity({ + ...baseConfig, + ssl: { rejectUnauthorized: true, ca: 'ca-one' }, + pool: { max: 2 }, + }); + const rotated = getPgNotificationDatabaseIdentity({ + ...baseConfig, + user: 'rotated-listener', + password: 'rotated-secret', + ssl: { ca: 'different-ca', rejectUnauthorized: false }, + pool: { max: 20, idleTimeoutMillis: 99_000 }, + }); + const otherHost = getPgNotificationDatabaseIdentity({ + ...baseConfig, + host: 'other-db.internal', + }); + const otherPort = getPgNotificationDatabaseIdentity({ + ...baseConfig, + port: 5433, + }); + const otherDatabase = getPgNotificationDatabaseIdentity({ + ...baseConfig, + database: 'other-customer', + ssl: { rejectUnauthorized: true, ca: 'ca-one' }, + }); + + expect(first).toBe(rotated); + expect(first).not.toBe(otherHost); + expect(first).not.toBe(otherPort); + expect(first).not.toBe(otherDatabase); + expect(first).toMatch( + /^pg-notification-database:v1:pg-target:v1:[a-f0-9]{64}$/ + ); + expect(first).not.toContain('listener'); + expect(first).not.toContain('secret'); + }); +}); diff --git a/postgres/pg-cache/src/index.ts b/postgres/pg-cache/src/index.ts index 77bbc13908..b7a5853489 100644 --- a/postgres/pg-cache/src/index.ts +++ b/postgres/pg-cache/src/index.ts @@ -18,6 +18,28 @@ export { PgPoolCapacityError, teardownPgPools } from './lru'; +export { + acquirePgNotificationBroker, + assertValidPgNotificationTopic, + DEFAULT_PG_NOTIFICATION_OPERATION_TIMEOUT_MS, + getPgNotificationBrokerIdentity, + getPgNotificationBrokerStats, + getPgNotificationDatabaseIdentity, + PG_NOTIFICATION_BROKER_FAILED_ERROR_CODE, + PG_NOTIFICATION_BROKER_IDENTITY_VERSION, + PG_NOTIFICATION_DATABASE_IDENTITY_VERSION, + PG_NOTIFICATION_LEASE_RELEASED_ERROR_CODE, + PG_NOTIFICATION_OPERATION_TIMEOUT_ERROR_CODE, + PG_NOTIFICATION_QUEUE_CAPACITY, + PG_NOTIFICATION_QUEUE_OVERFLOW_ERROR_CODE, + PG_NOTIFICATION_TOPIC_ERROR_CODE, + PgNotificationBrokerFailedError, + PgNotificationLeaseReleasedError, + PgNotificationOperationTimeoutError, + PgNotificationQueueOverflowError, + PgNotificationTopicError, + teardownPgNotificationBrokers, +} from './notification-broker'; export { assertPgNotificationRole, assertPgNotificationRoleClient, @@ -49,6 +71,13 @@ export type { PgPoolLease, PoolCleanupCallback, } from './lru'; +export type { + AcquirePgNotificationBrokerOptions, + PgAttestedNotificationBrokerLease, + PgNotificationBrokerLease, + PgNotificationBrokerStats, + PgNotificationListenerConfig, +} from './notification-broker'; export type { PgNotificationRoleAudit, PgNotificationRoleClient, diff --git a/postgres/pg-cache/src/notification-broker.ts b/postgres/pg-cache/src/notification-broker.ts new file mode 100644 index 0000000000..682c2ce7ce --- /dev/null +++ b/postgres/pg-cache/src/notification-broker.ts @@ -0,0 +1,1178 @@ +import type { PgConfig, PgPoolConfig } from 'pg-env'; + +import { + assertPgNotificationRoleClient, + type PgNotificationRoleAudit, + type PgNotificationRoleClient, + type PgNotificationRoleContract, +} from './notification-role'; +import { + acquirePgPool, + getPgDatabaseTargetIdentity, + getPgPoolConfig, + getPgPoolIdentity, +} from './pg'; + +export const PG_NOTIFICATION_BROKER_IDENTITY_VERSION = + 'pg-notification-broker:v1'; +export const PG_NOTIFICATION_DATABASE_IDENTITY_VERSION = + 'pg-notification-database:v1'; +export const PG_NOTIFICATION_QUEUE_CAPACITY = 256; +export const DEFAULT_PG_NOTIFICATION_OPERATION_TIMEOUT_MS = 5_000; + +export const PG_NOTIFICATION_TOPIC_ERROR_CODE = 'PG_NOTIFICATION_TOPIC_INVALID'; +export const PG_NOTIFICATION_BROKER_FAILED_ERROR_CODE = + 'PG_NOTIFICATION_BROKER_FAILED'; +export const PG_NOTIFICATION_QUEUE_OVERFLOW_ERROR_CODE = + 'PG_NOTIFICATION_QUEUE_OVERFLOW'; +export const PG_NOTIFICATION_LEASE_RELEASED_ERROR_CODE = + 'PG_NOTIFICATION_LEASE_RELEASED'; +export const PG_NOTIFICATION_OPERATION_TIMEOUT_ERROR_CODE = + 'PG_NOTIFICATION_OPERATION_TIMEOUT'; + +type PromiseOrDirect = T | Promise; + +export interface PgNotification { + channel: string; + payload?: string; +} + +export interface PgNotificationClient { + query(text: string, values?: readonly unknown[]): Promise; + on(event: string, listener: (...args: any[]) => void): unknown; + off(event: string, listener: (...args: any[]) => void): unknown; + release(error?: Error | boolean): PromiseOrDirect; +} + +export interface PgNotificationConnectionSource { + connect(): Promise; + release(): PromiseOrDirect; +} + +export interface PgNotificationBrokerLease { + /** Versioned digest of the complete listener connection contract. */ + readonly identity: string; + /** Frozen, exact PostgreSQL channels this lease may subscribe to. */ + readonly topics: readonly string[]; + /** Resolves on fatal broker failure or with null after graceful release. */ + readonly terminated: Promise; + subscribe(topic: string): AsyncIterableIterator; + /** Idempotent and awaited through UNLISTEN and connection release. */ + release(): Promise; +} + +/** + * A production lease whose login was audited on the same pinned PostgreSQL + * client before admission. Arbitrary SQL and the client itself stay private. + */ +export interface PgAttestedNotificationBrokerLease extends PgNotificationBrokerLease { + readonly roleAudit: PgNotificationRoleAudit; + revalidateRole(): Promise; +} + +export interface AcquirePgNotificationBrokerOptions { + /** Every channel this generation may observe. Prefix matching is never used. */ + topics: readonly string[]; +} + +export type PgNotificationListenerConfig = PgConfig & { pool?: PgPoolConfig }; + +export interface PgNotificationBrokerStats { + brokers: number; + listenerConnections: number; + leases: number; + topics: number; + subscribers: number; + acquisitions: number; + releases: number; + notifications: number; + ignoredNotifications: number; + queueOverflows: number; + fatalFailures: number; + roleAuditAttempts: number; + roleAuditFailures: number; +} + +type PgNotificationBrokerSnapshot = Pick< + PgNotificationBrokerStats, + 'listenerConnections' | 'leases' | 'topics' | 'subscribers' +>; + +interface MutableBrokerCounters { + acquisitions: number; + releases: number; + notifications: number; + ignoredNotifications: number; + queueOverflows: number; + fatalFailures: number; + roleAuditAttempts: number; + roleAuditFailures: number; +} + +interface Deferred { + promise: Promise; + resolve(value: T): void; + reject(error: unknown): void; +} + +const deferred = (): Deferred => { + let resolve!: (value: T) => void; + let reject!: (error: unknown) => void; + const promise = new Promise((resolvePromise, rejectPromise) => { + resolve = resolvePromise; + reject = rejectPromise; + }); + return { promise, resolve, reject }; +}; + +export class PgNotificationTopicError extends Error { + readonly code = PG_NOTIFICATION_TOPIC_ERROR_CODE; + + constructor( + readonly topic: unknown, + reason: string + ) { + super(`Invalid PostgreSQL notification topic: ${reason}`); + this.name = 'PgNotificationTopicError'; + } +} + +export class PgNotificationBrokerFailedError extends Error { + readonly code = PG_NOTIFICATION_BROKER_FAILED_ERROR_CODE; + + constructor(reason: unknown) { + const cause = reason instanceof Error ? reason : new Error(String(reason)); + super( + 'PostgreSQL notification broker failed; all subscribers were terminated', + { + cause, + } + ); + this.name = 'PgNotificationBrokerFailedError'; + } +} + +export class PgNotificationQueueOverflowError extends Error { + readonly code = PG_NOTIFICATION_QUEUE_OVERFLOW_ERROR_CODE; + + constructor( + readonly topic: string, + readonly capacity: number + ) { + super( + `PostgreSQL notification subscriber queue for ${JSON.stringify(topic)} ` + + `exceeded its fixed capacity of ${capacity}` + ); + this.name = 'PgNotificationQueueOverflowError'; + } +} + +export class PgNotificationLeaseReleasedError extends Error { + readonly code = PG_NOTIFICATION_LEASE_RELEASED_ERROR_CODE; + + constructor() { + super('PostgreSQL notification broker lease has been released'); + this.name = 'PgNotificationLeaseReleasedError'; + } +} + +export class PgNotificationOperationTimeoutError extends Error { + readonly code = PG_NOTIFICATION_OPERATION_TIMEOUT_ERROR_CODE; + + constructor( + readonly operation: 'role-audit' | 'listen' | 'unlisten', + readonly timeoutMs: number + ) { + super( + `PostgreSQL notification ${operation} exceeded its fixed ${timeoutMs}ms deadline` + ); + this.name = 'PgNotificationOperationTimeoutError'; + } +} + +class BrokerClosedError extends Error {} + +const containsUnpairedSurrogate = (value: string): boolean => { + for (let index = 0; index < value.length; index++) { + const code = value.charCodeAt(index); + if (code >= 0xd800 && code <= 0xdbff) { + const next = value.charCodeAt(index + 1); + if (!(next >= 0xdc00 && next <= 0xdfff)) return true; + index++; + } else if (code >= 0xdc00 && code <= 0xdfff) { + return true; + } + } + return false; +}; + +/** + * PostgreSQL identifiers are limited to 63 UTF-8 bytes. PostgreSQL truncates + * longer identifiers, so accepting them here could collapse distinct tenant + * topics onto one physical LISTEN channel. + */ +export function assertValidPgNotificationTopic( + topic: unknown +): asserts topic is string { + if (typeof topic !== 'string') { + throw new PgNotificationTopicError(topic, 'the topic must be a string'); + } + if (topic.length === 0) { + throw new PgNotificationTopicError(topic, 'the topic must not be empty'); + } + if (topic.includes('\0')) { + throw new PgNotificationTopicError(topic, 'NUL bytes are not allowed'); + } + if (containsUnpairedSurrogate(topic)) { + throw new PgNotificationTopicError( + topic, + 'unpaired UTF-16 surrogates are not allowed' + ); + } + const bytes = Buffer.byteLength(topic, 'utf8'); + if (bytes > 63) { + throw new PgNotificationTopicError( + topic, + `the UTF-8 encoding is ${bytes} bytes; PostgreSQL allows at most 63` + ); + } +} + +const normalizeTopics = (topics: readonly string[]): readonly string[] => { + if (!Array.isArray(topics) || topics.length === 0) { + throw new PgNotificationTopicError( + topics, + 'at least one exact topic is required' + ); + } + for (const topic of topics) assertValidPgNotificationTopic(topic); + return Object.freeze([...new Set(topics)]); +}; + +const quoteIdentifier = (identifier: string): string => + `"${identifier.replace(/"/g, '""')}"`; + +class BoundedNotificationQueue implements AsyncIterableIterator { + private readonly buffered: string[] = []; + private readonly waiting: Deferred>[] = []; + private terminal: 'open' | 'complete' | 'failed' = 'open'; + private failure: Error | null = null; + + constructor( + private readonly topic: string, + private readonly capacity: number, + private readonly onClose: () => void, + private readonly onOverflow: () => void + ) {} + + [Symbol.asyncIterator](): AsyncIterableIterator { + return this; + } + + next(): Promise> { + const buffered = this.buffered.shift(); + if (buffered !== undefined) { + return Promise.resolve({ done: false, value: buffered }); + } + if (this.terminal === 'failed') return Promise.reject(this.failure); + if (this.terminal === 'complete') { + return Promise.resolve({ done: true, value: undefined }); + } + + const result = deferred>(); + this.waiting.push(result); + return result.promise; + } + + return(value?: unknown): Promise> { + this.complete(); + return Promise.resolve({ done: true, value: value as string }); + } + + throw(error?: unknown): Promise> { + const failure = error instanceof Error ? error : new Error(String(error)); + this.fail(failure); + return Promise.reject(failure); + } + + push(payload: string): void { + if (this.terminal !== 'open') return; + const waiter = this.waiting.shift(); + if (waiter) { + waiter.resolve({ done: false, value: payload }); + return; + } + if (this.buffered.length >= this.capacity) { + this.onOverflow(); + this.fail( + new PgNotificationQueueOverflowError(this.topic, this.capacity) + ); + return; + } + this.buffered.push(payload); + } + + complete(): void { + if (this.terminal !== 'open') return; + this.terminal = 'complete'; + this.buffered.length = 0; + for (const waiter of this.waiting.splice(0)) { + waiter.resolve({ done: true, value: undefined }); + } + this.onClose(); + } + + fail(error: Error): void { + if (this.terminal !== 'open') return; + this.terminal = 'failed'; + this.failure = error; + this.buffered.length = 0; + for (const waiter of this.waiting.splice(0)) waiter.reject(error); + this.onClose(); + } +} + +type BrokerState = 'new' | 'active' | 'failed' | 'closing' | 'closed'; +type ConnectionSourceFactory = + () => PromiseOrDirect; +type NotificationOperation = PgNotificationOperationTimeoutError['operation']; + +const MAX_TIMER_DELAY_MS = 2_147_483_647; + +const assertNotificationOperationTimeoutMs = (timeoutMs: number): number => { + if ( + !Number.isSafeInteger(timeoutMs) || + timeoutMs <= 0 || + timeoutMs > MAX_TIMER_DELAY_MS + ) { + throw new TypeError( + 'PostgreSQL notification operation timeout must be an integer ' + + `between 1 and ${MAX_TIMER_DELAY_MS}` + ); + } + return timeoutMs; +}; + +const getNotificationOperationTimeoutMs = ( + listenerPgConfig: PgNotificationListenerConfig +): number => { + // Preserve this API's narrower deadline contract and stable error before the + // generic pool validator runs as part of identity construction. + const configured = listenerPgConfig.pool?.connectionTimeoutMillis; + return assertNotificationOperationTimeoutMs( + configured ?? + getPgPoolConfig(listenerPgConfig.pool).connectionTimeoutMillis ?? + DEFAULT_PG_NOTIFICATION_OPERATION_TIMEOUT_MS + ); +}; + +class NotificationBrokerLease implements PgAttestedNotificationBrokerLease { + readonly topics: readonly string[]; + readonly terminated: Promise; + private readonly allowedTopics: ReadonlySet; + private readonly termination = + deferred(); + private readonly queues = new Map>(); + private audit: PgNotificationRoleAudit | null = null; + private released = false; + private releasePromise: Promise | null = null; + + constructor( + readonly identity: string, + topics: readonly string[], + private readonly broker: NotificationBrokerRecord, + private readonly queueCapacity: number, + private readonly counters: MutableBrokerCounters, + readonly roleContract: Readonly | null + ) { + this.topics = topics; + this.terminated = this.termination.promise; + this.allowedTopics = new Set(topics); + } + + get subscriberCount(): number { + let count = 0; + for (const topicQueues of this.queues.values()) count += topicQueues.size; + return count; + } + + get isReleased(): boolean { + return this.released; + } + + get roleAudit(): PgNotificationRoleAudit { + if (!this.audit) { + throw new Error( + 'PostgreSQL notification broker lease is not role-attested' + ); + } + return this.audit; + } + + setRoleAudit(audit: PgNotificationRoleAudit): void { + this.audit = audit; + } + + revalidateRole(): Promise { + if (this.released) + return Promise.reject(new PgNotificationLeaseReleasedError()); + if (!this.roleContract) { + return Promise.reject( + new Error('PostgreSQL notification broker lease is not role-attested') + ); + } + return this.broker.revalidateLeaseRole(this); + } + + subscribe(topic: string): AsyncIterableIterator { + if (this.released) throw new PgNotificationLeaseReleasedError(); + this.broker.assertAvailable(); + if (!this.allowedTopics.has(topic)) { + throw new PgNotificationTopicError( + topic, + "the topic is not in this lease's exact allowlist" + ); + } + + let topicQueues = this.queues.get(topic); + if (!topicQueues) { + topicQueues = new Set(); + this.queues.set(topic, topicQueues); + } + let queue!: BoundedNotificationQueue; + queue = new BoundedNotificationQueue( + topic, + this.queueCapacity, + () => { + topicQueues!.delete(queue); + if (topicQueues!.size === 0) this.queues.delete(topic); + }, + () => { + this.counters.queueOverflows++; + } + ); + topicQueues.add(queue); + return queue; + } + + dispatch(topic: string, payload: string): void { + const queues = this.queues.get(topic); + if (!queues) return; + for (const queue of [...queues]) queue.push(payload); + } + + fail(error: PgNotificationBrokerFailedError): void { + this.termination.resolve(error); + for (const queues of [...this.queues.values()]) { + for (const queue of [...queues]) queue.fail(error); + } + } + + release(): Promise { + if (this.releasePromise) return this.releasePromise; + this.released = true; + for (const queues of [...this.queues.values()]) { + for (const queue of [...queues]) queue.complete(); + } + this.releasePromise = this.broker.releaseLease(this); + void this.releasePromise.then( + () => this.termination.resolve(null), + (error) => + this.termination.resolve( + error instanceof PgNotificationBrokerFailedError + ? error + : new PgNotificationBrokerFailedError(error) + ) + ); + return this.releasePromise; + } +} + +class NotificationBrokerRecord { + private state: BrokerState = 'new'; + private acceptingLeases = true; + private operation: Promise = Promise.resolve(); + private source: PgNotificationConnectionSource | null = null; + private client: PgNotificationClient | null = null; + private clientCleanup: Promise | null = null; + private sourceCleanup: Promise | null = null; + private fatalError: PgNotificationBrokerFailedError | null = null; + private readonly leases = new Set(); + private readonly topicReferences = new Map(); + /** Includes provisional LISTENs whose lease admission has not committed yet. */ + private readonly listenedTopics = new Set(); + + private readonly onNotification = (notification: PgNotification): void => { + if (this.state !== 'active') return; + if ( + !notification || + typeof notification.channel !== 'string' || + (notification.payload !== undefined && + typeof notification.payload !== 'string') + ) { + this.markFailed( + new Error('PostgreSQL listener emitted a malformed notification') + ); + return; + } + if (!this.topicReferences.has(notification.channel)) { + this.counters.ignoredNotifications++; + return; + } + this.counters.notifications++; + const payload = notification.payload ?? ''; + for (const lease of [...this.leases]) { + lease.dispatch(notification.channel, payload); + } + }; + + private readonly onClientError = (error: unknown): void => { + this.markFailed(error); + }; + + private readonly onClientEnd = (): void => { + this.markFailed( + new Error('PostgreSQL notification listener connection ended') + ); + }; + + constructor( + readonly identity: string, + private readonly sourcePromise: Promise, + private readonly queueCapacity: number, + private readonly operationTimeoutMs: number, + private readonly counters: MutableBrokerCounters, + private readonly onTerminal: (record: NotificationBrokerRecord) => void + ) {} + + get snapshot(): PgNotificationBrokerSnapshot { + let subscribers = 0; + for (const lease of this.leases) subscribers += lease.subscriberCount; + return { + listenerConnections: this.client ? 1 : 0, + leases: this.leases.size, + topics: this.topicReferences.size, + subscribers, + }; + } + + assertAvailable(): void { + if (this.state === 'failed') throw this.fatalError!; + if (this.state !== 'active') throw new PgNotificationLeaseReleasedError(); + } + + async acquire( + topics: readonly string[], + roleContract: Readonly | null = null + ): Promise { + const lease = new NotificationBrokerLease( + this.identity, + topics, + this, + this.queueCapacity, + this.counters, + roleContract + ); + await this.enqueue(async () => { + if (!this.acceptingLeases) throw new BrokerClosedError(); + if (this.state === 'failed') throw this.fatalError!; + if (this.state === 'closing' || this.state === 'closed') { + throw new BrokerClosedError(); + } + const client = await this.ensureClient(); + if (!this.acceptingLeases) throw new BrokerClosedError(); + if (roleContract) { + lease.setRoleAudit(await this.auditRole(client, roleContract)); + } + if (!this.acceptingLeases) throw new BrokerClosedError(); + for (const topic of topics) { + if ((this.topicReferences.get(topic) ?? 0) === 0) { + await this.executeListenerQuery( + client, + `LISTEN ${quoteIdentifier(topic)}` + ); + this.listenedTopics.add(topic); + } + } + if (!this.acceptingLeases || this.state !== 'active') { + if (this.fatalError) throw this.fatalError; + throw new BrokerClosedError(); + } + for (const topic of topics) { + this.topicReferences.set( + topic, + (this.topicReferences.get(topic) ?? 0) + 1 + ); + } + this.leases.add(lease); + this.counters.acquisitions++; + }); + return lease; + } + + async revalidateLeaseRole( + lease: NotificationBrokerLease + ): Promise { + return this.enqueue(async () => { + if (lease.isReleased || !this.leases.has(lease)) { + throw new PgNotificationLeaseReleasedError(); + } + if (this.state === 'failed') throw this.fatalError!; + if (this.state !== 'active' || !this.client || !lease.roleContract) { + throw new PgNotificationLeaseReleasedError(); + } + const audit = await this.auditRole(this.client, lease.roleContract); + lease.setRoleAudit(audit); + return audit; + }); + } + + async releaseLease(lease: NotificationBrokerLease): Promise { + return this.enqueue(async () => { + if (!this.leases.delete(lease)) return; + this.counters.releases++; + + const topicsToUnlisten: string[] = []; + for (const topic of lease.topics) { + const next = (this.topicReferences.get(topic) ?? 0) - 1; + if (next <= 0) { + this.topicReferences.delete(topic); + topicsToUnlisten.push(topic); + } else { + this.topicReferences.set(topic, next); + } + } + + let releaseError: Error | null = null; + if (this.state === 'active' && this.client) { + for (const topic of topicsToUnlisten) { + try { + await this.executeListenerQuery( + this.client, + `UNLISTEN ${quoteIdentifier(topic)}` + ); + this.listenedTopics.delete(topic); + } catch (error) { + releaseError = + this.fatalError ?? new PgNotificationBrokerFailedError(error); + break; + } + } + } + + if (this.leases.size === 0) await this.closeUnused(); + if (releaseError) throw releaseError; + }); + } + + async closeAll(): Promise { + this.acceptingLeases = false; + // Cross the serialized-operation barrier before snapshotting leases. This + // either rejects an acquisition already waiting on connect/LISTEN or makes + // its completed lease visible to the release snapshot below. + await this.enqueue((): void => undefined); + const releases = [...this.leases].map((lease) => lease.release()); + const releaseResults = await Promise.allSettled(releases); + const closeErrors = releaseResults + .filter( + (result): result is PromiseRejectedResult => + result.status === 'rejected' + ) + .map((result) => result.reason); + try { + await this.enqueue(() => this.closeUnused()); + } catch (error) { + closeErrors.push(error); + } + if (closeErrors.length === 1) throw closeErrors[0]; + if (closeErrors.length > 1) { + throw new PgNotificationBrokerFailedError( + new AggregateError( + closeErrors, + 'Multiple PostgreSQL notification broker close operations failed' + ) + ); + } + } + + async closeIfUnused(): Promise { + await this.enqueue(() => this.closeUnused()); + } + + private enqueue(operation: () => PromiseOrDirect): Promise { + const pending = this.operation.then(operation, operation); + this.operation = pending.then( + (): void => undefined, + (): void => undefined + ); + return pending; + } + + private async ensureClient(): Promise { + if (this.client) return this.client; + try { + this.source = await this.sourcePromise; + if (this.fatalError) throw this.fatalError; + const client = await this.source.connect(); + this.client = client; + client.on('notification', this.onNotification); + client.on('error', this.onClientError); + client.on('end', this.onClientEnd); + this.state = 'active'; + return client; + } catch (error) { + this.markFailed(error); + await this.awaitFailedClientCleanup(); + throw this.fatalError!; + } + } + + private async executeListenerQuery( + client: PgNotificationClient, + text: string + ): Promise { + try { + await this.runWithOperationDeadline( + text.startsWith('UNLISTEN') ? 'unlisten' : 'listen', + () => client.query(text) + ); + if (this.state === 'failed') throw this.fatalError!; + } catch (error) { + this.markFailed(error); + await this.awaitFailedClientCleanup(); + throw this.fatalError!; + } + } + + private async auditRole( + client: PgNotificationClient, + contract: Readonly + ): Promise { + this.counters.roleAuditAttempts++; + try { + const audit = await this.runWithOperationDeadline('role-audit', () => + assertPgNotificationRoleClient( + client as unknown as PgNotificationRoleClient, + contract + ) + ); + if (this.state === 'failed') throw this.fatalError!; + return audit; + } catch (error) { + this.counters.roleAuditFailures++; + this.markFailed(error); + await this.awaitFailedClientCleanup(); + // Preserve the stable unsafe-role error for startup and attestation + // diagnostics. Active leases separately observe the broker-failed latch. + throw error; + } + } + + private async runWithOperationDeadline( + operation: NotificationOperation, + task: () => PromiseOrDirect + ): Promise { + let timer: ReturnType | null = null; + const timeout = new Promise((_resolve, reject) => { + timer = setTimeout(() => { + const error = new PgNotificationOperationTimeoutError( + operation, + this.operationTimeoutMs + ); + // Latch failure and start client destruction at the exact deadline. The + // driver promise remains observed below, so a later rejection is safe. + this.markFailed(error); + reject(error); + }, this.operationTimeoutMs); + timer.unref?.(); + }); + // Promise.race installs a rejection handler on the driver query. If the + // deadline wins, destroying the client may settle that abandoned query + // later without producing an unhandled rejection. + const operationPromise = Promise.resolve().then(task); + try { + return await Promise.race([operationPromise, timeout]); + } finally { + if (timer) clearTimeout(timer); + } + } + + private markFailed(reason: unknown): void { + if ( + this.state === 'failed' || + this.state === 'closing' || + this.state === 'closed' + ) + return; + this.state = 'failed'; + this.fatalError = + reason instanceof PgNotificationBrokerFailedError + ? reason + : new PgNotificationBrokerFailedError(reason); + this.counters.fatalFailures++; + for (const lease of [...this.leases]) lease.fail(this.fatalError); + + const client = this.client; + this.client = null; + if (client) { + this.clientCleanup = this.releaseClient(client, this.fatalError); + // Event-driven failures may not have an immediate waiter. Keep the + // cleanup rejection observed; release/teardown still await and report it. + void this.clientCleanup.catch(() => {}); + } + } + + private async awaitFailedClientCleanup(): Promise { + try { + if (this.clientCleanup) await this.clientCleanup; + } catch (cleanupError) { + throw new PgNotificationBrokerFailedError( + new AggregateError( + [this.fatalError, cleanupError], + 'PostgreSQL notification failure cleanup did not complete safely', + { cause: this.fatalError ?? undefined } + ) + ); + } + } + + private async releaseClient( + client: PgNotificationClient, + error?: Error, + destroy = false + ): Promise { + client.off('notification', this.onNotification); + client.off('end', this.onClientEnd); + try { + await client.release(error ?? (destroy ? true : undefined)); + } catch (releaseError) { + if (error) { + throw new AggregateError( + [error, releaseError], + 'PostgreSQL notification failure and client cleanup both failed', + { cause: error } + ); + } + throw releaseError; + } finally { + client.off('error', this.onClientError); + } + } + + private async closeUnused(): Promise { + if (this.leases.size > 0 || this.state === 'closed') return; + + const cleanupErrors: unknown[] = []; + if (this.client && this.listenedTopics.size > 0) { + try { + // This also covers a shutdown racing between a successful LISTEN and + // lease admission, where no committed topic reference exists yet. + await this.executeListenerQuery(this.client, 'UNLISTEN *'); + this.listenedTopics.clear(); + } catch (error) { + cleanupErrors.push(error); + } + } + if (this.state !== 'failed') this.state = 'closing'; + + const client = this.client; + this.client = null; + if (client) { + const releaseError = + cleanupErrors.length > 0 + ? new PgNotificationBrokerFailedError(cleanupErrors[0]) + : undefined; + // Once the last exact-generation lease is gone, retaining an idle + // listener backend only delays PostgreSQL memory reclamation. Destroy it + // after UNLISTEN; the identity-only pool can create a fresh client later. + this.clientCleanup = this.releaseClient(client, releaseError, true); + } + try { + if (this.clientCleanup) await this.clientCleanup; + } catch (error) { + cleanupErrors.push(error); + } + + if (this.source && !this.sourceCleanup) { + const source = this.source; + this.source = null; + this.sourceCleanup = Promise.resolve(source.release()); + } + try { + if (this.sourceCleanup) await this.sourceCleanup; + } catch (error) { + cleanupErrors.push(error); + } + + this.state = 'closed'; + this.onTerminal(this); + if (cleanupErrors.length === 1) { + const [error] = cleanupErrors; + throw error instanceof PgNotificationBrokerFailedError + ? error + : new PgNotificationBrokerFailedError(error); + } + if (cleanupErrors.length > 1) { + throw new PgNotificationBrokerFailedError( + new AggregateError( + cleanupErrors, + 'Multiple PostgreSQL notification cleanup operations failed' + ) + ); + } + } +} + +/** + * Registry implementation exposed for deterministic unit tests. Production + * callers must use acquirePgNotificationBroker so identity and pool ownership + * always come from the canonical PgConfig path. + * + * @internal + */ +export class PgNotificationBrokerRegistry { + private readonly records = new Map(); + private closed = false; + private closePromise: Promise | null = null; + private readonly counters: MutableBrokerCounters = { + acquisitions: 0, + releases: 0, + notifications: 0, + ignoredNotifications: 0, + queueOverflows: 0, + fatalFailures: 0, + roleAuditAttempts: 0, + roleAuditFailures: 0, + }; + + constructor( + private readonly queueCapacity = PG_NOTIFICATION_QUEUE_CAPACITY, + private readonly defaultOperationTimeoutMs = DEFAULT_PG_NOTIFICATION_OPERATION_TIMEOUT_MS + ) { + if (!Number.isSafeInteger(queueCapacity) || queueCapacity <= 0) { + throw new Error( + 'PostgreSQL notification queue capacity must be a positive safe integer' + ); + } + assertNotificationOperationTimeoutMs(defaultOperationTimeoutMs); + } + + async acquireForTests( + identity: string, + sourceFactory: ConnectionSourceFactory, + topics: readonly string[], + operationTimeoutMs = this.defaultOperationTimeoutMs + ): Promise { + return this.acquireInternal( + identity, + sourceFactory, + topics, + null, + operationTimeoutMs + ); + } + + /** @internal Exercise production attestation without constructing PgConfig. */ + async acquireAttestedForTests( + identity: string, + sourceFactory: ConnectionSourceFactory, + topics: readonly string[], + roleContract: PgNotificationRoleContract, + operationTimeoutMs = this.defaultOperationTimeoutMs + ): Promise { + return this.acquireInternal( + identity, + sourceFactory, + topics, + roleContract, + operationTimeoutMs + ); + } + + private async acquireInternal( + identity: string, + sourceFactory: ConnectionSourceFactory, + topics: readonly string[], + roleContract: PgNotificationRoleContract | null, + operationTimeoutMs: number + ): Promise { + if (this.closed) + throw new Error('PostgreSQL notification broker registry is closed'); + if (typeof identity !== 'string' || identity.length === 0) { + throw new Error( + 'PostgreSQL notification broker identity must be a non-empty string' + ); + } + const normalizedTopics = normalizeTopics(topics); + const normalizedOperationTimeoutMs = + assertNotificationOperationTimeoutMs(operationTimeoutMs); + + for (;;) { + let record = this.records.get(identity); + if (!record) { + const sourcePromise = Promise.resolve(sourceFactory()); + // Acquisition consumes this immediately, but guard the small interval + // before its serialized operation attaches a rejection handler. + void sourcePromise.catch(() => {}); + record = new NotificationBrokerRecord( + identity, + sourcePromise, + this.queueCapacity, + normalizedOperationTimeoutMs, + this.counters, + (terminal) => { + if (this.records.get(identity) === terminal) + this.records.delete(identity); + } + ); + this.records.set(identity, record); + } + try { + const lease = await record.acquire(normalizedTopics, roleContract); + if (this.closed) { + await lease.release(); + throw new Error('PostgreSQL notification broker registry is closed'); + } + return lease; + } catch (error) { + if (error instanceof BrokerClosedError && !this.closed) continue; + // A failed broker remains pinned until every existing owner explicitly + // releases it. This prevents an acquisition attempt from silently + // replacing a listener after a possible notification gap. + await record.closeIfUnused(); + if (error instanceof BrokerClosedError && this.closed) { + throw new Error('PostgreSQL notification broker registry is closed'); + } + throw error; + } + } + } + + stats(): PgNotificationBrokerStats { + let listenerConnections = 0; + let leases = 0; + let topics = 0; + let subscribers = 0; + for (const record of this.records.values()) { + const snapshot = record.snapshot; + listenerConnections += snapshot.listenerConnections; + leases += snapshot.leases; + topics += snapshot.topics; + subscribers += snapshot.subscribers; + } + return { + brokers: this.records.size, + listenerConnections, + leases, + topics, + subscribers, + ...this.counters, + }; + } + + close(): Promise { + if (this.closePromise) return this.closePromise; + this.closed = true; + this.closePromise = (async () => { + const closeResults = await Promise.allSettled( + [...this.records.values()].map((record) => record.closeAll()) + ); + this.records.clear(); + const closeErrors = closeResults + .filter( + (result): result is PromiseRejectedResult => + result.status === 'rejected' + ) + .map((result) => result.reason); + if (closeErrors.length === 1) throw closeErrors[0]; + if (closeErrors.length > 1) { + throw new PgNotificationBrokerFailedError( + new AggregateError( + closeErrors, + 'Multiple PostgreSQL notification registries failed to close' + ) + ); + } + })(); + return this.closePromise; + } +} + +let brokerRegistry = new PgNotificationBrokerRegistry(); +let brokerTeardownTail: Promise = Promise.resolve(); + +/** Opaque identity over the complete canonical listener pool contract. */ +export const getPgNotificationBrokerIdentity = ( + listenerPgConfig: PgNotificationListenerConfig +): string => { + // The operation deadline is represented by the pool connection timeout in + // the identity below. Validate it before publishing an apparently usable key. + getNotificationOperationTimeoutMs(listenerPgConfig); + const poolIdentity = getPgPoolIdentity(listenerPgConfig, { + purpose: 'notification-broker', + }); + return `${PG_NOTIFICATION_BROKER_IDENTITY_VERSION}:${poolIdentity}`; +}; + +/** + * Opaque identity for one physical database target, deliberately excluding + * credentials, TLS policy, pool sizing, and checkout behavior. Those inputs + * split listener pools, but must not let two active listener contracts silently + * fragment one database's broker. + */ +export const getPgNotificationDatabaseIdentity = ( + listenerPgConfig: PgNotificationListenerConfig +): string => { + const targetIdentity = getPgDatabaseTargetIdentity(listenerPgConfig); + return `${PG_NOTIFICATION_DATABASE_IDENTITY_VERSION}:${targetIdentity}`; +}; + +/** + * Acquire a generation lease over one process-local listener. The supplied + * config must name the dedicated least-privilege notification login; this API + * never falls back to a request runtime or control-plane credential. + */ +export const acquirePgNotificationBroker = async ( + listenerPgConfig: PgNotificationListenerConfig, + options: AcquirePgNotificationBrokerOptions +): Promise => { + const operationTimeoutMs = + getNotificationOperationTimeoutMs(listenerPgConfig); + const identity = getPgNotificationBrokerIdentity(listenerPgConfig); + return brokerRegistry.acquireAttestedForTests( + identity, + () => { + const poolLease = acquirePgPool(listenerPgConfig, { + purpose: 'notification-broker', + }); + return { + connect: () => + poolLease.pool.connect() as Promise, + release: () => poolLease.release(), + }; + }, + options.topics, + { + role: listenerPgConfig.user, + database: listenerPgConfig.database, + }, + operationTimeoutMs + ); +}; + +export const getPgNotificationBrokerStats = (): PgNotificationBrokerStats => + brokerRegistry.stats(); + +/** Await every UNLISTEN and checked-out connection release, then reset. */ +export const teardownPgNotificationBrokers = (): Promise => { + const closing = brokerRegistry; + brokerRegistry = new PgNotificationBrokerRegistry(); + const teardown = brokerTeardownTail.then(() => closing.close()); + // A later teardown must wait until this registry has fully drained even when + // this caller observes a cleanup failure. + brokerTeardownTail = teardown.then( + (): void => undefined, + (): void => undefined + ); + return teardown; +}; From 72f55cace2b9a29f48a64fb8aa71268a309a907b Mon Sep 17 00:00:00 2001 From: zetazzz Date: Mon, 17 Aug 2026 16:55:37 +0800 Subject: [PATCH 3/4] Add PostgreSQL notification integration coverage --- .../notification-broker.integration.test.ts | 230 ++++++++++++++++++ .../notification-role.integration.test.ts | 132 ++++++++++ 2 files changed, 362 insertions(+) create mode 100644 postgres/pg-cache/src/__tests__/notification-broker.integration.test.ts create mode 100644 postgres/pg-cache/src/__tests__/notification-role.integration.test.ts diff --git a/postgres/pg-cache/src/__tests__/notification-broker.integration.test.ts b/postgres/pg-cache/src/__tests__/notification-broker.integration.test.ts new file mode 100644 index 0000000000..850f43b688 --- /dev/null +++ b/postgres/pg-cache/src/__tests__/notification-broker.integration.test.ts @@ -0,0 +1,230 @@ +import type pg from 'pg'; +import { getPgEnvOptions, type PgConfig } from 'pg-env'; + +import { teardownPgPools } from '../lru'; +import { + acquirePgNotificationBroker, + getPgNotificationBrokerStats, + PgNotificationTopicError, + teardownPgNotificationBrokers, +} from '../notification-broker'; +import { defaultPgPoolFactory, getPgPool } from '../pg'; + +// Production acquisition always audits the login on its pinned listener, so +// this test requires the dedicated least-privilege notification fixture. +const describeWithPostgres = + process.env.PG_CACHE_RUN_NOTIFICATION_ROLE_INTEGRATION === '1' + ? describe + : describe.skip; + +describeWithPostgres('notification broker against PostgreSQL', () => { + let observerPool: pg.Pool; + let listenerPgConfig: PgConfig & { pool: { max: number } }; + + beforeAll(() => { + listenerPgConfig = { + ...getPgEnvOptions(), + pool: { max: 1 }, + }; + observerPool = defaultPgPoolFactory( + { ...listenerPgConfig, pool: { max: 1 } }, + { purpose: 'notification-broker-integration-observer' } + ) as pg.Pool; + }); + + afterAll(async () => { + await teardownPgNotificationBrokers(); + await teardownPgPools(); + await observerPool?.end(); + }); + + it('shares one LISTEN backend across three isolated generation leases and releases it', async () => { + const nonce = `${process.pid.toString(36)}_${Date.now().toString(36)}`; + const topics = [ + `pg_cache_it_${nonce}_a`, + `pg_cache_it_${nonce}_b`, + `pg_cache_it_${nonce}_c`, + ]; + const listenQueries = topics.map((topic) => `LISTEN "${topic}"`); + + const first = await acquirePgNotificationBroker(listenerPgConfig, { + topics: [topics[0]], + }); + const second = await acquirePgNotificationBroker(listenerPgConfig, { + topics: [topics[1]], + }); + const third = await acquirePgNotificationBroker(listenerPgConfig, { + topics: [topics[2]], + }); + const brokerPool = getPgPool(listenerPgConfig, { + purpose: 'notification-broker', + }); + + expect( + new Set([first.identity, second.identity, third.identity]).size + ).toBe(1); + expect(getPgNotificationBrokerStats()).toMatchObject({ + brokers: 1, + listenerConnections: 1, + leases: 3, + topics: 3, + }); + expect(brokerPool.totalCount).toBe(1); + expect(brokerPool.idleCount).toBe(0); + + const activeListeners = await observerPool.query<{ + pid: number; + query: string; + }>( + ` + SELECT pid, query + FROM pg_stat_activity + WHERE datname = current_database() + AND usename = current_user + AND pid <> pg_backend_pid() + AND query = ANY($1::text[]) + `, + [listenQueries] + ); + expect(activeListeners.rows).toEqual([ + { pid: expect.any(Number), query: listenQueries[2] }, + ]); + const listenerPid = activeListeners.rows[0].pid; + + expect(() => first.subscribe(topics[1])).toThrow(PgNotificationTopicError); + const firstStream = first.subscribe(topics[0]); + const secondStream = second.subscribe(topics[1]); + const thirdStream = third.subscribe(topics[2]); + let firstResolved = false; + let thirdResolved = false; + const firstNext = firstStream.next().then((result) => { + firstResolved = true; + return result; + }); + const secondNext = secondStream.next(); + const thirdNext = thirdStream.next().then((result) => { + thirdResolved = true; + return result; + }); + + await observerPool.query('SELECT pg_notify($1, $2)', [ + topics[1], + 'for-second', + ]); + await expect(secondNext).resolves.toEqual({ + done: false, + value: 'for-second', + }); + // Delivery to every lease happens synchronously inside one notification + // callback, so these flags prove the second topic did not reach its peers. + expect(firstResolved).toBe(false); + expect(thirdResolved).toBe(false); + + await observerPool.query('SELECT pg_notify($1, $2)', [ + topics[0], + 'for-first', + ]); + await observerPool.query('SELECT pg_notify($1, $2)', [ + topics[2], + 'for-third', + ]); + await expect(firstNext).resolves.toEqual({ + done: false, + value: 'for-first', + }); + await expect(thirdNext).resolves.toEqual({ + done: false, + value: 'for-third', + }); + + await second.release(); + await first.release(); + expect(getPgNotificationBrokerStats()).toMatchObject({ + brokers: 1, + listenerConnections: 1, + leases: 1, + topics: 1, + }); + expect(brokerPool.idleCount).toBe(0); + + await third.release(); + expect(getPgNotificationBrokerStats()).toMatchObject({ + brokers: 0, + listenerConnections: 0, + leases: 0, + topics: 0, + }); + expect(brokerPool.totalCount).toBe(0); + expect(brokerPool.idleCount).toBe(0); + + let releasedListenerRows: Array<{ pid: number }> = []; + for (let attempt = 0; attempt < 50; attempt++) { + const releasedListener = await observerPool.query<{ pid: number }>( + ` + SELECT pid + FROM pg_stat_activity + WHERE pid = $1 + `, + [listenerPid] + ); + releasedListenerRows = releasedListener.rows; + if (releasedListenerRows.length === 0) break; + await new Promise((resolve) => setTimeout(resolve, 20)); + } + expect(releasedListenerRows).toEqual([]); + }); + + it('replaces a released generation with a new backend and no old topic owner', async () => { + const nonce = `${process.pid.toString(36)}_${Date.now().toString(36)}`; + const oldTopic = `pg_cache_generation_${nonce}_old`; + const newTopic = `pg_cache_generation_${nonce}_new`; + + const oldLease = await acquirePgNotificationBroker(listenerPgConfig, { + topics: [oldTopic], + }); + const oldListener = await observerPool.query<{ pid: number }>( + ` + SELECT pid + FROM pg_stat_activity + WHERE datname = current_database() + AND usename = current_user + AND pid <> pg_backend_pid() + AND query = $1 + `, + [`LISTEN "${oldTopic}"`] + ); + expect(oldListener.rows).toHaveLength(1); + const oldPid = oldListener.rows[0].pid; + + await oldLease.release(); + const newLease = await acquirePgNotificationBroker(listenerPgConfig, { + topics: [newTopic], + }); + const newListener = await observerPool.query<{ pid: number }>( + ` + SELECT pid + FROM pg_stat_activity + WHERE datname = current_database() + AND usename = current_user + AND pid <> pg_backend_pid() + AND query = $1 + `, + [`LISTEN "${newTopic}"`] + ); + expect(newListener.rows).toHaveLength(1); + expect(newListener.rows[0].pid).not.toBe(oldPid); + + const next = newLease.subscribe(newTopic).next(); + await observerPool.query('SELECT pg_notify($1, $2)', [oldTopic, 'stale']); + await observerPool.query('SELECT pg_notify($1, $2)', [newTopic, 'current']); + await expect(next).resolves.toEqual({ done: false, value: 'current' }); + + await newLease.release(); + expect(getPgNotificationBrokerStats()).toMatchObject({ + brokers: 0, + listenerConnections: 0, + leases: 0, + topics: 0, + }); + }); +}); diff --git a/postgres/pg-cache/src/__tests__/notification-role.integration.test.ts b/postgres/pg-cache/src/__tests__/notification-role.integration.test.ts new file mode 100644 index 0000000000..94a89f9657 --- /dev/null +++ b/postgres/pg-cache/src/__tests__/notification-role.integration.test.ts @@ -0,0 +1,132 @@ +import type pg from 'pg'; +import { getPgEnvOptions } from 'pg-env'; + +import { teardownPgPools } from '../lru'; +import { + acquirePgNotificationBroker, + getPgNotificationBrokerStats, + teardownPgNotificationBrokers, +} from '../notification-broker'; +import { + assertPgNotificationRole, + auditPgNotificationRole, +} from '../notification-role'; +import { defaultPgPoolFactory, getPgPool } from '../pg'; + +const describeWithNotificationRole = + process.env.PG_CACHE_RUN_NOTIFICATION_ROLE_INTEGRATION === '1' + ? describe + : describe.skip; + +describeWithNotificationRole( + 'dedicated notification role against PostgreSQL', + () => { + const pgConfig = getPgEnvOptions(); + let pool: pg.Pool; + + beforeAll(() => { + pool = defaultPgPoolFactory( + { ...pgConfig, pool: { max: 1 } }, + { purpose: 'notification-role-integration' } + ) as pg.Pool; + }); + + afterAll(async () => { + await teardownPgNotificationBrokers(); + await teardownPgPools(); + await pool?.end(); + }); + + it('accepts only the exact credential-free role/database contract', async () => { + const audit = await assertPgNotificationRole(pool, { + role: pgConfig.user, + database: pgConfig.database, + }); + + expect(audit).toMatchObject({ + role: pgConfig.user, + database: pgConfig.database, + safe: true, + violations: [], + }); + expect(Object.keys(audit).sort()).toEqual([ + 'database', + 'role', + 'safe', + 'version', + 'violations', + ]); + expect(audit).not.toHaveProperty('password'); + expect(audit).not.toHaveProperty('host'); + + const wrongRole = await auditPgNotificationRole(pool, { + role: `wrong_${process.pid}`, + database: pgConfig.database, + }); + expect(wrongRole).toMatchObject({ + safe: false, + violations: expect.arrayContaining(['LOGIN_ROLE_MISMATCH']), + }); + + const wrongDatabase = await auditPgNotificationRole(pool, { + role: pgConfig.user, + database: `wrong_${process.pid}`, + }); + expect(wrongDatabase).toMatchObject({ + safe: false, + violations: expect.arrayContaining([ + 'DATABASE_MISMATCH', + 'TARGET_DATABASE_MISSING', + 'TARGET_CONNECT_REQUIRED', + 'CROSS_DATABASE_CONNECT', + ]), + }); + }); + + it('retains enough privilege for isolated LISTEN and NOTIFY delivery', async () => { + const nonce = `${process.pid}_${Date.now().toString(36)}`; + const topics = [0, 1, 2].map( + (index) => `notify_role_it_${nonce}_${index}` + ); + const listenerConfig = { ...pgConfig, pool: { max: 1 } }; + const statsBefore = getPgNotificationBrokerStats(); + const [first, second, third] = await Promise.all( + topics.map((topic) => + acquirePgNotificationBroker(listenerConfig, { topics: [topic] }) + ) + ); + const brokerPool = getPgPool(listenerConfig, { + purpose: 'notification-broker', + }); + await first.revalidateRole(); + const next = second.subscribe(topics[1]).next(); + + await pool.query('SELECT pg_notify($1, $2)', [ + topics[1], + 'safe-listener', + ]); + await expect(next).resolves.toEqual({ + done: false, + value: 'safe-listener', + }); + expect(getPgNotificationBrokerStats()).toMatchObject({ + brokers: 1, + listenerConnections: 1, + leases: 3, + topics: 3, + roleAuditAttempts: statsBefore.roleAuditAttempts + 4, + roleAuditFailures: statsBefore.roleAuditFailures, + }); + expect(brokerPool.totalCount).toBe(1); + expect(brokerPool.idleCount).toBe(0); + + await Promise.all([first.release(), second.release(), third.release()]); + expect(getPgNotificationBrokerStats()).toMatchObject({ + brokers: 0, + listenerConnections: 0, + leases: 0, + topics: 0, + }); + }); + } +); From c95de1e013a3e1b47e25c3a2747ce931326f3e65 Mon Sep 17 00:00:00 2001 From: zetazzz Date: Wed, 9 Sep 2026 04:33:56 +0000 Subject: [PATCH 4/4] test(pg-cache): run attested notification broker integration in CI --- .github/workflows/run-tests.yaml | 8 ++++++++ postgres/pg-cache/README.md | 4 ++++ .../src/__tests__/fixtures/notification-role.sql | 12 ++++++++++++ 3 files changed, 24 insertions(+) create mode 100644 postgres/pg-cache/src/__tests__/fixtures/notification-role.sql diff --git a/.github/workflows/run-tests.yaml b/.github/workflows/run-tests.yaml index a6fb3b4739..ece96f4ed5 100644 --- a/.github/workflows/run-tests.yaml +++ b/.github/workflows/run-tests.yaml @@ -332,6 +332,14 @@ jobs: echo "::endgroup::" done + # Run last: the strict listener fixture revokes PUBLIC cross-database + # CONNECT in this disposable service container. + - name: Test dedicated notification role and broker + if: matrix.batch == 'pg-postgres' + run: | + docker exec -i -e PGUSER=postgres ${{ job.services.pg_db.id }} psql -d postgres -v ON_ERROR_STOP=1 < postgres/pg-cache/src/__tests__/fixtures/notification-role.sql + PGUSER=cnc_notify_fixture PGPASSWORD=pg-cache-notify-test PGDATABASE=cnc_notify_fixture PG_CACHE_RUN_NOTIFICATION_ROLE_INTEGRATION=1 pnpm --filter pg-cache test --runInBand --runTestsByPath src/__tests__/notification-role.integration.test.ts src/__tests__/notification-broker.integration.test.ts + # ========================================================================= # TIER 3 – Integration tests (PostgreSQL + MinIO) # ========================================================================= diff --git a/postgres/pg-cache/README.md b/postgres/pg-cache/README.md index 8df67bc703..9394d65f52 100644 --- a/postgres/pg-cache/README.md +++ b/postgres/pg-cache/README.md @@ -148,3 +148,7 @@ This package is designed to be extended. For example, `graphile-cache` uses the ### 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. + +### Dedicated notification listener validation + +The broker accepts only an audited, dedicated login and exact channel allowlists. Its two real PostgreSQL integration suites require `PG_CACHE_RUN_NOTIFICATION_ROLE_INTEGRATION=1`. CI provisions `src/__tests__/fixtures/notification-role.sql` and runs both suites after the `pg-postgres` batch. Run this fixture only in a disposable cluster: it revokes PUBLIC cross-database CONNECT to test the listener isolation contract. The normal unit run keeps these environment-specific suites gated. diff --git a/postgres/pg-cache/src/__tests__/fixtures/notification-role.sql b/postgres/pg-cache/src/__tests__/fixtures/notification-role.sql new file mode 100644 index 0000000000..efbada8fe3 --- /dev/null +++ b/postgres/pg-cache/src/__tests__/fixtures/notification-role.sql @@ -0,0 +1,12 @@ +-- Run only in a disposable test PostgreSQL cluster, after other tests finish. +-- The listener contract forbids CONNECT to every database except its target. +CREATE ROLE cnc_notify_fixture LOGIN NOINHERIT NOSUPERUSER NOBYPASSRLS + NOCREATEDB NOCREATEROLE NOREPLICATION PASSWORD 'pg-cache-notify-test'; +CREATE DATABASE cnc_notify_fixture TEMPLATE template0; +SELECT format('REVOKE CONNECT ON DATABASE %I FROM PUBLIC', datname) +FROM pg_database +\gexec +REVOKE ALL ON DATABASE cnc_notify_fixture FROM PUBLIC; +GRANT CONNECT ON DATABASE cnc_notify_fixture TO cnc_notify_fixture; +\connect cnc_notify_fixture +REVOKE ALL ON SCHEMA public FROM PUBLIC;