From 269b23c2ff82b0b6295cb1822f5affddfbc511cb Mon Sep 17 00:00:00 2001 From: yyyyaaa Date: Mon, 6 Jul 2026 17:16:50 +0700 Subject: [PATCH] =?UTF-8?q?feat(server):=20blueprint=20pooling=20core=20?= =?UTF-8?q?=E2=80=94=20qualified-SQL=20rewrite=20seam=20(inert=20modules)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The mechanism layer for sharing one PostGraphile instance per schema-SHAPE. Nothing imports these modules yet — this PR is pure, unit-tested logic; the dispatcher wiring (and the opt-in flag becoming reachable) lands in the next PR of the stack. No search_path anywhere: instances are built fully qualified against a canonical tenant, and per-request tenant routing happens by rewriting schema IDENTIFIERS in SQL text. blueprint.ts — shared-instance identity: - stripSchemaHashPrefix: physical `-<8hex>-` → logical name. - Shape fingerprint: sha256 over sorted [logicalSchema, relname] pairs from one qualified catalog scan. - computeBlueprintKey: 'bp:' + sha256({schemas, fingerprint, flags, apiName, mode, dbname}) — dbname included so same-shape tenants in different physical databases NEVER share an instance. - hasBm25Indexes: catalog probe backing the bm25 structural opt-out below. pooling-decision.ts — per-service decision, memoized per svc_key: - Structural opt-outs (memoized until flush): realtime instances, empty schema lists, and shapes carrying bm25 indexes — the bm25 adapter embeds its build-time index name as a bind VALUE, which the seam deliberately never rewrites (answers the #1325 review question on bm25.ts; the adapter itself is fully qualified, but pooled routing can't reach a value). - Probe failures are TRANSIENT: not memoized, so one flaky catalog query can't pin a tenant to a dedicated instance until the next flush. - Per-database invalidation (clearPoolDecisionsForDatabase) so a schema change in tenant X rebuilds only X's blueprint, not the fleet. rewrite-pool.ts — the pool seam: - Wraps the pg Pool; a checkout learns its tenant from the transaction-local GUC constructive.pool_schemas (set via the adaptor's set_config statement, classified by EXACT statement text — SETTINGS_QUERY_TEXT, byte-pinned against the installed @dataplan/pg by settings-text-pin.test.ts, so a dependency bump fails CI instead of silently disabling routing). - Rewrites double-quoted canonical schema identifiers in SQL text to the requesting tenant's schemas (lexer-based: string literals, dollar-quotes, comments are opaque). Bind values are never touched. - Prepared-statement names namespaced per tenant (sha256-prefixed) with per-connection DEALLOCATE LRU; memoized rewrite/statement caches (2000-entry LRUs — 98.4% hit rate in the #1325 soak). - FAIL-CLOSED: a checkout with no tenant mapping refuses tenant-schema SQL rather than letting it run against the canonical tenant. - Composes the introspection filter (previous PR) for pooled instances. Also: ApiStructure gains optional `logicalSchemas` (consumed by the keying), and the metrics sampler reports rewrite/fail-closed counters. Isolation evidence from #1325: 13/13 scope-pollution suites clean, zero cross-tenant bleed over 104,818 authenticated requests, SDL byte-identical (MD5) between pooled and dedicated builds of the same tenant. Extracted from #1325 (commits 445d4c3a8, e38f4b52d, b3bad9e17, 308806201). Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_0122xqM2VkNbuAmZshK1YNSb --- graphql/server/package.json | 1 + .../__tests__/metrics-sampler.test.ts | 1 + .../server/src/diagnostics/metrics-sampler.ts | 3 + .../middleware/__tests__/blueprint.test.ts | 149 ++++ .../__tests__/graphile-pooling.test.ts | 274 +++++++ .../__tests__/introspection-filter.test.ts | 76 ++ .../middleware/__tests__/rewrite-pool.test.ts | 637 ++++++++++++++++ .../__tests__/settings-text-pin.test.ts | 24 + graphql/server/src/middleware/blueprint.ts | 159 ++++ .../server/src/middleware/pooling-decision.ts | 162 ++++ graphql/server/src/middleware/rewrite-pool.ts | 696 ++++++++++++++++++ packages/express-context/src/types.ts | 6 + pnpm-lock.yaml | 3 + 13 files changed, 2191 insertions(+) create mode 100644 graphql/server/src/middleware/__tests__/blueprint.test.ts create mode 100644 graphql/server/src/middleware/__tests__/graphile-pooling.test.ts create mode 100644 graphql/server/src/middleware/__tests__/rewrite-pool.test.ts create mode 100644 graphql/server/src/middleware/__tests__/settings-text-pin.test.ts create mode 100644 graphql/server/src/middleware/blueprint.ts create mode 100644 graphql/server/src/middleware/pooling-decision.ts create mode 100644 graphql/server/src/middleware/rewrite-pool.ts diff --git a/graphql/server/package.json b/graphql/server/package.json index 3f6926dfa..274bbebb6 100644 --- a/graphql/server/package.json +++ b/graphql/server/package.json @@ -78,6 +78,7 @@ }, "devDependencies": { "@aws-sdk/client-s3": "^3.1052.0", + "@dataplan/pg": "1.0.3", "@types/cookie-parser": "^1.4.10", "@types/cors": "^2.8.17", "@types/express": "^5.0.6", diff --git a/graphql/server/src/diagnostics/__tests__/metrics-sampler.test.ts b/graphql/server/src/diagnostics/__tests__/metrics-sampler.test.ts index ee21ede77..18154b493 100644 --- a/graphql/server/src/diagnostics/__tests__/metrics-sampler.test.ts +++ b/graphql/server/src/diagnostics/__tests__/metrics-sampler.test.ts @@ -47,6 +47,7 @@ describe('collectMetricsSample', () => { expect(typeof sample.counters.evictions.lru).toBe('number'); expect(typeof sample.counters.builds).toBe('number'); expect(typeof sample.counters.buildQueueDepth).toBe('number'); + expect(typeof sample.counters.rewritePool.rewrittenQueries).toBe('number'); expect(typeof sample.counters.introspectionFilter.swaps).toBe('number'); expect(typeof sample.counters.introspectionFilter.discoveries).toBe('number'); expect(typeof sample.gc).toBe('object'); diff --git a/graphql/server/src/diagnostics/metrics-sampler.ts b/graphql/server/src/diagnostics/metrics-sampler.ts index 792dfb588..dceab4655 100644 --- a/graphql/server/src/diagnostics/metrics-sampler.ts +++ b/graphql/server/src/diagnostics/metrics-sampler.ts @@ -6,6 +6,7 @@ import { getCacheCounters, getCacheStats } from 'graphile-cache'; import { getBuildQueueDepth, getGraphileCounters } from '../middleware/graphile'; import { getIntrospectionFilterCounters } from '../middleware/introspection-filter'; +import { getRewritePoolCounters } from '../middleware/rewrite-pool'; import { getConnectionErrorGuardCounters } from './connection-error-guard'; const log = new Logger('metrics-sampler'); @@ -127,6 +128,7 @@ export interface MetricsSample { ReturnType & { buildQueueDepth: number; connGuard: ReturnType; + rewritePool: ReturnType; introspectionFilter: ReturnType; }; gc: GcStats; @@ -156,6 +158,7 @@ export const collectMetricsSample = (): MetricsSample => { ...getGraphileCounters(), buildQueueDepth: getBuildQueueDepth(), connGuard: getConnectionErrorGuardCounters(), + rewritePool: getRewritePoolCounters(), introspectionFilter: getIntrospectionFilterCounters() }, gc: snapshotGcStats() diff --git a/graphql/server/src/middleware/__tests__/blueprint.test.ts b/graphql/server/src/middleware/__tests__/blueprint.test.ts new file mode 100644 index 000000000..80e190a73 --- /dev/null +++ b/graphql/server/src/middleware/__tests__/blueprint.test.ts @@ -0,0 +1,149 @@ +import { + computeBlueprintKey, + hasBm25Indexes, + stripSchemaHashPrefix +} from '../blueprint'; + +describe('hasBm25Indexes', () => { + it('is true when the catalog reports a bm25 index in the served schemas', async () => { + const query = jest.fn().mockResolvedValueOnce({ rows: [{ exists: true }] }); + await expect(hasBm25Indexes({ query } as any, ['s1', 's2'])).resolves.toBe(true); + // The probe is a single qualified catalog query over the served schemas. + expect(query).toHaveBeenCalledTimes(1); + expect(query).toHaveBeenCalledWith(expect.stringContaining("am.amname = 'bm25'"), [['s1', 's2']]); + }); + + it('is false when the catalog reports none (or an empty result)', async () => { + const q1 = jest.fn().mockResolvedValueOnce({ rows: [{ exists: false }] }); + await expect(hasBm25Indexes({ query: q1 } as any, ['s1'])).resolves.toBe(false); + const q2 = jest.fn().mockResolvedValueOnce({ rows: [] }); + await expect(hasBm25Indexes({ query: q2 } as any, ['s1'])).resolves.toBe(false); + }); +}); + +describe('stripSchemaHashPrefix', () => { + it('strips a hashed tenant prefix down to the logical schema', () => { + expect(stripSchemaHashPrefix('marketplace-db-tenant1-5e6b13b2-app-public')).toBe( + 'app-public' + ); + }); + + it('returns control-plane / non-hashed schema names unchanged', () => { + expect(stripSchemaHashPrefix('services_public')).toBe('services_public'); + expect(stripSchemaHashPrefix('metaschema_public')).toBe('metaschema_public'); + expect(stripSchemaHashPrefix('app-public')).toBe('app-public'); + }); + + it('handles multi-dash database names in the prefix', () => { + expect(stripSchemaHashPrefix('my-cool-db-name-deadbeef-auth-private')).toBe( + 'auth-private' + ); + expect(stripSchemaHashPrefix('app-5e6b13b2-public')).toBe('public'); + }); + + it('only strips through the FIRST 8-hex segment, keeping later ones intact', () => { + expect( + stripSchemaHashPrefix('marketplace-db-tenant1-5e6b13b2-app-deadbeef-zone') + ).toBe('app-deadbeef-zone'); + }); + + it('does not treat non-hex or wrong-length segments as a hash', () => { + // 'tenant11' is 8 chars but not all hex; 'abcdef1' is 7 hex chars. + expect(stripSchemaHashPrefix('marketplace-tenant11-app-public')).toBe( + 'marketplace-tenant11-app-public' + ); + expect(stripSchemaHashPrefix('marketplace-abcdef1-app-public')).toBe( + 'marketplace-abcdef1-app-public' + ); + }); +}); + +describe('computeBlueprintKey', () => { + const base = { + logicalSchemas: ['app-public', 'auth-public'], + shapeFingerprint: 'fingerprint-a', + flags: { a: 1, b: 2 } as Record, + apiName: 'customer-api', + mode: 'domain-lookup' + }; + + it('produces a stable "bp:"-prefixed sha256 hex key', () => { + const key = computeBlueprintKey(base); + expect(key).toMatch(/^bp:[0-9a-f]{64}$/); + expect(computeBlueprintKey(base)).toBe(key); + }); + + it('is stable across the key order of flags', () => { + const key1 = computeBlueprintKey({ ...base, flags: { a: 1, b: 2 } }); + const key2 = computeBlueprintKey({ ...base, flags: { b: 2, a: 1 } }); + expect(key1).toBe(key2); + }); + + it('is stable across the order of logical schemas', () => { + const key1 = computeBlueprintKey({ + ...base, + logicalSchemas: ['app-public', 'auth-public'] + }); + const key2 = computeBlueprintKey({ + ...base, + logicalSchemas: ['auth-public', 'app-public'] + }); + expect(key1).toBe(key2); + }); + + it('differs when the shape fingerprint differs', () => { + const key1 = computeBlueprintKey({ ...base, shapeFingerprint: 'fingerprint-a' }); + const key2 = computeBlueprintKey({ ...base, shapeFingerprint: 'fingerprint-b' }); + expect(key1).not.toBe(key2); + }); + + it('differs when flag values differ', () => { + const key1 = computeBlueprintKey({ ...base, flags: { a: 1, b: 2 } }); + const key2 = computeBlueprintKey({ ...base, flags: { a: 1, b: 3 } }); + expect(key1).not.toBe(key2); + }); + + it('treats undefined flags like an empty flag set', () => { + const key1 = computeBlueprintKey({ ...base, flags: undefined }); + const key2 = computeBlueprintKey({ ...base, flags: {} }); + expect(key1).toBe(key2); + }); + + it('differs when the mode or api name differs', () => { + expect(computeBlueprintKey({ ...base, mode: 'domain-lookup' })).not.toBe( + computeBlueprintKey({ ...base, mode: 'api-name-header' }) + ); + expect(computeBlueprintKey({ ...base, apiName: 'customer-api' })).not.toBe( + computeBlueprintKey({ ...base, apiName: 'other-api' }) + ); + }); + + it('treats null and empty api name identically', () => { + expect(computeBlueprintKey({ ...base, apiName: null })).toBe( + computeBlueprintKey({ ...base, apiName: '' }) + ); + }); +}); + +describe('computeBlueprintKey dbname isolation (W3 fix)', () => { + const { computeBlueprintKey } = require('../blueprint'); + const base = { + logicalSchemas: ['app-public'], + shapeFingerprint: 'f'.repeat(64), + flags: undefined as Record | undefined, + apiName: 'api', + mode: 'public' + }; + + it('same-shape tenants in DIFFERENT physical databases get DIFFERENT keys', () => { + const a = computeBlueprintKey({ ...base, dbname: 'db_a' }); + const b = computeBlueprintKey({ ...base, dbname: 'db_b' }); + expect(a).not.toBe(b); + }); + + it('same dbname yields a stable key', () => { + expect(computeBlueprintKey({ ...base, dbname: 'db_a' })).toBe( + computeBlueprintKey({ ...base, dbname: 'db_a' }) + ); + }); +}); diff --git a/graphql/server/src/middleware/__tests__/graphile-pooling.test.ts b/graphql/server/src/middleware/__tests__/graphile-pooling.test.ts new file mode 100644 index 000000000..cd9b7f5c0 --- /dev/null +++ b/graphql/server/src/middleware/__tests__/graphile-pooling.test.ts @@ -0,0 +1,274 @@ +import type { Pool } from 'pg'; + +import type { ApiStructure } from '../../types'; +import { isBlueprintPoolingEnabled } from '../blueprint'; +import { clearPoolDecisions, computePoolDecision, resolvePoolDecision } from '../pooling-decision'; + +type QueryResult = { rows: Array> }; + +const makeApi = (overrides: Partial = {}): ApiStructure => + ({ + dbname: 'constructive', + anonRole: 'anon', + roleName: 'authenticated', + schema: ['marketplace-db-tenant1-5e6b13b2-app-public'], + apiModules: [], + ...overrides + } as ApiStructure); + +// A mock Pool whose query() resolves queued results in call order. computePoolDecision +// issues a single shape-fingerprint query. +const makePool = (results: QueryResult[]): { pool: Pool; query: jest.Mock } => { + const query = jest.fn(); + results.forEach((r) => query.mockResolvedValueOnce(r)); + return { pool: { query } as unknown as Pool, query }; +}; + +describe('computePoolDecision', () => { + afterEach(() => { + clearPoolDecisions(); + jest.clearAllMocks(); + }); + + it('falls back (pooling=false, key=svc_key) when realtime is enabled, without touching the DB', async () => { + const { pool, query } = makePool([]); + const api = makeApi({ databaseSettings: { enableRealtime: true } as ApiStructure['databaseSettings'] }); + + const decision = await computePoolDecision('svc-1', api, pool); + + expect(decision).toEqual({ key: 'svc-1', pooling: false }); + expect(query).not.toHaveBeenCalled(); + }); + + it('falls back when the API exposes no schemas, without touching the DB', async () => { + const { pool, query } = makePool([]); + const api = makeApi({ schema: [] }); + + const decision = await computePoolDecision('svc-2', api, pool); + + expect(decision).toEqual({ key: 'svc-2', pooling: false }); + expect(query).not.toHaveBeenCalled(); + }); + + it('pools a clean single-tenant shape and returns a stable bp: key', async () => { + const rows = [ + { nspname: 'marketplace-db-tenant1-5e6b13b2-app-public', relname: 'products' }, + { nspname: 'marketplace-db-tenant1-5e6b13b2-app-public', relname: 'orders' } + ]; + const { pool, query } = makePool([{ rows }, { rows: [] }]); + const api = makeApi(); + + const decision = await computePoolDecision('svc-3', api, pool); + + expect(decision.pooling).toBe(true); + expect(decision.key).toMatch(/^bp:[0-9a-f]{64}$/); + expect(query).toHaveBeenCalledTimes(2); // shape scan + bm25 probe, nothing else + }); + + it('produces the SAME bp: key for two different physical tenants of the same shape', async () => { + const shapeRows = (hash: string) => [ + { nspname: `marketplace-db-tenant-${hash}-app-public`, relname: 'orders' }, + { nspname: `marketplace-db-tenant-${hash}-app-public`, relname: 'products' } + ]; + + const t1 = makePool([{ rows: shapeRows('5e6b13b2') }, { rows: [] }]); + const d1 = await computePoolDecision( + 'svc-t1', + makeApi({ schema: ['marketplace-db-tenant-5e6b13b2-app-public'] }), + t1.pool + ); + + const t2 = makePool([{ rows: shapeRows('deadbeef') }, { rows: [] }]); + const d2 = await computePoolDecision( + 'svc-t2', + makeApi({ schema: ['marketplace-db-tenant-deadbeef-app-public'] }), + t2.pool + ); + + expect(d1.pooling).toBe(true); + expect(d2.pooling).toBe(true); + expect(d1.key).toBe(d2.key); // same logical shape → shared blueprint instance + }); + + it('threads api.logicalSchemas into the key, deriving it from physical schema only when absent', async () => { + // Identical fingerprint rows across all three calls: only logicalSchemas varies, + // so any key difference is attributable solely to the logicalSchemas input. + const fpRows = [{ nspname: 'shop-5e6b13b2-app-public', relname: 'orders' }]; + + // Explicit logicalSchemas that DIFFER from the stripped physical name. + const a = makePool([{ rows: fpRows }, { rows: [] }]); + const dExplicitDiff = await computePoolDecision( + 'svc-a', + makeApi({ schema: ['shop-5e6b13b2-app-public'], logicalSchemas: ['custom-logical'] }), + a.pool + ); + + // No logicalSchemas → derived from schema.map(stripSchemaHashPrefix) = ['app-public']. + const b = makePool([{ rows: fpRows }, { rows: [] }]); + const dDerived = await computePoolDecision( + 'svc-b', + makeApi({ schema: ['shop-5e6b13b2-app-public'], logicalSchemas: undefined }), + b.pool + ); + + // Explicit logicalSchemas EQUAL to the derived value. + const c = makePool([{ rows: fpRows }, { rows: [] }]); + const dExplicitSame = await computePoolDecision( + 'svc-c', + makeApi({ schema: ['shop-5e6b13b2-app-public'], logicalSchemas: ['app-public'] }), + c.pool + ); + + expect(dExplicitDiff.key).not.toBe(dDerived.key); // provided logicalSchemas is honored + expect(dExplicitSame.key).toBe(dDerived.key); // ?? fallback derivation matches + }); + + it('relation-name collisions across schemas are poolable under qualified SQL', async () => { + const rows = [ + { nspname: 'shop-5e6b13b2-auth-public', relname: 'identity_providers' }, + { nspname: 'shop-5e6b13b2-auth-private', relname: 'identity_providers' } + ]; + const { pool } = makePool([{ rows }, { rows: [] }]); + const api = makeApi({ schema: ['shop-5e6b13b2-auth-public', 'shop-5e6b13b2-auth-private'] }); + + const decision = await computePoolDecision('svc-4', api, pool); + + expect(decision.pooling).toBe(true); + expect(decision.key).toMatch(/^bp:[0-9a-f]{64}$/); + }); + + it('falls back (never throws) when a catalog probe rejects', async () => { + const query = jest.fn().mockRejectedValue(new Error('connection reset')); + const api = makeApi(); + + const decision = await computePoolDecision('svc-5', api, { query } as unknown as Pool); + + expect(decision).toEqual({ key: 'svc-5', pooling: false, transient: true }); + }); +}); + +describe('resolvePoolDecision caching', () => { + afterEach(() => { + clearPoolDecisions(); + jest.clearAllMocks(); + }); + + it('memoizes the decision per svc_key (one probe pair) until cleared', async () => { + const { pool, query } = makePool([ + { rows: [{ nspname: 'shop-5e6b13b2-app-public', relname: 'orders' }] }, + { rows: [] } + ]); + const api = makeApi({ schema: ['shop-5e6b13b2-app-public'] }); + + const first = await resolvePoolDecision('svc-cache', api, pool); + const second = await resolvePoolDecision('svc-cache', api, pool); + + expect(first).toBe(second); // same cached object identity + expect(first.pooling).toBe(true); + expect(query).toHaveBeenCalledTimes(2); // one probe pair — second call served from cache + + // clearPoolDecisions() forces the next request to re-probe (a fresh pair). + clearPoolDecisions(); + query.mockResolvedValueOnce({ rows: [{ nspname: 'shop-5e6b13b2-app-public', relname: 'orders' }] }); + query.mockResolvedValueOnce({ rows: [] }); + await resolvePoolDecision('svc-cache', api, pool); + expect(query).toHaveBeenCalledTimes(4); + }); +}); + +describe('blueprint pooling flag gate (flag-off ⇒ key stays svc_key in the dispatcher)', () => { + const original = process.env.GRAPHILE_BLUEPRINT_POOLING; + + afterEach(() => { + if (original === undefined) delete process.env.GRAPHILE_BLUEPRINT_POOLING; + else process.env.GRAPHILE_BLUEPRINT_POOLING = original; + }); + + it('is disabled by default — the dispatcher never runs a decision and uses req.svc_key', () => { + delete process.env.GRAPHILE_BLUEPRINT_POOLING; + expect(isBlueprintPoolingEnabled()).toBe(false); + }); + + it("enables only on '1' or 'true'", () => { + process.env.GRAPHILE_BLUEPRINT_POOLING = '1'; + expect(isBlueprintPoolingEnabled()).toBe(true); + process.env.GRAPHILE_BLUEPRINT_POOLING = 'true'; + expect(isBlueprintPoolingEnabled()).toBe(true); + process.env.GRAPHILE_BLUEPRINT_POOLING = 'yes'; + expect(isBlueprintPoolingEnabled()).toBe(false); + }); +}); + +describe('transient probe failures are not memoized (W3 fix)', () => { + afterEach(() => { + clearPoolDecisions(); + jest.clearAllMocks(); + }); + + it('re-probes on the next request after a thrown catalog probe', async () => { + const query = jest.fn(); + query.mockRejectedValueOnce(new Error('connection reset')); + query.mockResolvedValueOnce({ + rows: [{ nspname: 'shop-5e6b13b2-app-public', relname: 'orders' }] + }); + query.mockResolvedValueOnce({ rows: [] }); + const pool = { query } as unknown as Pool; + const api = makeApi({ schema: ['shop-5e6b13b2-app-public'] }); + + const first = await resolvePoolDecision('svc-transient', api, pool); + expect(first.pooling).toBe(false); + expect(first.transient).toBe(true); + + const second = await resolvePoolDecision('svc-transient', api, pool); + expect(second.pooling).toBe(true); // re-probed and now poolable + expect(query).toHaveBeenCalledTimes(3); // failed probe, then a full pair + }); +}); + +describe('bm25 structural opt-out', () => { + afterEach(() => { + clearPoolDecisions(); + jest.clearAllMocks(); + }); + + it('does not pool a shape that carries bm25 indexes (memoized structural opt-out)', async () => { + const { pool, query } = makePool([ + { rows: [{ nspname: 'shop-5e6b13b2-app-public', relname: 'documents' }] }, + { rows: [{ exists: true }] } + ]); + const api = makeApi({ schema: ['shop-5e6b13b2-app-public'] }); + + const decision = await resolvePoolDecision('svc-bm25', api, pool); + expect(decision.pooling).toBe(false); + expect(decision.key).toBe('svc-bm25'); + expect(decision.transient).toBeUndefined(); // structural: memoized until flush + + const second = await resolvePoolDecision('svc-bm25', api, pool); + expect(second).toBe(decision); // cached — no re-probe + expect(query).toHaveBeenCalledTimes(2); + }); + + it('pools when the bm25 probe finds no bm25 indexes', async () => { + const { pool } = makePool([ + { rows: [{ nspname: 'shop-5e6b13b2-app-public', relname: 'orders' }] }, + { rows: [{ exists: false }] } + ]); + const api = makeApi({ schema: ['shop-5e6b13b2-app-public'] }); + + const decision = await computePoolDecision('svc-nobm25', api, pool); + expect(decision.pooling).toBe(true); + expect(decision.key).toMatch(/^bp:[0-9a-f]{64}$/); + }); + + it('a thrown bm25 probe falls back transiently (same contract as the shape probe)', async () => { + const query = jest.fn(); + query.mockResolvedValueOnce({ + rows: [{ nspname: 'shop-5e6b13b2-app-public', relname: 'orders' }] + }); + query.mockRejectedValueOnce(new Error('connection reset')); + const api = makeApi({ schema: ['shop-5e6b13b2-app-public'] }); + + const decision = await computePoolDecision('svc-bm25-err', api, { query } as unknown as Pool); + expect(decision).toEqual({ key: 'svc-bm25-err', pooling: false, transient: true }); + }); +}); diff --git a/graphql/server/src/middleware/__tests__/introspection-filter.test.ts b/graphql/server/src/middleware/__tests__/introspection-filter.test.ts index bcaef4eef..0c13f4a96 100644 --- a/graphql/server/src/middleware/__tests__/introspection-filter.test.ts +++ b/graphql/server/src/middleware/__tests__/introspection-filter.test.ts @@ -9,6 +9,11 @@ import { isIntrospectionFilterEnabled, isIntrospectionQuery } from '../introspection-filter'; +import { + createRewriteCounters, + createRewritingPool, + POOL_SCHEMAS_GUC +} from '../rewrite-pool'; // The exact runtime gate substring (single backslash before the underscore). // Derived here so tests fail loudly if the installed package ever drifts. @@ -271,3 +276,74 @@ describe('createIntrospectionFilterPool — dedicated wrapper', () => { }); }); }); + +// ============================================================================= +// 7. rewrite-pool integration +// ============================================================================= + +const logicalName = (name: string): string => name.split('-').slice(2).join('-'); +const CANON_APP = 'app-11111111-app-public'; +const TENANT_APP = 'app-22222222-app-public'; + +const settingsQuery = (schemas: string[]) => ({ + text: 'select set_config(el->>0, el->>1, true) from json_array_elements($1::json) el', + values: [JSON.stringify([['role', 'app_user'], [POOL_SCHEMAS_GUC, JSON.stringify(schemas)]])] +}); + +describe('createRewritingPool — introspection filter integration', () => { + beforeEach(() => { + process.env.GRAPHILE_INTROSPECTION_FILTER = '1'; + }); + + it('applies settings, swaps introspection (not identifier-rewritten), still rewrites tenant SQL', async () => { + const client = introspectingClient([[], []]); + const pool = { connect: jest.fn().mockResolvedValue(client), query: jest.fn(), on: jest.fn() }; + const counters = createRewriteCounters(); + const wrapped = createRewritingPool(pool, { + canonicalSchemas: [CANON_APP], + logicalName, + counters, + introspectionFilter: { servedSchemas: [CANON_APP] } + }); + + const c = await wrapped.connect(); + + // Settings establish the tenant map. + await c.query(settingsQuery([TENANT_APP, 'public'])); + expect(counters.settingsParses).toBe(1); + + // Introspection: swapped, NOT identifier-rewritten, no fail-closed. + await c.query({ text: STOCK }); + const introspectionCall = client.query.mock.calls.find( + (call: any[]) => call[0] && typeof call[0] === 'object' && + typeof call[0].text === 'string' && call[0].text.includes('as introspection') + ); + expect(introspectionCall).toBeDefined(); + const swapped = introspectionCall[0].text as string; + expect(countOccurrences(swapped, 'nspname = any (array[')).toBe(4); + expect(countOccurrences(swapped, GATE_SUBSTRING)).toBe(0); + // Introspection was not treated as a rewrite target. + expect(counters.failClosed).toBe(0); + + // A normal tenant query is still identifier-rewritten as before. + await c.query({ text: `select from "${CANON_APP}"."t"` }); + const rewritten = sentConfigTexts(client).find((t) => t && t.includes(TENANT_APP)); + expect(rewritten).toBe(`select from "${TENANT_APP}"."t"`); + expect(counters.rewrittenQueries).toBeGreaterThanOrEqual(1); + }); + + it('feeds the process-wide introspection-filter counters', async () => { + const before = getIntrospectionFilterCounters().swaps; + const client = introspectingClient([[]]); + const pool = { connect: jest.fn().mockResolvedValue(client), query: jest.fn(), on: jest.fn() }; + const wrapped = createRewritingPool(pool, { + canonicalSchemas: [CANON_APP], + logicalName, + introspectionFilter: { servedSchemas: [CANON_APP] } + }); + const c = await wrapped.connect(); + await c.query(settingsQuery([TENANT_APP, 'public'])); + await c.query({ text: STOCK }); + expect(getIntrospectionFilterCounters().swaps).toBe(before + 1); + }); +}); diff --git a/graphql/server/src/middleware/__tests__/rewrite-pool.test.ts b/graphql/server/src/middleware/__tests__/rewrite-pool.test.ts new file mode 100644 index 000000000..0113e8ad8 --- /dev/null +++ b/graphql/server/src/middleware/__tests__/rewrite-pool.test.ts @@ -0,0 +1,637 @@ +import { + containsQuotedIdentifier, + createRewriteCounters, + createRewritingPool, + getRewritePoolCounters, + POOL_SCHEMAS_GUC, + RewriteCounters, + rewriteQuotedIdentifiers +} from '../rewrite-pool'; + +// ----------------------------------------------------------------------------- +// Fixtures: physical schemas shaped `--`; the logical name is +// everything past the first two dash segments, so a1111111/a2222222 tenants of +// the same shape collapse to the same logical name. +// ----------------------------------------------------------------------------- + +const logicalName = (name: string): string => name.split('-').slice(2).join('-'); + +const CANON_APP = 'app-11111111-app-public'; +const CANON_AUTH = 'app-11111111-auth-public'; +const TENANT_APP = 'app-22222222-app-public'; +const TENANT_AUTH = 'app-22222222-auth-public'; + +const canonicalSchemas = [CANON_APP, CANON_AUTH]; +const tenantSchemas = [TENANT_APP, TENANT_AUTH, 'public']; + +const settingsQuery = (schemas: string[]) => ({ + text: 'select set_config(el->>0, el->>1, true) from json_array_elements($1::json) el', + values: [JSON.stringify([['role', 'app_user'], [POOL_SCHEMAS_GUC, JSON.stringify(schemas)]])] +}); + +const makeClient = () => ({ + query: jest.fn().mockResolvedValue({ rows: [] }), + release: jest.fn(), + on: jest.fn(), + escapeIdentifier: jest.fn((s: string) => `"${s}"`) +}); + +const makePool = (client: ReturnType) => ({ + query: jest.fn().mockResolvedValue({ rows: [] }), + connect: jest.fn().mockResolvedValue(client), + on: jest.fn(), + end: jest.fn() +}); + +const setup = (overrides: Partial[1]> = {}) => { + const counters = createRewriteCounters(); + const client = makeClient(); + const pool = makePool(client); + const wrapped = createRewritingPool(pool, { + canonicalSchemas, + logicalName, + counters, + ...overrides + }); + return { counters, client, pool, wrapped }; +}; + +// Text of the Nth (0-based) call to the raw client query mock. +const sentText = (client: ReturnType, n: number): string => { + const arg = client.query.mock.calls[n][0]; + return typeof arg === 'string' ? arg : arg.text; +}; + +// ============================================================================= +// Pure lexer: rewriteQuotedIdentifiers +// ============================================================================= + +describe('rewriteQuotedIdentifiers', () => { + const map = new Map([[CANON_APP, TENANT_APP]]); + + it('replaces every quoted occurrence, including ::cast enum refs', () => { + const input = `select "${CANON_APP}"."t".*, x::"${CANON_APP}"."my_enum" from "${CANON_APP}"."t"`; + const expected = `select "${TENANT_APP}"."t".*, x::"${TENANT_APP}"."my_enum" from "${TENANT_APP}"."t"`; + expect(rewriteQuotedIdentifiers(input, map)).toBe(expected); + }); + + it('returns the input unchanged (byte-identical) when nothing matches', () => { + const input = 'select "users"."id" from "users"'; + expect(rewriteQuotedIdentifiers(input, map)).toBe(input); + }); + + describe('never touches non-identifier regions', () => { + const m = new Map([['canon', 'TENANT']]); + + it('single-quoted literal incl. doubled-quote escape', () => { + const input = `select '"canon" it''s "canon"', "canon"`; + expect(rewriteQuotedIdentifiers(input, m)).toBe(`select '"canon" it''s "canon"', "TENANT"`); + }); + + it("E-string with backslash escapes and embedded quotes", () => { + // Actual text: E'a\'b "canon" c' || "canon" + const input = `E'a\\'b "canon" c' || "canon"`; + expect(rewriteQuotedIdentifiers(input, m)).toBe(`E'a\\'b "canon" c' || "TENANT"`); + }); + + it('$$ and $tag$ dollar-quoted bodies', () => { + const input = `$$ "canon" $$ || $tag$ x "canon" y $tag$ || "canon"`; + expect(rewriteQuotedIdentifiers(input, m)).toBe(`$$ "canon" $$ || $tag$ x "canon" y $tag$ || "TENANT"`); + }); + + it('line comments and nested block comments', () => { + const line = `-- "canon"\n"canon"`; + expect(rewriteQuotedIdentifiers(line, m)).toBe(`-- "canon"\n"TENANT"`); + + const block = `/* "canon" /* "canon" */ still "canon" */ "canon"`; + expect(rewriteQuotedIdentifiers(block, m)).toBe(`/* "canon" /* "canon" */ still "canon" */ "TENANT"`); + }); + }); + + it('does not replace an identifier that merely prefixes a longer one', () => { + const m = new Map([['canon', 'TENANT']]); + expect(rewriteQuotedIdentifiers('"canon_extra" "canon"', m)).toBe('"canon_extra" "TENANT"'); + }); + + it('matches on the unescaped value of an identifier with embedded ""', () => { + const m = new Map([['ca"non', 'TENANT']]); + expect(rewriteQuotedIdentifiers('"ca""non"', m)).toBe('"TENANT"'); + }); + + it('escapes a " inside the replacement value', () => { + const m = new Map([['canon', 'ten"ant']]); + expect(rewriteQuotedIdentifiers('"canon"', m)).toBe('"ten""ant"'); + }); +}); + +// ============================================================================= +// Pure lexer: containsQuotedIdentifier +// ============================================================================= + +describe('containsQuotedIdentifier', () => { + const names = new Set([CANON_APP]); + + it('true only for a complete quoted token match', () => { + expect(containsQuotedIdentifier(`from "${CANON_APP}"."t"`, names)).toBe(true); + expect(containsQuotedIdentifier(`from "${CANON_APP}_extra"."t"`, names)).toBe(false); + }); + + it('ignores matches inside strings and comments', () => { + expect(containsQuotedIdentifier(`select '"${CANON_APP}"'`, names)).toBe(false); + expect(containsQuotedIdentifier(`-- "${CANON_APP}"`, names)).toBe(false); + expect(containsQuotedIdentifier(`$$ "${CANON_APP}" $$`, names)).toBe(false); + }); +}); + +// ============================================================================= +// Wrapper: end-to-end +// ============================================================================= + +describe('createRewritingPool — settings + rewriting', () => { + it('passes the settings query through byte-identical, then rewrites later queries', async () => { + const { client, wrapped } = setup(); + const c = await wrapped.connect(); + + const sq = settingsQuery(tenantSchemas); + await c.query(sq); + // Forwarded untouched (same object, identical text/values). + expect(client.query).toHaveBeenNthCalledWith(1, sq); + + await c.query({ text: `select * from "${CANON_APP}"."users" join "${CANON_AUTH}"."roles" using (id)` }); + expect(sentText(client, 1)).toBe( + `select * from "${TENANT_APP}"."users" join "${TENANT_AUTH}"."roles" using (id)` + ); + }); + + it('supports string and (text, values) call forms', async () => { + const { client, wrapped } = setup(); + const c = await wrapped.connect(); + await c.query(settingsQuery(tenantSchemas)); + + await c.query(`select from "${CANON_APP}"."t"`); + expect(sentText(client, 1)).toBe(`select from "${TENANT_APP}"."t"`); + + await c.query(`select from "${CANON_APP}"."t" where x = $1`, [1]); + expect(sentText(client, 2)).toBe(`select from "${TENANT_APP}"."t" where x = $1`); + expect(client.query.mock.calls[2][1]).toEqual([1]); + }); + + it('does not mutate the caller-supplied query config', async () => { + const { client, wrapped } = setup(); + const c = await wrapped.connect(); + await c.query(settingsQuery(tenantSchemas)); + + const config = { text: `select from "${CANON_APP}"."t"`, name: 'p1' }; + await c.query(config); + expect(config.text).toBe(`select from "${CANON_APP}"."t"`); + expect(config.name).toBe('p1'); + // The forwarded object is a clone with rewritten fields. + expect(client.query.mock.calls[1][0]).not.toBe(config); + expect(sentText(client, 1)).toBe(`select from "${TENANT_APP}"."t"`); + }); + + it('wraps clients handed back via callback-style connect', (done) => { + const client = makeClient(); + const pool = { + connect: (cb: any) => cb(null, client, jest.fn()), + query: jest.fn() + }; + const wrapped = createRewritingPool(pool, { canonicalSchemas, logicalName }); + + wrapped.connect((err: any, c: any) => { + expect(err).toBeNull(); + c.query(settingsQuery(tenantSchemas)); + c.query({ text: `select from "${CANON_APP}"."t"` }); + expect(sentText(client, 1)).toBe(`select from "${TENANT_APP}"."t"`); + done(); + }); + }); + + it('forwards non-query members bound to the raw client', async () => { + const { client, wrapped } = setup(); + const c = await wrapped.connect(); + expect(c.escapeIdentifier('x')).toBe('"x"'); + expect(client.escapeIdentifier).toHaveBeenCalledWith('x'); + }); +}); + +// ============================================================================= +// Wrapper: identity mode (canonical tenant querying itself) +// ============================================================================= + +describe('createRewritingPool — identity mode', () => { + it('leaves text and prepared names untouched when the tenant is canonical', async () => { + const { client, counters, wrapped } = setup(); + const c = await wrapped.connect(); + await c.query(settingsQuery([CANON_APP, CANON_AUTH, 'public'])); + + await c.query({ text: `select from "${CANON_APP}"."t"`, name: 'p1' }); + expect(sentText(client, 1)).toBe(`select from "${CANON_APP}"."t"`); + expect(client.query.mock.calls[1][0].name).toBe('p1'); + expect(counters.rewrittenQueries).toBe(0); + expect(counters.nameRewrites).toBe(0); + }); +}); + +// ============================================================================= +// Wrapper: prepared-statement name namespacing +// ============================================================================= + +describe('createRewritingPool — prepared name namespacing', () => { + it('is deterministic per tenant and differs across tenants', async () => { + const a = setup(); + const ca = await a.wrapped.connect(); + await ca.query(settingsQuery(tenantSchemas)); + await ca.query({ text: 'select 1', name: 'stmt' }); + await ca.query({ text: 'select 2', name: 'stmt' }); + const nameA1 = a.client.query.mock.calls[1][0].name; + const nameA2 = a.client.query.mock.calls[2][0].name; + expect(nameA1).toMatch(/^bp_[0-9a-f]{24}$/); + expect(nameA2).toBe(nameA1); // same tenant + original name => same + + const b = setup(); + const cb = await b.wrapped.connect(); + await cb.query(settingsQuery(['app-33333333-app-public', 'app-33333333-auth-public', 'public'])); + await cb.query({ text: 'select 1', name: 'stmt' }); + const nameB1 = b.client.query.mock.calls[1][0].name; + expect(nameB1).not.toBe(nameA1); // different tenant => different name + }); +}); + +// ============================================================================= +// Wrapper: prepared-statement lifecycle (the wrapper owns eviction + DEALLOCATE) +// ============================================================================= + +describe('createRewritingPool — prepared-statement lifecycle', () => { + // The `deallocate "bp_…"` statements the raw client received, in order. + const deallocs = (client: ReturnType): string[] => + client.query.mock.calls + .map((call: any[]) => call[0]) + .filter((arg: any): arg is string => typeof arg === 'string' && arg.startsWith('deallocate')); + + it('passes DEALLOCATE text through untouched (adaptor concern, not ours)', async () => { + const { client, wrapped } = setup(); + const c = await wrapped.connect(); + await c.query(settingsQuery(tenantSchemas)); + + await c.query('deallocate stmt'); + await c.query('deallocate all'); + expect(sentText(client, 1)).toBe('deallocate stmt'); + expect(sentText(client, 2)).toBe('deallocate all'); + }); + + it('evicts the oldest hashed name and deallocates it once the cap is exceeded', async () => { + const { client, wrapped } = setup({ maxPreparedNames: 2 }); + const c = await wrapped.connect(); + await c.query(settingsQuery(tenantSchemas)); + + await c.query({ text: 'select 1', name: 'a' }); + await c.query({ text: 'select 2', name: 'b' }); + const hashedA = client.query.mock.calls[1][0].name; + expect(hashedA).toMatch(/^bp_[0-9a-f]{24}$/); + + await c.query({ text: 'select 3', name: 'c' }); // 3rd distinct name -> evict 'a' + + expect(deallocs(client)).toEqual([`deallocate "${hashedA}"`]); + }); + + it('clears node-postgres parsedStatements bookkeeping for the evicted name', async () => { + const { client, wrapped } = setup({ maxPreparedNames: 2 }); + const c = await wrapped.connect(); + await c.query(settingsQuery(tenantSchemas)); + + await c.query({ text: 'select 1', name: 'a' }); + const hashedA = client.query.mock.calls[1][0].name; + // Simulate node-postgres having parsed/prepared under the hashed name. + (client as any).connection = { parsedStatements: { [hashedA]: 'select 1' } }; + + await c.query({ text: 'select 2', name: 'b' }); + await c.query({ text: 'select 3', name: 'c' }); // evict 'a' + await new Promise((resolve) => setImmediate(resolve)); // flush the fire-and-forget then() + + // Without this cleanup, re-preparing 'a' would skip Parse and fail on PG + // with "prepared statement does not exist". + expect((client as any).connection.parsedStatements[hashedA]).toBeUndefined(); + }); + + it('refreshes recency: re-touching name 1 evicts name 2 instead', async () => { + const { client, wrapped } = setup({ maxPreparedNames: 2 }); + const c = await wrapped.connect(); + await c.query(settingsQuery(tenantSchemas)); + + await c.query({ text: 'select 1', name: 'a' }); // [a] + await c.query({ text: 'select 2', name: 'b' }); // [a, b] + const hashedA = client.query.mock.calls[1][0].name; + const hashedB = client.query.mock.calls[2][0].name; + + await c.query({ text: 'select 1 again', name: 'a' }); // refresh a -> [b, a] + await c.query({ text: 'select 3', name: 'c' }); // insert c -> evict b + + expect(deallocs(client)).toEqual([`deallocate "${hashedB}"`]); + expect(deallocs(client)).not.toContain(`deallocate "${hashedA}"`); + }); + + it('re-using the same name neither grows the LRU nor deallocates', async () => { + const { client, wrapped } = setup({ maxPreparedNames: 2 }); + const c = await wrapped.connect(); + await c.query(settingsQuery(tenantSchemas)); + + await c.query({ text: 'select 1', name: 'a' }); + await c.query({ text: 'select 1 v2', name: 'a' }); + await c.query({ text: 'select 1 v3', name: 'a' }); + + expect(deallocs(client)).toEqual([]); + }); + + it('keeps the prepared-name LRU across release (it lives on the physical connection)', async () => { + const { client, wrapped } = setup({ maxPreparedNames: 2 }); + + const c1 = await wrapped.connect(); + await c1.query(settingsQuery(tenantSchemas)); + await c1.query({ text: 'select 1', name: 'a' }); // [a] + const hashedA = client.query.mock.calls[1][0].name; + c1.release(); + + // Same physical client returns; tenant state reset, prepared-name LRU intact. + const c2 = await wrapped.connect(); + await c2.query(settingsQuery(tenantSchemas)); + await c2.query({ text: 'select 2', name: 'b' }); // [a, b] + await c2.query({ text: 'select 3', name: 'c' }); // evict a (survived release) + + expect(deallocs(client)).toEqual([`deallocate "${hashedA}"`]); + }); + + it('honors DATAPLAN_PG_PREPARED_STATEMENT_CACHE_SIZE for the default cap', async () => { + const original = process.env.DATAPLAN_PG_PREPARED_STATEMENT_CACHE_SIZE; + process.env.DATAPLAN_PG_PREPARED_STATEMENT_CACHE_SIZE = '1'; + try { + const { client, wrapped } = setup(); // no explicit maxPreparedNames + const c = await wrapped.connect(); + await c.query(settingsQuery(tenantSchemas)); + + await c.query({ text: 'select 1', name: 'a' }); + const hashedA = client.query.mock.calls[1][0].name; + await c.query({ text: 'select 2', name: 'b' }); // cap 1 -> evict a + + expect(deallocs(client)).toEqual([`deallocate "${hashedA}"`]); + } finally { + if (original === undefined) delete process.env.DATAPLAN_PG_PREPARED_STATEMENT_CACHE_SIZE; + else process.env.DATAPLAN_PG_PREPARED_STATEMENT_CACHE_SIZE = original; + } + }); +}); + +// ============================================================================= +// Wrapper: fail-closed +// ============================================================================= + +describe('createRewritingPool — fail closed', () => { + it('rejects a rewrite-needing query issued before any settings query', async () => { + const { client, counters, wrapped } = setup(); + const c = await wrapped.connect(); + + await expect(c.query({ text: `select from "${CANON_APP}"."t"` })).rejects.toThrow( + /requires a tenant mapping but none is established/ + ); + expect(counters.failClosed).toBe(1); + expect(client.query).not.toHaveBeenCalled(); + }); + + it('passes benign non-rewrite statements through untouched', async () => { + const { client, counters, wrapped } = setup(); + const c = await wrapped.connect(); + + await c.query('begin'); + await c.query({ text: 'listen "x"' }); + await c.query({ text: 'select 1 from pg_catalog.pg_class' }); + await c.query({ text: 'select set_config($1,$2,true)', values: ['role', 'app_user'] }); + + expect(counters.failClosed).toBe(0); + expect(client.query.mock.calls.map((call: any[]) => call[0])).toEqual([ + 'begin', + { text: 'listen "x"' }, + { text: 'select 1 from pg_catalog.pg_class' }, + { text: 'select set_config($1,$2,true)', values: ['role', 'app_user'] } + ]); + }); + + it('fails closed with a callback when called callback-style', async () => { + const { wrapped } = setup(); + const c = await wrapped.connect(); + const err: Error = await new Promise((resolve) => { + c.query({ text: `select from "${CANON_APP}"."t"` }, (e: Error) => resolve(e)); + }); + expect(err.message).toMatch(/requires a tenant mapping/); + }); +}); + +// ============================================================================= +// Wrapper: settings detection keys on the adaptor statement TEXT, not on +// user-influenceable parameter contents (A1 — misclassification hardening). +// ============================================================================= + +describe('createRewritingPool — settings detection is text-gated', () => { + it('does not misclassify a data query as settings when a bind value contains the GUC substring', async () => { + const { client, counters, wrapped } = setup(); + const c = await wrapped.connect(); + await c.query(settingsQuery(tenantSchemas)); + expect(counters.settingsParses).toBe(1); + + // A genuine data query whose user-controlled bind value contains the GUC + // string. Under parameter-only detection this was treated as the settings + // query and forwarded UNREWRITTEN (canonical SQL on the tenant connection). + await c.query({ + text: `select from "${CANON_APP}"."t" where x = $1`, + values: [`${POOL_SCHEMAS_GUC} = whatever`] + }); + + // Rewritten to the tenant schema — not forwarded with canonical identifiers. + expect(sentText(client, 1)).toBe(`select from "${TENANT_APP}"."t" where x = $1`); + // The malicious value was NOT parsed as settings; checkout map is intact. + expect(counters.settingsParses).toBe(1); + }); + + it('a GUC-substring bind value cannot null the established tenant map (no self-DoS)', async () => { + const { client, counters, wrapped } = setup(); + const c = await wrapped.connect(); + await c.query(settingsQuery(tenantSchemas)); + + // Malicious non-JSON value that, under parameter-only detection, would have + // re-run applySettings and nulled tenantMap (forcing later fail-closed). + await c.query({ + text: `select from "${CANON_APP}"."t"`, + values: [`${POOL_SCHEMAS_GUC}: not json`] + }); + expect(sentText(client, 1)).toBe(`select from "${TENANT_APP}"."t"`); + + // Sibling query still rewrites (map not clobbered) rather than failing closed. + await c.query({ text: `select from "${CANON_AUTH}"."r"` }); + expect(sentText(client, 2)).toBe(`select from "${TENANT_AUTH}"."r"`); + expect(counters.failClosed).toBe(0); + expect(counters.settingsParses).toBe(1); + }); + + it('a crafted GUC-bearing JSON bind value cannot remap the checkout to another tenant', async () => { + const { client, wrapped } = setup(); + const c = await wrapped.connect(); + await c.query(settingsQuery(tenantSchemas)); // maps canonical -> app-22222222 + + const victim = ['app-99999999-app-public', 'app-99999999-auth-public', 'public']; + // Same value shape the real settings query would carry, but on a DATA query. + await c.query({ + text: `select from "${CANON_APP}"."t"`, + values: [JSON.stringify([[POOL_SCHEMAS_GUC, JSON.stringify(victim)]])] + }); + + // Still the original tenant (22222222), never the injected victim (99999999). + expect(sentText(client, 1)).toBe(`select from "${TENANT_APP}"."t"`); + }); + + it('fails closed (does not swallow as settings) for a pre-settings query carrying a GUC value', async () => { + const { client, counters, wrapped } = setup(); + const c = await wrapped.connect(); + + await expect( + c.query({ text: `select from "${CANON_APP}"."t"`, values: [`${POOL_SCHEMAS_GUC} x`] }) + ).rejects.toThrow(/requires a tenant mapping/); + expect(counters.failClosed).toBe(1); + expect(counters.settingsParses).toBe(0); + expect(client.query).not.toHaveBeenCalled(); + }); +}); + +// ============================================================================= +// Wrapper: release clears state +// ============================================================================= + +describe('createRewritingPool — release clears state', () => { + it('re-checkout of the same client requires a fresh settings query', async () => { + const { client, wrapped } = setup(); + const c = await wrapped.connect(); + await c.query(settingsQuery(tenantSchemas)); + await c.query({ text: `select from "${CANON_APP}"."t"` }); // ok + + c.release(); + expect(client.release).toHaveBeenCalled(); + + await expect(c.query({ text: `select from "${CANON_APP}"."t"` })).rejects.toThrow( + /requires a tenant mapping/ + ); + }); +}); + +// ============================================================================= +// Wrapper: logical mismatch => invalid checkout +// ============================================================================= + +describe('createRewritingPool — logical mismatch', () => { + it('marks the checkout invalid and fails closed with the reason', async () => { + const { counters, wrapped } = setup(); + const c = await wrapped.connect(); + // Tenant list is missing an auth-public counterpart. + await c.query(settingsQuery([TENANT_APP, 'public'])); + + await expect(c.query({ text: `select from "${CANON_AUTH}"."t"` })).rejects.toThrow( + /auth-public/ + ); + expect(counters.failClosed).toBe(1); + }); +}); + +// ============================================================================= +// Wrapper: LRU memo cap +// ============================================================================= + +describe('createRewritingPool — LRU memo cap', () => { + it('respects the cap while keeping every rewrite correct', async () => { + const { client, counters, wrapped } = setup({ maxMemoEntries: 2 }); + const c = await wrapped.connect(); + await c.query(settingsQuery(tenantSchemas)); + + const q1 = { text: `select from "${CANON_APP}"."a"` }; + const q2 = { text: `select from "${CANON_APP}"."b"` }; + const q3 = { text: `select from "${CANON_APP}"."c"` }; + + await c.query(q1); + await c.query(q2); + await c.query(q3); // fills then evicts q1 (cap 2) + expect(counters.memoHits).toBe(0); + + await c.query(q1); // q1 was evicted => miss (proves the cap held) + expect(counters.memoHits).toBe(0); + + await c.query(q3); // q3 still cached => hit + expect(counters.memoHits).toBe(1); + + // Every rewrite stayed correct regardless of eviction. + expect(sentText(client, 1)).toBe(`select from "${TENANT_APP}"."a"`); + expect(sentText(client, 3)).toBe(`select from "${TENANT_APP}"."c"`); + expect(sentText(client, 4)).toBe(`select from "${TENANT_APP}"."a"`); + expect(sentText(client, 5)).toBe(`select from "${TENANT_APP}"."c"`); + }); +}); + +// ============================================================================= +// Wrapper: counters +// ============================================================================= + +describe('createRewritingPool — counters', () => { + it('increments each counter exactly as specified', async () => { + const { wrapped, counters } = setup(); + const c = await wrapped.connect(); + + await c.query(settingsQuery(tenantSchemas)); // settingsParses + await c.query({ text: `select from "${CANON_APP}"."t"` }); // rewrittenQueries (miss) + await c.query({ text: `select from "${CANON_APP}"."t"` }); // rewrittenQueries + memoHits + await c.query({ text: 'select 1', name: 'p1' }); // nameRewrites + + const expected: RewriteCounters = { + settingsParses: 1, + rewrittenQueries: 2, + memoHits: 1, + failClosed: 0, + nameRewrites: 1 + }; + expect(counters).toEqual(expected); + }); + + it('shares a caller-provided counters object', async () => { + const counters = createRewriteCounters(); + const { wrapped } = setup({ counters }); + const c = await wrapped.connect(); + await c.query(settingsQuery(tenantSchemas)); + expect(counters.settingsParses).toBe(1); + }); + + it('feeds the process-wide getRewritePoolCounters() snapshot when no counters are passed', async () => { + const before = getRewritePoolCounters().settingsParses; + // No explicit counters -> activity lands in the module-level sink. + const client = makeClient(); + const wrapped = createRewritingPool(makePool(client), { canonicalSchemas, logicalName }); + const c = await wrapped.connect(); + await c.query(settingsQuery(tenantSchemas)); + + expect(getRewritePoolCounters().settingsParses).toBe(before + 1); + // Returns a snapshot copy, not the live object. + const snap = getRewritePoolCounters(); + snap.settingsParses = -999; + expect(getRewritePoolCounters().settingsParses).toBe(before + 1); + }); +}); + +// ============================================================================= +// Pool-level one-shot query +// ============================================================================= + +describe('createRewritingPool — pool.query one-shot', () => { + it('passes benign queries through and fails closed on rewrite-needing ones', async () => { + const { pool, wrapped } = setup(); + + await wrapped.query('select 1'); + expect(pool.query).toHaveBeenCalledWith('select 1'); + + await expect(wrapped.query({ text: `select from "${CANON_APP}"."t"` })).rejects.toThrow( + /requires a tenant mapping/ + ); + }); +}); diff --git a/graphql/server/src/middleware/__tests__/settings-text-pin.test.ts b/graphql/server/src/middleware/__tests__/settings-text-pin.test.ts new file mode 100644 index 000000000..ebc706a68 --- /dev/null +++ b/graphql/server/src/middleware/__tests__/settings-text-pin.test.ts @@ -0,0 +1,24 @@ +import { readFileSync } from 'fs'; + +import { SETTINGS_QUERY_TEXT } from '../rewrite-pool'; + +/** + * Pins SETTINGS_QUERY_TEXT against the INSTALLED @dataplan/pg adaptor source. + * + * The rewriting pool classifies the adaptor's pgSettings statement by exact + * text equality (defense against user bind values that merely contain the + * pool_schemas GUC substring). If a dependency bump changes the emitted + * statement, classification silently stops matching: applySettings never runs, + * no checkout ever gets a tenant mapping, and every pooled data query fails + * closed. That is the SAFE direction, but it disables pooling entirely and + * only surfaces at runtime — this test makes the drift fail loudly here + * instead. Mirrors the pg-introspection pinning approach in + * introspection-filter.test.ts. + */ +describe('SETTINGS_QUERY_TEXT pins the installed @dataplan/pg adaptor', () => { + it('the adaptor source still emits the exact statement we byte-match', () => { + const adaptorPath = require.resolve('@dataplan/pg/adaptors/pg'); + const source = readFileSync(adaptorPath, 'utf8'); + expect(source).toContain(SETTINGS_QUERY_TEXT); + }); +}); diff --git a/graphql/server/src/middleware/blueprint.ts b/graphql/server/src/middleware/blueprint.ts new file mode 100644 index 000000000..bfa687859 --- /dev/null +++ b/graphql/server/src/middleware/blueprint.ts @@ -0,0 +1,159 @@ +import { createHash } from 'node:crypto'; +import type { Pool } from 'pg'; + +// ============================================================================= +// Blueprint pooling identity helpers +// +// OPT-IN "blueprint pooling" shares one PostGraphile instance per schema-shape. +// Tenants are routed per request by the rewriting pool (canonical→tenant schema- +// identifier rewrite; see ./rewrite-pool). These helpers derive the identity of a +// shared instance (shape fingerprint + blueprint key). +// ============================================================================= + +/** + * Whether blueprint pooling is enabled via the GRAPHILE_BLUEPRINT_POOLING env + * flag. Accepts '1' or 'true'; anything else (incl. unset) means disabled. + */ +export const isBlueprintPoolingEnabled = (): boolean => { + const value = process.env.GRAPHILE_BLUEPRINT_POOLING; + return value === '1' || value === 'true'; +}; + +/** + * Strip the tenant hash prefix from a physical schema name. + * + * Tenant physical schemas look like `-<8hex>-`, e.g. + * `marketplace-db-tenant1-5e6b13b2-app-public` -> `app-public`. We strip through + * the FIRST `-<8 lowercase hex>-` occurrence and return the remainder. Control + * plane schemas (e.g. `services_public`) contain no such segment and are + * returned unchanged. + */ +export const stripSchemaHashPrefix = (name: string): string => { + const match = /-[0-9a-f]{8}-/.exec(name); + if (!match) return name; + return name.slice(match.index + match[0].length); +}; + +export interface SchemaRelation { + nspname: string; + relname: string; +} + +/** + * Single catalog scan backing the shape fingerprint. + */ +export const fetchSchemaRelations = async ( + pool: Pool, + physicalSchemas: string[] +): Promise => { + const result = await pool.query( + `SELECT n.nspname, c.relname + FROM pg_class c + JOIN pg_namespace n ON n.oid = c.relnamespace + WHERE n.nspname = ANY($1) AND c.relkind IN ('r','v','m','p') + ORDER BY 1, 2`, + [physicalSchemas] + ); + return result.rows; +}; + +/** + * Pure fingerprint over pre-fetched relations: sha256 of the sorted + * `[logicalSchema, relname]` pairs (physical names mapped to logical form so + * same-shape tenants collapse to the same fingerprint). + * + * NOTE (v1 granularity): the fingerprint covers relation NAMES only — not + * columns, functions or enum labels. Same-relname column drift between tenants + * is not detected here; it is bounded by flushService's flush-all-pooled-on- + * schema:update semantics. Extending the fingerprint to attributes/procs is a + * documented follow-up. + */ +export const fingerprintFromRelations = (rows: SchemaRelation[]): string => { + const pairs: [string, string][] = rows.map((row) => [ + stripSchemaHashPrefix(row.nspname), + row.relname + ]); + pairs.sort((a, b) => { + if (a[0] !== b[0]) return a[0] < b[0] ? -1 : 1; + if (a[1] !== b[1]) return a[1] < b[1] ? -1 : 1; + return 0; + }); + return createHash('sha256').update(JSON.stringify(pairs)).digest('hex'); +}; + +/** + * Compute a fingerprint of the relational shape shared by a set of physical + * schemas. Physical schema names are mapped to their logical form (hash prefix + * stripped) so that different tenants of the same shape collapse to the same + * fingerprint. The fingerprint is a sha256 hex digest over the sorted + * `[logicalSchema, relname]` pairs. + */ +export const computeShapeFingerprint = async ( + pool: Pool, + physicalSchemas: string[] +): Promise => fingerprintFromRelations(await fetchSchemaRelations(pool, physicalSchemas)); + +/** + * Compute the stable blueprint key that identifies a poolable PostGraphile + * instance. Two requests share an instance iff they agree on logical schemas, + * shape fingerprint, gather/schema flags, api name and mode. Inputs are + * normalized (schemas sorted, flag entries sorted by key) so the key is + * independent of ordering. + */ +export const computeBlueprintKey = (input: { + logicalSchemas: string[]; + shapeFingerprint: string; + flags: Record | undefined; + apiName?: string | null; + mode: string; + /** + * Physical database backing the instance's pool. REQUIRED for correctness in + * multi-database fleets: without it, same-shape tenants living in DIFFERENT + * physical databases would share one instance whose pool points at only one + * of them. + */ + dbname?: string | null; +}): string => { + const payload = { + s: [...input.logicalSchemas].sort(), + f: input.shapeFingerprint, + g: Object.entries(input.flags || {}).sort((a, b) => + a[0] < b[0] ? -1 : a[0] > b[0] ? 1 : 0 + ), + a: input.apiName || '', + m: input.mode, + d: input.dbname || '' + }; + + return 'bp:' + createHash('sha256').update(JSON.stringify(payload)).digest('hex'); +}; + +/** + * Detect bm25 indexes (pg_textsearch / VectorChord-BM25 access method) in the + * given physical schemas. + * + * WHY THIS GATES POOLING: graphile-search's bm25 adapter passes the + * schema-qualified index name to to_bm25query() as a BIND VALUE captured at + * build time. The rewriting pool maps canonical→tenant schema identifiers in + * SQL TEXT only — bind values are user data and are deliberately never + * rewritten — so a pooled instance would aim every tenant's bm25 scoring at + * the CANONICAL tenant's index. Shapes that carry bm25 indexes therefore keep + * per-tenant instances (see pooling-decision.ts) until the seam learns to + * rewrite regclass-shaped values (documented follow-up). + */ +export const hasBm25Indexes = async ( + pool: Pool, + physicalSchemas: string[] +): Promise => { + const result = await pool.query<{ exists: boolean }>( + `SELECT EXISTS ( + SELECT 1 + FROM pg_class c + JOIN pg_namespace n ON n.oid = c.relnamespace + JOIN pg_am am ON am.oid = c.relam + WHERE n.nspname = ANY($1) AND c.relkind = 'i' AND am.amname = 'bm25' + ) AS exists`, + [physicalSchemas] + ); + return result.rows[0]?.exists === true; +}; diff --git a/graphql/server/src/middleware/pooling-decision.ts b/graphql/server/src/middleware/pooling-decision.ts new file mode 100644 index 000000000..fcb94e054 --- /dev/null +++ b/graphql/server/src/middleware/pooling-decision.ts @@ -0,0 +1,162 @@ +import { Logger } from '@pgpmjs/logger'; +import type { Pool } from 'pg'; + +import type { ApiStructure } from '../types'; +import { + computeBlueprintKey, + fetchSchemaRelations, + fingerprintFromRelations, + hasBm25Indexes, + stripSchemaHashPrefix +} from './blueprint'; + +// ============================================================================= +// Blueprint Pooling: per-svc decision + decision cache +// +// Decides whether a service (svc_key) may attach to a SHARED PostGraphile +// instance (blueprint pooling, tenant-routed by the rewriting pool — see +// ./rewrite-pool) or must keep a per-tenant instance. +// Kept in its own module — free of the heavy graphile.ts import chain — so the +// decision logic is unit-testable without loading PostGraphile. +// ============================================================================= + +const log = new Logger('graphile:pooling'); + +export interface PoolDecision { + /** Effective graphileCache key: a `bp:` blueprint key when pooling, else the svc_key. */ + key: string; + /** Whether this svc attaches to a shared (blueprint-pooled) instance. */ + pooling: boolean; + /** + * True when the fallback came from a thrown catalog probe (a possibly + * transient DB error) rather than a structural opt-out. Transient decisions + * are NOT memoized, so the next request re-probes instead of permanently + * pinning the svc to a per-tenant instance until the next flush. + */ + transient?: boolean; + /** + * The tenant database this decision belongs to — lets flushService invalidate + * ONLY the affected database's decisions/blueprint instead of a fleet-wide + * flush on every schema:update. + */ + databaseId?: string; +} + +/** + * Memoized pooling decision per svc_key. Computing a decision runs one catalog + * probe (the shape fingerprint), so the result is cached and invalidated on + * flush (see clearPoolDecisions / flushService). + */ +const poolDecisions = new Map(); + +/** + * Clear all cached pooling decisions (explicit /flush admin route). + */ +export function clearPoolDecisions(): void { + poolDecisions.clear(); +} + +/** + * Invalidate ONLY the decisions belonging to one tenant database and return the + * `bp:` keys it was attached to (so flushService can rebuild exactly the + * affected blueprint). A schema change in tenant X: + * - X's decisions must recompute (its fingerprint may have changed), and + * - the blueprint instance X was pooled into must rebuild (its shared schema + * was derived from a catalog that included X's old shape), + * while every OTHER blueprint keeps serving untouched. A brand-new tenant has + * no memoized decisions => provisioning events are a no-op here (new tenants + * probe on first request; instances are shape-generic and need no rebuild to + * accept another same-shape tenant). + */ +export function clearPoolDecisionsForDatabase(databaseId: string): string[] { + const bpKeys = new Set(); + for (const [svcKey, decision] of poolDecisions) { + if (decision.databaseId === databaseId) { + if (decision.pooling && decision.key.startsWith('bp:')) { + bpKeys.add(decision.key); + } + poolDecisions.delete(svcKey); + } + } + return [...bpKeys]; +} + +/** + * Compute (WITHOUT caching) the blueprint-pooling decision for a service. + * + * Returns `{ pooling: false, key: svcKey }` — a per-tenant instance keyed by the + * normal svc_key — when the API opts out of pooling: realtime is enabled, there + * are no schemas, or the catalog probe throws. Otherwise returns + * `{ pooling: true, key: 'bp:...' }` so same-shape tenants share one instance. + * + * Exported for unit testing; the dispatcher uses the cached resolvePoolDecision. + */ +export const computePoolDecision = async ( + svcKey: string, + api: ApiStructure, + pool: Pool +): Promise => { + // Realtime instances hold a LISTEN/NOTIFY connection + RealtimeManager per + // instance and can't be safely shared across tenants — never pool them. + if (api.databaseSettings?.enableRealtime === true) { + return { key: svcKey, pooling: false }; + } + if (!api.schema || api.schema.length === 0) { + return { key: svcKey, pooling: false }; + } + + try { + // One catalog scan feeds the fingerprint. + const relations = await fetchSchemaRelations(pool, api.schema); + + // Structural opt-out: bm25-indexed shapes are not poolable. The bm25 + // adapter embeds the build-time schema-qualified index name as a bind + // VALUE, which the rewrite seam deliberately never touches — a shared + // instance would aim every tenant's bm25 scoring at the canonical + // tenant's index (see hasBm25Indexes). Memoized like any structural + // decision; a schema change flush re-probes. + if (await hasBm25Indexes(pool, api.schema)) { + log.info( + `svc=${svcKey} not poolable — schemas carry bm25 indexes; using per-tenant instance` + ); + return { key: svcKey, pooling: false }; + } + + const key = computeBlueprintKey({ + logicalSchemas: api.logicalSchemas ?? api.schema.map(stripSchemaHashPrefix), + shapeFingerprint: fingerprintFromRelations(relations), + flags: api.databaseSettings as Record | undefined, + apiName: (api as { apiName?: string | null }).apiName ?? null, + mode: 'public', + dbname: api.dbname + }); + return { key, pooling: true }; + } catch (err) { + log.warn( + `svc=${svcKey} not poolable — shape probe failed: ${err instanceof Error ? err.message : String(err)}; using per-tenant instance (not memoized)` + ); + return { key: svcKey, pooling: false, transient: true }; + } +}; + +/** + * Resolve the pooling decision for a service, memoized per svc_key. Exported for + * unit testing of the memoization/invalidation behaviour. + */ +export const resolvePoolDecision = async ( + svcKey: string, + api: ApiStructure, + pool: Pool +): Promise => { + const cached = poolDecisions.get(svcKey); + if (cached) return cached; + const decision = await computePoolDecision(svcKey, api, pool); + if (!decision.transient) { + // Stamp the owning database once, at the single memoization point, so + // flushService can invalidate per-database instead of fleet-wide. + const memoized: PoolDecision = { ...decision, databaseId: (api as any).databaseId ?? undefined }; + poolDecisions.set(svcKey, memoized); + return memoized; + } + return decision; +}; diff --git a/graphql/server/src/middleware/rewrite-pool.ts b/graphql/server/src/middleware/rewrite-pool.ts new file mode 100644 index 000000000..528afd9cb --- /dev/null +++ b/graphql/server/src/middleware/rewrite-pool.ts @@ -0,0 +1,696 @@ +import { createHash } from 'node:crypto'; + +import { + createIntrospectionInterceptor, + type IntrospectionInterceptor, + isIntrospectionQuery, + noteIntrospectionCallbackPassthrough +} from './introspection-filter'; + +// ============================================================================= +// Rewriting pool: per-request tenant schema rewriting for a pooled instance +// +// One PostGraphile instance is pooled across many tenants whose PostgreSQL +// schemas are shape-identical but differently prefixed (e.g. canonical +// `myapp-a1b2c3d4-app-public` vs tenant `other-9f8e7d6c-app-public`). The +// pooled instance emits fully-qualified SQL naming the CANONICAL tenant's +// schemas. This module wraps the `pg` Pool handed to the PostGraphile adaptor +// and, per request, rewrites the canonical schema identifiers to the requesting +// tenant's — plus namespaces prepared-statement names per tenant so one tenant's +// prepared plan never serves another. +// +// Tenant identity arrives inside the adaptor's pgSettings query via a custom GUC +// (POOL_SCHEMAS_GUC). Until a checkout has seen that GUC it has NO mapping; any +// query that would otherwise ship canonical SQL FAILS CLOSED rather than read +// the wrong tenant's data. +// +// The SQL rewriting is a single-pass lexer that rewrites ONLY double-quoted +// identifier tokens — never string literals, dollar-quoted bodies, comments or +// E-strings — so output is byte-identical to input except for replaced idents. +// ============================================================================= + +export const POOL_SCHEMAS_GUC = 'constructive.pool_schemas'; + +export interface RewriteCounters { + settingsParses: number; + rewrittenQueries: number; + memoHits: number; + failClosed: number; + nameRewrites: number; +} + +export interface RewritingPoolOptions { + /** Physical schema names the pooled instance was built against. */ + canonicalSchemas: string[]; + /** Maps a physical schema name to its logical (prefix-stripped) name. Injected. */ + logicalName: (physical: string) => string; + /** LRU cap for both the needs-rewrite and rewrite-result memos. Default 2000. */ + maxMemoEntries?: number; + /** + * Per-connection cap on the prepared-statement names we issue. Defaults to the + * adaptor's own DATAPLAN_PG_PREPARED_STATEMENT_CACHE_SIZE parse (unset -> 100). + */ + maxPreparedNames?: number; + /** Optional shared counters; a fresh set is created when omitted. */ + counters?: RewriteCounters; + /** + * When set, the pooled instance's catalog introspection query is scoped to the + * given served schemas (plus their transitive cross-schema references, public + * and pg_catalog) before it runs — see ./introspection-filter. `false`/omitted + * leaves introspection unfiltered (today's behavior). + */ + introspectionFilter?: { servedSchemas: string[] } | false; +} + +/** Build a zeroed counters object (companion to RewritingPoolOptions.counters). */ +export function createRewriteCounters(): RewriteCounters { + return { + settingsParses: 0, + rewrittenQueries: 0, + memoHits: 0, + failClosed: 0, + nameRewrites: 0 + }; +} + +// Process-wide counters: the default sink when a pool is created without its own +// counters, so getRewritePoolCounters() (read by the metrics sampler) reflects +// real pooled activity aggregated across every pooled instance. +const moduleCounters = createRewriteCounters(); + +/** Snapshot the process-wide rewrite-pool counters (metrics sampler entry point). Returns a copy so callers cannot mutate live state. */ +export function getRewritePoolCounters(): RewriteCounters { + return { ...moduleCounters }; +} + +// ----------------------------------------------------------------------------- +// SQL lexer +// ----------------------------------------------------------------------------- + +const isIdentChar = (ch: string | undefined): boolean => { + if (ch === undefined) return false; + return ( + (ch >= 'a' && ch <= 'z') || + (ch >= 'A' && ch <= 'Z') || + (ch >= '0' && ch <= '9') || + ch === '_' || + ch === '$' + ); +}; + +const isTagStartChar = (ch: string | undefined): boolean => { + if (ch === undefined) return false; + return (ch >= 'a' && ch <= 'z') || (ch >= 'A' && ch <= 'Z') || ch === '_'; +}; + +const isTagChar = (ch: string | undefined): boolean => + isTagStartChar(ch) || (ch !== undefined && ch >= '0' && ch <= '9'); + +/** + * If `text[i]` opens a dollar-quote tag (`$$` or `$tag$` where the tag follows + * unquoted-identifier rules), return the full opening tag (e.g. `$$`, `$body$`); + * otherwise null. Positional params like `$1` return null (digit-led tag). + */ +const matchDollarTag = (text: string, i: number): string | null => { + // text[i] === '$' + const next = text[i + 1]; + if (next === '$') return '$$'; + if (!isTagStartChar(next)) return null; + let j = i + 2; + while (j < text.length && isTagChar(text[j])) j++; + if (text[j] === '$') return text.slice(i, j + 1); + return null; +}; + +interface IdentifierToken { + /** Index of the opening `"`. */ + start: number; + /** Index just past the closing `"`. */ + end: number; + /** Unescaped identifier value (`""` collapsed to `"`). */ + value: string; +} + +/** + * Single left-to-right scan collecting every double-quoted identifier token, + * skipping over string literals, E-strings, dollar-quoted bodies, and line / + * (nested) block comments. This is the ONLY place identifier boundaries are + * decided; both public helpers below reuse it. + */ +const scanIdentifierTokens = (text: string): IdentifierToken[] => { + const tokens: IdentifierToken[] = []; + const n = text.length; + let i = 0; + + while (i < n) { + const ch = text[i]; + + // Line comment -- ... to end of line + if (ch === '-' && text[i + 1] === '-') { + i += 2; + while (i < n && text[i] !== '\n') i++; + continue; + } + + // Block comment /* ... */ (nested) + if (ch === '/' && text[i + 1] === '*') { + i += 2; + let depth = 1; + while (i < n && depth > 0) { + if (text[i] === '/' && text[i + 1] === '*') { + depth++; + i += 2; + } else if (text[i] === '*' && text[i + 1] === '/') { + depth--; + i += 2; + } else { + i++; + } + } + continue; + } + + // Double-quoted identifier — the only token we ever rewrite + if (ch === '"') { + const start = i; + i++; + let value = ''; + while (i < n) { + if (text[i] === '"') { + if (text[i + 1] === '"') { + value += '"'; + i += 2; + continue; + } + i++; + break; + } + value += text[i]; + i++; + } + tokens.push({ start, end: i, value }); + continue; + } + + // Dollar-quoted string $tag$ ... $tag$ + if (ch === '$') { + const tag = matchDollarTag(text, i); + if (tag !== null) { + i += tag.length; + const close = text.indexOf(tag, i); + i = close === -1 ? n : close + tag.length; + continue; + } + } + + // E-string: a standalone e/E immediately followed by a quote. Both \\-style + // backslash escapes AND '' count; a backslash escapes the next char. + if ((ch === 'e' || ch === 'E') && text[i + 1] === "'" && !isIdentChar(text[i - 1])) { + i += 2; + while (i < n) { + const c = text[i]; + if (c === '\\') { + i += 2; + continue; + } + if (c === "'") { + if (text[i + 1] === "'") { + i += 2; + continue; + } + i++; + break; + } + i++; + } + continue; + } + + // Regular single-quoted string literal ('' is an escaped quote) + if (ch === "'") { + i++; + while (i < n) { + if (text[i] === "'") { + if (text[i + 1] === "'") { + i += 2; + continue; + } + i++; + break; + } + i++; + } + continue; + } + + i++; + } + + return tokens; +}; + +/** + * Rewrite ONLY double-quoted identifier tokens whose unescaped value is a key of + * `map`. String literals, dollar-quoted bodies, comments, E-strings and + * unquoted words are never touched. Output is byte-identical to input except for + * replaced tokens (a replacement is re-quoted with embedded `"` doubled). + */ +export function rewriteQuotedIdentifiers(text: string, map: Map): string { + const tokens = scanIdentifierTokens(text); + if (tokens.length === 0) return text; + + let out = ''; + let last = 0; + for (const tok of tokens) { + const replacement = map.get(tok.value); + // Non-matches keep their original bytes: they stay inside the next verbatim + // slice because `last` only advances past tokens we actually replace. + if (replacement === undefined) continue; + out += text.slice(last, tok.start); + out += '"' + replacement.replace(/"/g, '""') + '"'; + last = tok.end; + } + if (last === 0) return text; + out += text.slice(last); + return out; +} + +/** + * True iff `text` contains any of `names` as a complete double-quoted identifier + * token (same lexer as rewriteQuotedIdentifiers — matches on the unescaped + * value, so an identifier merely prefixing a longer one does not count). + */ +export function containsQuotedIdentifier(text: string, names: Set): boolean { + const tokens = scanIdentifierTokens(text); + for (const tok of tokens) { + if (names.has(tok.value)) return true; + } + return false; +} + +// ----------------------------------------------------------------------------- +// Small insertion-order LRU (bounded memo). A stored `false`/'' is a real value, +// so callers distinguish a miss via `has()` rather than an undefined return. +// ----------------------------------------------------------------------------- + +class LruCache { + private readonly map = new Map(); + constructor(private readonly max: number) {} + + has(key: string): boolean { + return this.map.has(key); + } + + get(key: string): V | undefined { + if (!this.map.has(key)) return undefined; + const value = this.map.get(key); + // Touch: move to most-recently-used position. + this.map.delete(key); + this.map.set(key, value); + return value; + } + + set(key: string, value: V): void { + if (this.map.has(key)) this.map.delete(key); + this.map.set(key, value); + if (this.map.size > this.max) { + const oldest = this.map.keys().next().value; + this.map.delete(oldest); + } + } +} + +// ----------------------------------------------------------------------------- +// Pool / client wrapper +// ----------------------------------------------------------------------------- + +interface CheckoutState { + /** Canonical -> tenant physical schema map; null until a settings query lands. */ + tenantMap: Map | null; + /** Per-tenant namespace tag; null in identity mode (canonical querying itself). */ + tenantTag: string | null; + /** Set when the tenant schema set can't satisfy the canonical shape (fail closed). */ + invalidReason: string | null; + /** Per-checkout introspection interceptor (memoizes discovery); null when the filter is off. */ + introspectionInterceptor: IntrospectionInterceptor | null; +} + +const freshState = (): CheckoutState => ({ + tenantMap: null, + tenantTag: null, + invalidReason: null, + introspectionInterceptor: null +}); + +const sha256hex = (input: string): string => createHash('sha256').update(input).digest('hex'); + +/** Deterministic per-tenant prepared-statement name: same tenant+name -> same result. */ +const hashName = (tenantTag: string, name: string): string => + 'bp_' + sha256hex(tenantTag + ':' + name).slice(0, 24); + +// The adaptor's fixed pgSettings statement (@dataplan/pg adaptors/pg.js: the +// `select set_config(el->>0, el->>1, true) from json_array_elements($1::json) el` +// it issues with the JSON-serialized pgSettings as $1). Settings detection MUST +// key on THIS TEXT — never on parameter contents alone: a data query whose +// user-controlled bind value merely contains the GUC substring must never be +// classified as the settings query, or it would be forwarded UNREWRITTEN (canonical +// SQL on the tenant connection) and its value could remap or null the checkout's +// tenant state for sibling queries. This literal byte-matches SQL emitted upstream. +// Exported ONLY for the pinning test that asserts it against the installed +// @dataplan/pg source — a dependency bump that changes the statement must fail +// loudly at test time, not silently fail-closed at runtime. +export const SETTINGS_QUERY_TEXT = + 'select set_config(el->>0, el->>1, true) from json_array_elements($1::json) el'; + +const isSettingsText = (text: string): boolean => text === SETTINGS_QUERY_TEXT; + +const isSettingsValues = (values: any): boolean => + Array.isArray(values) && + values.length > 0 && + typeof values[0] === 'string' && + values[0].includes(POOL_SCHEMAS_GUC); + +// Per-connection prepared-statement state lives on the RAW client under this +// module-private Symbol: prepared statements are physical-connection scoped and +// must survive checkout/release (unlike per-checkout tenant state). +const BP_PREPARED = Symbol('constructive.rewrite-pool.prepared'); + +/** + * Replicate @dataplan/pg's DATAPLAN_PG_PREPARED_STATEMENT_CACHE_SIZE parse: + * unset / empty / non-numeric -> 100; an explicit 0 -> 0 (upstream then disables + * prepared statements entirely, so our LRU is never exercised). + */ +const parsePreparedStatementCacheSize = (raw: string | undefined): number => { + const fromEnv = raw ? parseInt(raw, 10) : null; + return !!fromEnv || fromEnv === 0 ? fromEnv : 100; +}; + +export function createRewritingPool(pool: any, opts: RewritingPoolOptions): any { + const counters = opts.counters ?? moduleCounters; + const maxMemoEntries = opts.maxMemoEntries ?? 2000; + const maxPreparedNames = + opts.maxPreparedNames ?? + parsePreparedStatementCacheSize(process.env.DATAPLAN_PG_PREPARED_STATEMENT_CACHE_SIZE); + const canonicalSchemas = opts.canonicalSchemas; + const canonicalNames = new Set(canonicalSchemas); + const logicalName = opts.logicalName; + + // Caches live at pool scope (shared across checkouts). `needs` depends only on + // the text; the rewrite result is disambiguated per tenant via tenantTag. + const needsCache = new LruCache(maxMemoEntries); + const rewriteCache = new LruCache(maxMemoEntries); + + const needsRewrite = (text: string): boolean => { + if (needsCache.has(text)) return needsCache.get(text); + const result = containsQuotedIdentifier(text, canonicalNames); + needsCache.set(text, result); + return result; + }; + + const rewriteText = (tenantTag: string, text: string, map: Map): string => { + const key = tenantTag + ' ' + text; + if (rewriteCache.has(key)) { + counters.rewrittenQueries += 1; + counters.memoHits += 1; + return rewriteCache.get(key); + } + const out = rewriteQuotedIdentifiers(text, map); + rewriteCache.set(key, out); + counters.rewrittenQueries += 1; + return out; + }; + + // We rename prepared statements to bp_, which defeats the adaptor's own + // dispose (it keys its LRU by the ORIGINAL name, but pg's parsedStatements is + // now keyed by our hashed name, so its `deallocate ` never fires). + // So the wrapper OWNS the lifecycle: an insertion-ordered LRU of hashed names, + // kept on the RAW client so it survives release, evicts the oldest and fires a + // best-effort DEALLOCATE for it once the per-connection cap is exceeded. + const trackPreparedName = (rawClient: any, hashedName: string): void => { + if (maxPreparedNames <= 0) return; + let store: Map = rawClient[BP_PREPARED]; + if (!store) { + store = new Map(); + rawClient[BP_PREPARED] = store; + } + if (store.has(hashedName)) { + // Refresh recency. + store.delete(hashedName); + store.set(hashedName, true); + return; + } + store.set(hashedName, true); + if (store.size > maxPreparedNames) { + const evicted = store.keys().next().value; + store.delete(evicted); + // Fire-and-forget on the raw connection (hashed names need no escaping). + Promise.resolve(rawClient.query('deallocate "' + evicted + '"')) + .then(() => { + // Mirror the adaptor's dispose: node-postgres tracks prepared names in + // connection.parsedStatements. Without clearing the entry, re-preparing + // the evicted name later would skip the Parse step and fail on PG with + // "prepared statement does not exist". + const parsed = rawClient.connection?.parsedStatements; + if (parsed) delete parsed[evicted]; + }) + .catch((err: any) => { + // eslint-disable-next-line no-console + console.error('[rewrite-pool] failed to deallocate prepared statement ' + evicted, err); + }); + } + }; + + // Parse the adaptor's pgSettings query and (re)build the checkout's tenant map. + const applySettings = (text: string, values: any[], state: CheckoutState): void => { + // Defense in depth: only the adaptor's fixed settings statement may mutate + // checkout tenant state. The call site already gates on this text; guard again + // so no stray query whose parameter merely contains the GUC substring can + // suppress rewriting or remap the checkout via parameter contents alone. + if (!isSettingsText(text)) return; + counters.settingsParses += 1; + + let tenantSchemas: string[] | null = null; + try { + const parsed = JSON.parse(values[0]); + let raw: any; + if (Array.isArray(parsed)) { + for (const entry of parsed) { + if (Array.isArray(entry) && entry[0] === POOL_SCHEMAS_GUC) { + raw = entry[1]; + break; + } + } + } else if (parsed && typeof parsed === 'object') { + raw = parsed[POOL_SCHEMAS_GUC]; + } + if (typeof raw === 'string') tenantSchemas = JSON.parse(raw); + else if (Array.isArray(raw)) tenantSchemas = raw; + } catch { + tenantSchemas = null; + } + + if (!Array.isArray(tenantSchemas)) { + state.tenantMap = null; + state.tenantTag = null; + state.invalidReason = `malformed ${POOL_SCHEMAS_GUC} value`; + return; + } + + const tenantByLogical = new Map(); + for (const t of tenantSchemas) tenantByLogical.set(logicalName(t), t); + + const map = new Map(); + for (const c of canonicalSchemas) { + const logical = logicalName(c); + const t = tenantByLogical.get(logical); + if (t === undefined) { + state.tenantMap = null; + state.tenantTag = null; + state.invalidReason = `canonical schema "${c}" (logical "${logical}") has no tenant schema in [${tenantSchemas.join(', ')}]`; + return; + } + if (t !== c) map.set(c, t); + } + + state.tenantMap = map; + state.invalidReason = null; + if (map.size === 0) { + // Canonical tenant querying itself: identity mode, no rewriting/namespacing. + state.tenantTag = null; + } else { + state.tenantTag = sha256hex([...tenantSchemas].sort().join(',')).slice(0, 16); + } + }; + + const failClosed = (state: CheckoutState, args: any[]): any => { + counters.failClosed += 1; + const reason = state.invalidReason ?? 'no tenant schema mapping has been applied to this connection'; + const err = new Error( + `[rewrite-pool] pooled query requires a tenant mapping but none is established on this connection: ${reason}` + ); + const last = args[args.length - 1]; + if (typeof last === 'function') { + last(err); + return undefined; + } + return Promise.reject(err); + }; + + const interceptQuery = (rawQuery: any, thisArg: any, args: any[], state: CheckoutState): any => { + const first = args[0]; + + // Introspection filter (runs BEFORE settings parsing and any needs-rewrite / + // fail-closed logic). The introspection statement carries schema names only + // as string literals, so the identifier rewrite pass never applies to it — + // we forward the swapped text directly, bypassing rewriting entirely. + if (state.introspectionInterceptor) { + let iText: string | undefined; + if (typeof first === 'string') iText = first; + else if (first && typeof first === 'object' && typeof first.text === 'string') iText = first.text; + if (iText !== undefined && isIntrospectionQuery(iText)) { + // A callback-form introspection query would break callback semantics if + // rewritten async — pass it through unfiltered and count it. + if (typeof args[args.length - 1] === 'function') { + noteIntrospectionCallbackPassthrough(); + return rawQuery.apply(thisArg, args); + } + const swap = state.introspectionInterceptor(iText, { + query: (sql: string, values?: any[]) => thisArg.query(sql, values) + }); + if (swap !== null) { + return Promise.resolve(swap).then((swapped: string) => { + if (swapped === iText) return rawQuery.apply(thisArg, args); + if (typeof first === 'string') { + return rawQuery.apply(thisArg, [swapped, ...args.slice(1)]); + } + return rawQuery.apply(thisArg, [{ ...first, text: swapped }, ...args.slice(1)]); + }); + } + // swap === null: filter disabled / no served schemas — fall through. + } + } + + let text: string; + let values: any; + let name: any; + let isConfig = false; + + if (typeof first === 'string') { + text = first; + if (Array.isArray(args[1])) values = args[1]; + } else if (first && typeof first === 'object' && typeof first.text === 'string') { + isConfig = true; + text = first.text; + values = first.values; + name = first.name; + } else { + // Unknown shape (e.g. a Submittable) — forward untouched. + return rawQuery.apply(thisArg, args); + } + + // Settings query: parse tenant identity, forward the query itself untouched. + // Classification keys on the adaptor's fixed statement TEXT (plus the GUC-bearing + // value shape), never on parameter contents alone — a data query carrying a + // user-controlled value that contains the GUC substring has different text and + // therefore falls through to the needs-rewrite / fail-closed path below. + if (isSettingsText(text) && isSettingsValues(values)) { + applySettings(text, values, state); + return rawQuery.apply(thisArg, args); + } + + let newText = text; + if (needsRewrite(text)) { + if (state.invalidReason !== null || state.tenantMap === null) { + return failClosed(state, args); + } + if (state.tenantMap.size > 0) { + newText = rewriteText(state.tenantTag, text, state.tenantMap); + } + // else identity mode: leave text unchanged. + } + + // Prepared-statement name namespacing (config form only). thisArg is the raw + // client on the checkout path (the only path where tenantTag is ever set), so + // the per-connection prepared-name LRU hangs off it. + let newName = name; + if (isConfig && name != null && state.tenantTag !== null) { + newName = hashName(state.tenantTag, String(name)); + counters.nameRewrites += 1; + trackPreparedName(thisArg, newName); + } + + if (newText === text && newName === name) { + return rawQuery.apply(thisArg, args); + } + + // Clone rather than mutate the caller's object. + if (isConfig) { + const cloned = { ...first, text: newText }; + if (newName !== name) cloned.name = newName; + return rawQuery.apply(thisArg, [cloned, ...args.slice(1)]); + } + return rawQuery.apply(thisArg, [newText, ...args.slice(1)]); + }; + + const introspectionFilter = opts.introspectionFilter; + + const wrapClient = (client: any): any => { + const state = freshState(); + // Per-checkout introspection interceptor: memoizes the closure discovery so a + // single build's introspection is filtered once and reused. A fresh interceptor + // per checkout keeps the memo connection-scoped. + if (introspectionFilter && Array.isArray(introspectionFilter.servedSchemas)) { + state.introspectionInterceptor = createIntrospectionInterceptor({ + servedSchemas: introspectionFilter.servedSchemas + }); + } + return new Proxy(client, { + get(target, prop, receiver) { + if (prop === 'query') { + return (...args: any[]) => interceptQuery(target.query, target, args, state); + } + if (prop === 'release') { + return (...args: any[]) => { + // Per-checkout state must not leak into the next borrower. + state.tenantMap = null; + state.tenantTag = null; + state.invalidReason = null; + state.introspectionInterceptor = null; + return target.release.apply(target, args); + }; + } + const value = Reflect.get(target, prop, receiver); + return typeof value === 'function' ? value.bind(target) : value; + } + }); + }; + + return new Proxy(pool, { + get(target, prop, receiver) { + if (prop === 'connect') { + return (...args: any[]) => { + // Callback style: connect((err, client, release) => ...) + if (typeof args[0] === 'function') { + const cb = args[0]; + return target.connect((err: any, client: any, done: any) => { + if (err) { + cb(err, client, done); + return; + } + cb(err, wrapClient(client), done); + }); + } + // Promise style (used by the PostGraphile adaptor). + return Promise.resolve(target.connect(...args)).then((client: any) => wrapClient(client)); + }; + } + if (prop === 'query') { + // Pool-level one-shot: no established checkout => tenantMap stays null, + // so anything needing a rewrite fails closed; everything else passes. + return (...args: any[]) => interceptQuery(target.query, target, args, freshState()); + } + const value = Reflect.get(target, prop, receiver); + return typeof value === 'function' ? value.bind(target) : value; + } + }); +} diff --git a/packages/express-context/src/types.ts b/packages/express-context/src/types.ts index 39e0a79c5..f6a860a0e 100644 --- a/packages/express-context/src/types.ts +++ b/packages/express-context/src/types.ts @@ -106,6 +106,12 @@ export interface ApiStructure { anonRole: string; roleName: string; schema: string[]; + /** + * Logical schema names (tenant hash prefix stripped) derived from `schema`. + * Used by blueprint pooling to key a shared PostGraphile instance by + * schema-shape rather than by physical tenant schema. + */ + logicalSchemas?: string[]; apiModules: ApiModule[]; rlsModule?: RlsModule; domains?: string[]; diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index ca7b74485..0a8c79ca0 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -1762,6 +1762,9 @@ importers: '@aws-sdk/client-s3': specifier: ^3.1052.0 version: 3.1052.0 + '@dataplan/pg': + specifier: 1.0.3 + version: 1.0.3(@dataplan/json@1.0.0(grafast@1.0.2(graphql@16.13.0)))(grafast@1.0.2(graphql@16.13.0))(graphile-config@1.0.1)(graphql@16.13.0)(pg-sql2@5.0.1)(pg@8.21.0) '@types/cookie-parser': specifier: ^1.4.10 version: 1.4.10(@types/express@5.0.6)