Add attested PostgreSQL notification broker lifecycle

.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)
   # =========================================================================

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.

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;

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