diff --git a/.github/workflows/run-tests.yaml b/.github/workflows/run-tests.yaml index eecb3795cf..be081ef33f 100644 --- a/.github/workflows/run-tests.yaml +++ b/.github/workflows/run-tests.yaml @@ -246,7 +246,7 @@ jobs: - batch: pg-graphql packages: 'graphile/graphile-search graphile/graphile-ltree graphile/graphile-bulk-mutations graphile/graphile-function-bindings graphile/graphile-history graphile/graphile-meta graphile/graphile-schema graphql/orm-test graphql/test graphql/playwright-test' - batch: pg-graphile-extras - packages: 'graphile/graphile-i18n graphile/graphile-pg-aggregates graphile/graphile-query graphile/graphile-realtime-test' + packages: 'graphile/graphile-cache graphile/graphile-i18n graphile/graphile-pg-aggregates graphile/graphile-query graphile/graphile-realtime-test' env: PGHOST: localhost diff --git a/graphile/graphile-cache/README.md b/graphile/graphile-cache/README.md index 5eb5c674ba..d60718bf8f 100644 --- a/graphile/graphile-cache/README.md +++ b/graphile/graphile-cache/README.md @@ -37,6 +37,32 @@ When you import this package, it automatically registers a cleanup callback with ## Usage +### Exact build identity + +The GraphQL server and Explorer cache by `createGraphileBuildCacheKey(domain, +snapshot)`. A key includes the effective build inputs, the exact pool reference, +and the configuration owner. Server and Explorer use separate domains. Plain +data is copied with `snapshotGraphileBuildValue` before it is used for both the +fingerprint and the preset; opaque objects and functions retain reference +identity. Keys are HMAC fingerprints with a process-local secret and must not be +persisted or compared across processes. + +Entries carry `serviceKey`, `databaseId` (when known), and `poolKey` separately +from the fingerprint. Use `clearGraphileEntriesForService`, +`clearGraphileEntriesForDatabase`, or `clearGraphileEntriesForPool` to invalidate +all build variants for that owner. Pool cleanup matches `poolKey` exactly. + +The server consumes the API, physical pool, service key, and transaction settings +from `req.constructive`. Request settings remain request-scoped and are not +captured in the shared schema. Build refusals retain their canonical error code +and HTTP status; unknown failures return a generic 500 response with the original +cause kept internally. + +Bulk disposal drains report all failures for generations scheduled before the +call. Work scheduled later belongs to the next drain. An observed bulk failure +is acknowledged for reporting, while its exact-entry release promise remains +rejected; acknowledging an error never establishes successful resource release. + ### Basic Usage ```typescript diff --git a/graphile/graphile-cache/jest.config.js b/graphile/graphile-cache/jest.config.js index 057a9420ed..df73257b5a 100644 --- a/graphile/graphile-cache/jest.config.js +++ b/graphile/graphile-cache/jest.config.js @@ -7,7 +7,8 @@ module.exports = { 'ts-jest', { babelConfig: false, - tsconfig: 'tsconfig.json', + // ts-jest requires isolated transformation for NodeNext package exports. + tsconfig: { isolatedModules: true }, }, ], }, diff --git a/graphile/graphile-cache/package.json b/graphile/graphile-cache/package.json index baf35eb1a8..faa8f25acf 100644 --- a/graphile/graphile-cache/package.json +++ b/graphile/graphile-cache/package.json @@ -30,6 +30,7 @@ }, "dependencies": { "12factor-env": "workspace:^", + "@constructive-io/errors": "workspace:^", "@pgpmjs/logger": "workspace:^", "express": "^5.2.1", "grafserv": "1.0.1", @@ -42,6 +43,7 @@ "@types/express": "^5.0.6", "makage": "^0.8.0", "nodemon": "^3.1.14", + "pgsql-test": "workspace:^", "ts-node": "^10.9.2" }, "keywords": [ diff --git a/graphile/graphile-cache/src/__tests__/build-identity.test.ts b/graphile/graphile-cache/src/__tests__/build-identity.test.ts new file mode 100644 index 0000000000..b9e16968d0 --- /dev/null +++ b/graphile/graphile-cache/src/__tests__/build-identity.test.ts @@ -0,0 +1,138 @@ +import { + createGraphileBuildCacheKey, + referenceGraphileBuildValue, + snapshotGraphileBuildValue +} from '../build-identity'; + +const makeIdentity = (overrides: Record = {}) => ({ + owner: Symbol.for('server-owner'), + serviceKey: 'api:db-1:public-api', + pool: referenceGraphileBuildValue({}), + pgConfig: { host: 'db.internal', database: 'tenant_one', password: 'secret-value' }, + databaseName: 'tenant_one', + databaseId: 'db-1', + apiId: 'api-1', + schemas: ['app', 'auth'], + authRoles: { anon: 'app_anon', served: 'app_user', introspection: 'app_reader' }, + featureFlags: { enableSearch: true, enableRealtime: false }, + compute: [{ schemaName: 'compute', bindingsTableName: 'bindings' }], + explain: false, + surface: { graphqlPath: '/graphql', graphiql: true }, + callback: (): void => {}, + ...overrides +}); + +describe('Graphile build identity', () => { + it('canonicalizes plain object keys without depending on insertion order', () => { + expect(createGraphileBuildCacheKey('server', { a: 1, b: 2 })) + .toBe(createGraphileBuildCacheKey('server', { b: 2, a: 1 })); + }); + + it('preserves array order and distinguishes undefined from null', () => { + expect(createGraphileBuildCacheKey('server', ['app', 'auth'])) + .not.toBe(createGraphileBuildCacheKey('server', ['auth', 'app'])); + expect(createGraphileBuildCacheKey('server', { role: undefined })) + .not.toBe(createGraphileBuildCacheKey('server', { role: null })); + }); + + it.each([ + ['configuration owner', 'owner', Symbol.for('other-owner')], + ['service key', 'serviceKey', 'api:db-2:public-api'], + ['PostgreSQL config', 'pgConfig', { host: 'db.internal', database: 'tenant_two', password: 'secret-value' }], + ['database name', 'databaseName', 'tenant_two'], + ['database id', 'databaseId', 'db-2'], + ['API id', 'apiId', 'api-2'], + ['schema order', 'schemas', ['auth', 'app']], + ['served role', 'authRoles', { anon: 'app_anon', served: 'app_admin', introspection: 'app_reader' }], + ['introspection role', 'authRoles', { anon: 'app_anon', served: 'app_user', introspection: 'app_introspector' }], + ['plugin flags', 'featureFlags', { enableSearch: false, enableRealtime: false }], + ['compute bindings', 'compute', [{ schemaName: 'compute', bindingsTableName: 'other_bindings' }]], + ['explain', 'explain', true], + ['effective surface', 'surface', { graphqlPath: '/api/graphql', graphiql: true }] + ])('changes the key when %s changes', (_label, key, value) => { + const base = makeIdentity(); + const changed = { ...base, [key as string]: value }; + expect(createGraphileBuildCacheKey('server', changed)) + .not.toBe(createGraphileBuildCacheKey('server', base)); + }); + + it('uses function, symbol, opaque object, and explicit plain-object identity', () => { + const callback = (): void => {}; + class OpaqueInstance {} + const opaque = new OpaqueInstance(); + const plainPool = {}; + const same = { + callback, + symbol: Symbol.for('stable-symbol'), + opaque, + pool: referenceGraphileBuildValue(plainPool) + }; + const equivalent = { + callback, + symbol: Symbol.for('stable-symbol'), + opaque, + pool: referenceGraphileBuildValue(plainPool) + }; + expect(createGraphileBuildCacheKey('server', same)) + .toBe(createGraphileBuildCacheKey('server', equivalent)); + expect(createGraphileBuildCacheKey('server', { ...same, callback: (): void => {} })) + .not.toBe(createGraphileBuildCacheKey('server', same)); + expect(createGraphileBuildCacheKey('server', { ...same, symbol: Symbol('stable-symbol') })) + .not.toBe(createGraphileBuildCacheKey('server', same)); + expect(createGraphileBuildCacheKey('server', { ...same, opaque: new OpaqueInstance() })) + .not.toBe(createGraphileBuildCacheKey('server', same)); + expect(createGraphileBuildCacheKey('server', { + ...same, + pool: referenceGraphileBuildValue({}) + })).not.toBe(createGraphileBuildCacheKey('server', same)); + }); + + it('separates artifact domains and type-tagged adversarial values', () => { + const date = new Date('2025-01-01T00:00:00.000Z'); + const spoofedDate = { [Symbol.toStringTag]: 'Date', value: date.getTime() }; + const identity = makeIdentity(); + expect(createGraphileBuildCacheKey('server', identity)) + .not.toBe(createGraphileBuildCacheKey('explorer', identity)); + expect(createGraphileBuildCacheKey('server', date)) + .not.toBe(createGraphileBuildCacheKey('server', spoofedDate)); + expect(createGraphileBuildCacheKey('server', NaN)) + .not.toBe(createGraphileBuildCacheKey('server', 'NaN')); + expect(createGraphileBuildCacheKey('server', 0)) + .not.toBe(createGraphileBuildCacheKey('server', '0')); + expect(createGraphileBuildCacheKey('server', [, undefined])) + .not.toBe(createGraphileBuildCacheKey('server', [undefined, undefined])); + }); + + it('returns secret-free fingerprints and rejects cyclic or accessor data safely', () => { + const secret = 'unique-password-sentinel'; + const key = createGraphileBuildCacheKey('server', { password: secret }); + expect(key).not.toContain(secret); + expect(() => { + const cyclic: Record = {}; + cyclic.self = cyclic; + createGraphileBuildCacheKey('server', cyclic); + }).toThrow('Invalid Graphile build identity'); + expect(() => createGraphileBuildCacheKey('server', { + get password() { + throw new Error('getter should not run'); + } + })).toThrow('Invalid Graphile build identity'); + }); + + it('uses the immutable snapshot for both the key and consumed build input', () => { + const source = { pg: { ssl: { rejectUnauthorized: false } }, schemas: ['app'] }; + const snapshot = snapshotGraphileBuildValue(source); + const before = createGraphileBuildCacheKey('server', snapshot); + + source.pg.ssl.rejectUnauthorized = true; + source.schemas.push('private'); + + expect(createGraphileBuildCacheKey('server', snapshot)).toBe(before); + expect(snapshot.pg.ssl.rejectUnauthorized).toBe(false); + expect(snapshot.schemas).toEqual(['app']); + expect(Object.isFrozen(snapshot)).toBe(true); + expect(Object.isFrozen(snapshot.pg)).toBe(true); + expect(Object.isFrozen(snapshot.pg.ssl)).toBe(true); + expect(Object.isFrozen(snapshot.schemas)).toBe(true); + }); +}); diff --git a/graphile/graphile-cache/src/__tests__/build-invalidation.test.ts b/graphile/graphile-cache/src/__tests__/build-invalidation.test.ts new file mode 100644 index 0000000000..723c4c0dff --- /dev/null +++ b/graphile/graphile-cache/src/__tests__/build-invalidation.test.ts @@ -0,0 +1,82 @@ +import { + clearGraphileCache, + clearGraphileEntriesForDatabase, + clearGraphileEntriesForPool, + clearGraphileEntriesForService, + graphileCache, + type GraphileCacheEntry +} from '../graphile-cache'; + +const makeEntry = ( + cacheKey: string, + metadata: Pick +): GraphileCacheEntry => ({ + pgl: { release: jest.fn().mockResolvedValue(undefined) } as any, + serv: {} as any, + handler: {} as any, + httpServer: { listening: false } as any, + cacheKey, + createdAt: Date.now(), + ...metadata +}); + +describe('Graphile cache identity invalidation', () => { + afterEach(async () => { + await clearGraphileCache(); + }); + + it.each([undefined, null, ''])('ignores an absent service key (%s)', (serviceKey) => { + const legacy = makeEntry('legacy-key', {}); + const tagged = makeEntry('graphile:server:one', { serviceKey: 'service-a' }); + graphileCache.set(legacy.cacheKey, legacy); + graphileCache.set(tagged.cacheKey, tagged); + + expect(clearGraphileEntriesForService(serviceKey as unknown as string)).toBe(0); + expect(graphileCache.get(legacy.cacheKey)).toBe(legacy); + expect(graphileCache.get(tagged.cacheKey)).toBe(tagged); + expect(legacy.pgl.release).not.toHaveBeenCalled(); + expect(tagged.pgl.release).not.toHaveBeenCalled(); + }); + + it('clears every exact-key variant for only the matching service', () => { + const variants = [ + makeEntry('graphile:server:one', { serviceKey: 'service-a', databaseId: 'db-a', poolKey: 'pool-a' }), + makeEntry('graphile:server:two', { serviceKey: 'service-a', databaseId: 'db-a', poolKey: 'pool-a' }), + makeEntry('graphile:explorer:three', { serviceKey: 'service-b', databaseId: 'db-a', poolKey: 'pool-a' }) + ]; + for (const entry of variants) graphileCache.set(entry.cacheKey, entry); + + expect(clearGraphileEntriesForService('service-a')).toBe(2); + expect(graphileCache.has('graphile:server:one')).toBe(false); + expect(graphileCache.has('graphile:server:two')).toBe(false); + expect(graphileCache.has('graphile:explorer:three')).toBe(true); + }); + + it('clears every service variant for only the matching database', () => { + const variants = [ + makeEntry('graphile:server:one', { serviceKey: 'service-a', databaseId: 'db-a', poolKey: 'pool-a' }), + makeEntry('graphile:server:two', { serviceKey: 'service-b', databaseId: 'db-a', poolKey: 'pool-a' }), + makeEntry('graphile:explorer:three', { serviceKey: 'service-a', databaseId: 'db-b', poolKey: 'pool-b' }) + ]; + for (const entry of variants) graphileCache.set(entry.cacheKey, entry); + + expect(clearGraphileEntriesForDatabase('db-a')).toBe(2); + expect(graphileCache.has('graphile:server:one')).toBe(false); + expect(graphileCache.has('graphile:server:two')).toBe(false); + expect(graphileCache.has('graphile:explorer:three')).toBe(true); + }); + + it('clears every Graphile variant backed by only the disposed pool key', () => { + const variants = [ + makeEntry('graphile:server:one', { serviceKey: 'service-a', databaseId: 'db-a', poolKey: 'pool-a' }), + makeEntry('graphile:explorer:two', { serviceKey: 'service-b', databaseId: null, poolKey: 'pool-a' }), + makeEntry('graphile:server:three', { serviceKey: 'service-a', databaseId: 'db-a', poolKey: 'pool-ab' }) + ]; + for (const entry of variants) graphileCache.set(entry.cacheKey, entry); + + expect(clearGraphileEntriesForPool('pool-a')).toBe(2); + expect(graphileCache.has('graphile:server:one')).toBe(false); + expect(graphileCache.has('graphile:explorer:two')).toBe(false); + expect(graphileCache.has('graphile:server:three')).toBe(true); + }); +}); diff --git a/graphile/graphile-cache/src/__tests__/build-readiness.test.ts b/graphile/graphile-cache/src/__tests__/build-readiness.test.ts new file mode 100644 index 0000000000..ccc1b2851c --- /dev/null +++ b/graphile/graphile-cache/src/__tests__/build-readiness.test.ts @@ -0,0 +1,164 @@ +import { awaitGraphileBuildReadiness } from '../build-readiness'; + +interface Deferred { + promise: Promise; + resolve(value: T): void; + reject(error: Error): void; +} + +const deferred = (): Deferred => { + let resolve!: (value: T) => void; + let reject!: (error: Error) => void; + const promise = new Promise((resolvePromise, rejectPromise) => { + resolve = resolvePromise; + reject = rejectPromise; + }); + return { promise, resolve, reject }; +}; + +const flushPromises = (): Promise => + new Promise((resolve) => setImmediate(resolve)); + +describe('awaitGraphileBuildReadiness', () => { + it('does not resolve before schema gathering and Grafserv are ready', async () => { + const schemaResult = deferred(); + const ready = deferred(); + const release = jest.fn().mockResolvedValue(undefined); + let resolved = false; + const buildPromise = awaitGraphileBuildReadiness({ + schemaResult: schemaResult.promise, + addTo: jest.fn().mockResolvedValue(undefined), + ready: () => ready.promise, + release, + }).then(() => { + resolved = true; + }); + + schemaResult.resolve({}); + await flushPromises(); + expect(resolved).toBe(false); + + ready.resolve(undefined); + await buildPromise; + expect(release).not.toHaveBeenCalled(); + }); + + it('does not start readiness checks before the adapter is attached', async () => { + const addTo = deferred(); + const ready = jest.fn().mockResolvedValue(undefined); + const buildPromise = awaitGraphileBuildReadiness({ + schemaResult: Promise.resolve({}), + addTo: () => addTo.promise, + ready, + release: jest.fn().mockResolvedValue(undefined), + }); + + await flushPromises(); + expect(ready).not.toHaveBeenCalled(); + + addTo.resolve(undefined); + await buildPromise; + expect(ready).toHaveBeenCalledTimes(1); + }); + + it('waits for schema construction before releasing after adapter attachment fails', async () => { + const schemaResult = deferred(); + const attachmentFailure = new Error('adapter attachment failed'); + const release = jest.fn().mockResolvedValue(undefined); + const buildPromise = awaitGraphileBuildReadiness({ + schemaResult: schemaResult.promise, + addTo: jest.fn().mockRejectedValue(attachmentFailure), + ready: jest.fn().mockResolvedValue(undefined), + release, + }); + + await flushPromises(); + expect(release).not.toHaveBeenCalled(); + + schemaResult.resolve({}); + await expect(buildPromise).rejects.toBe(attachmentFailure); + expect(release).toHaveBeenCalledTimes(1); + }); + + it('waits for schema construction before releasing after readiness fails', async () => { + const schemaResult = deferred(); + const readinessFailure = new Error('grafserv readiness failed'); + const release = jest.fn().mockResolvedValue(undefined); + const buildPromise = awaitGraphileBuildReadiness({ + schemaResult: schemaResult.promise, + addTo: jest.fn().mockResolvedValue(undefined), + ready: jest.fn().mockRejectedValue(readinessFailure), + release, + }); + + await flushPromises(); + expect(release).not.toHaveBeenCalled(); + + schemaResult.resolve({}); + await expect(buildPromise).rejects.toBe(readinessFailure); + expect(release).toHaveBeenCalledTimes(1); + }); + + it('observes schema failure while adapter attachment is pending', async () => { + const schemaResult = deferred(); + const addTo = deferred(); + const release = jest.fn().mockResolvedValue(undefined); + const failure = new Error('schema build failed early'); + const buildPromise = awaitGraphileBuildReadiness({ + schemaResult: schemaResult.promise, + addTo: () => addTo.promise, + ready: jest.fn().mockResolvedValue(undefined), + release, + }); + + schemaResult.reject(failure); + await flushPromises(); + expect(release).not.toHaveBeenCalled(); + + addTo.resolve(undefined); + await expect(buildPromise).rejects.toBe(failure); + expect(release).toHaveBeenCalledTimes(1); + }); + + it('awaits failed-generation release before rejecting', async () => { + const schemaResult = deferred(); + const release = deferred(); + const releaseFn = jest.fn(() => release.promise); + const failure = new Error('schema build failed'); + let rejected = false; + const buildPromise = awaitGraphileBuildReadiness({ + schemaResult: schemaResult.promise, + addTo: jest.fn().mockResolvedValue(undefined), + ready: jest.fn().mockResolvedValue(undefined), + release: releaseFn, + }).catch((error) => { + rejected = true; + throw error; + }); + + schemaResult.reject(failure); + await flushPromises(); + expect(releaseFn).toHaveBeenCalledTimes(1); + expect(rejected).toBe(false); + + release.resolve(undefined); + await expect(buildPromise).rejects.toBe(failure); + }); + + it('preserves the build failure when cleanup also fails', async () => { + const failure = new Error('schema build failed'); + const cleanupFailure = new Error('release failed'); + const onReleaseError = jest.fn(); + + await expect( + awaitGraphileBuildReadiness({ + schemaResult: Promise.reject(failure), + addTo: jest.fn().mockResolvedValue(undefined), + ready: jest.fn().mockResolvedValue(undefined), + release: jest.fn().mockRejectedValue(cleanupFailure), + onReleaseError, + }) + ).rejects.toMatchObject({ errors: [failure, cleanupFailure], cause: failure }); + expect(onReleaseError).toHaveBeenCalledWith(cleanupFailure); + }); +}); diff --git a/graphile/graphile-cache/src/__tests__/disposal-drain.test.ts b/graphile/graphile-cache/src/__tests__/disposal-drain.test.ts new file mode 100644 index 0000000000..63026b44fe --- /dev/null +++ b/graphile/graphile-cache/src/__tests__/disposal-drain.test.ts @@ -0,0 +1,90 @@ +import { pgCache } from 'pg-cache'; + +import { + clearGraphileCache, + closeAllCaches, + disposeUncachedEntry, + graphileCache, + waitForActiveDisposals, + type GraphileCacheEntry +} from '../graphile-cache'; + +const entry = (key: string, release: () => Promise): GraphileCacheEntry => ({ + cacheKey: key, createdAt: Date.now(), pgl: { release }, + serv: {}, handler: {}, httpServer: { listening: false } +} as unknown as GraphileCacheEntry); + +afterEach(async () => { + jest.restoreAllMocks(); + await clearGraphileCache(); +}); + +it('reports already-settled failures to concurrent drains without repeating acknowledged failures', async () => { + const failure = new Error('release failed'); + const failed = entry('failed', async () => { throw failure; }); + await expect(disposeUncachedEntry(failed)).rejects.toBe(failure); + const drains = [waitForActiveDisposals(), waitForActiveDisposals()]; + await Promise.all(drains.map(drain => expect(drain).rejects.toBe(failure))); + await expect(waitForActiveDisposals()).resolves.toBeUndefined(); + await expect(disposeUncachedEntry(failed)).rejects.toBe(failure); +}); + +it('reports multiple generations in scheduling order and attempts every resource', async () => { + const stopFailure = new Error('stop failed'); + const releaseFailure = new Error('release failed'); + const laterFailure = new Error('later failed'); + const first = entry('first', async () => { throw releaseFailure; }); + first.realtimeManager = { stop: async () => { throw stopFailure; } }; + first.releasePresetServices = jest.fn().mockResolvedValue(undefined); + const firstDisposal = disposeUncachedEntry(first); + const later = disposeUncachedEntry(entry('later', async () => { throw laterFailure; })); + await expect(firstDisposal).rejects.toMatchObject({ errors: [stopFailure, releaseFailure] }); + await expect(later).rejects.toBe(laterFailure); + await expect(waitForActiveDisposals()).rejects.toMatchObject({ + errors: [expect.objectContaining({ errors: [stopFailure, releaseFailure] }), laterFailure] + }); + expect(first.releasePresetServices).toHaveBeenCalledTimes(1); +}); + +it('leaves work scheduled after the drain boundary for the next caller', async () => { + let finishFirst!: () => void; + let finishLater!: () => void; + const first = disposeUncachedEntry(entry('first', () => new Promise(resolve => { finishFirst = resolve; }))); + const drain = waitForActiveDisposals(); + const later = disposeUncachedEntry(entry('later', () => new Promise(resolve => { finishLater = resolve; }))); + finishFirst(); + await drain; + await first; + let laterDrained = false; + const next = waitForActiveDisposals().then(() => { laterDrained = true; }); + await new Promise(resolve => setImmediate(resolve)); + expect(laterDrained).toBe(false); + finishLater(); + await Promise.all([next, later]); +}); + +it('still closes pools when cache cleanup fails', async () => { + const failure = new Error('generation failed'); + const poolFailure = new Error('pool failed'); + graphileCache.set('failed', entry('failed', async () => { throw failure; })); + const close = jest.spyOn(pgCache, 'close').mockRejectedValue(poolFailure); + await expect(closeAllCaches()).rejects.toMatchObject({ errors: [failure, poolFailure] }); + expect(close).toHaveBeenCalledTimes(1); +}); + +it('includes synchronous reentrant disposal scheduled while clearing residents', async () => { + const failure = new Error('nested release failed'); + const nested = entry('nested', async () => { throw failure; }); + graphileCache.set('outer', entry('outer', () => disposeUncachedEntry(nested))); + await expect(clearGraphileCache()).rejects.toMatchObject({ errors: [failure, failure] }); +}); + +it('preserves a close callback error and still releases the remaining resources', async () => { + const failure = new Error('server close failed'); + const release = jest.fn().mockResolvedValue(undefined); + const resident = entry('server', release); + resident.httpServer = { listening: true, close: (callback: (error: Error) => void) => callback(failure) } as any; + await expect(disposeUncachedEntry(resident)).rejects.toBe(failure); + expect(release).toHaveBeenCalledTimes(1); + await expect(waitForActiveDisposals()).rejects.toBe(failure); +}); diff --git a/graphile/graphile-cache/src/__tests__/disposal-lifecycle.test.ts b/graphile/graphile-cache/src/__tests__/disposal-lifecycle.test.ts new file mode 100644 index 0000000000..62cf6420e5 --- /dev/null +++ b/graphile/graphile-cache/src/__tests__/disposal-lifecycle.test.ts @@ -0,0 +1,177 @@ +jest.mock('@pgpmjs/logger', () => ({ + Logger: jest.fn(() => ({ + debug: jest.fn(), + error: jest.fn(), + })), +})); + +import type { GraphileCacheEntry } from '../graphile-cache'; +import { + clearGraphileCache, + disposeUncachedEntry, + graphileCache, + waitForEntryDisposal, +} from '../graphile-cache'; + +interface Deferred { + promise: Promise; + resolve(value: T): void; +} + +const deferred = (): Deferred => { + let resolve!: (value: T) => void; + const promise = new Promise((resolvePromise) => { + resolve = resolvePromise; + }); + return { promise, resolve }; +}; + +const flushPromises = (): Promise => + new Promise((resolve) => setImmediate(resolve)); + +const makeEntry = ( + cacheKey: string, + release = jest.fn().mockResolvedValue(undefined), + releasePresetServices = jest.fn().mockResolvedValue(undefined) +): GraphileCacheEntry => + ({ + pgl: { release }, + serv: {}, + handler: {}, + httpServer: { listening: false }, + cacheKey, + createdAt: Date.now(), + releasePresetServices, + }) as unknown as GraphileCacheEntry; + +describe('Graphile cache disposal lifecycle', () => { + afterEach(async () => { + await clearGraphileCache(); + }); + + it('coalesces concurrent disposal of one exact entry', async () => { + const release = jest.fn().mockResolvedValue(undefined); + const releasePresetServices = jest.fn().mockResolvedValue(undefined); + const entry = makeEntry('same-entry', release, releasePresetServices); + + const first = disposeUncachedEntry(entry); + const second = disposeUncachedEntry(entry); + + expect(second).toBe(first); + await Promise.all([first, second]); + expect(release).toHaveBeenCalledTimes(1); + expect(releasePresetServices).toHaveBeenCalledTimes(1); + }); + + it('disposes distinct generations that reuse the same cache key', async () => { + const firstRelease = jest.fn().mockResolvedValue(undefined); + const secondRelease = jest.fn().mockResolvedValue(undefined); + const first = makeEntry('shared-key', firstRelease); + const second = makeEntry('shared-key', secondRelease); + + await Promise.all([ + disposeUncachedEntry(first), + disposeUncachedEntry(second), + ]); + + expect(firstRelease).toHaveBeenCalledTimes(1); + expect(secondRelease).toHaveBeenCalledTimes(1); + }); + + it('continues cleanup and exposes the first disposal failure', async () => { + const failure = new Error('realtime stop failed'); + const release = jest.fn().mockResolvedValue(undefined); + const releasePresetServices = jest.fn().mockResolvedValue(undefined); + const entry = makeEntry('failed-cleanup', release, releasePresetServices); + entry.realtimeManager = { stop: jest.fn().mockRejectedValue(failure) }; + + await expect(disposeUncachedEntry(entry)).rejects.toBe(failure); + expect(release).toHaveBeenCalledTimes(1); + expect(releasePresetServices).toHaveBeenCalledTimes(1); + await expect(clearGraphileCache()).rejects.toBe(failure); + }); + + it('lets callers await an eviction through the exact entry', async () => { + const release = deferred(); + const releasePresetServices = jest.fn().mockResolvedValue(undefined); + const entry = makeEntry( + 'evicted-entry', + jest.fn(() => release.promise), + releasePresetServices + ); + graphileCache.set(entry.cacheKey, entry); + graphileCache.delete(entry.cacheKey); + + let disposed = false; + const waiting = waitForEntryDisposal(entry).then(() => { + disposed = true; + }); + await flushPromises(); + expect(disposed).toBe(false); + + release.resolve(undefined); + await waiting; + expect(disposed).toBe(true); + expect(releasePresetServices).toHaveBeenCalledTimes(1); + }); + + it('does not resolve a cache clear before resident disposal completes', async () => { + const release = deferred(); + const entry = makeEntry( + 'clear-entry', + jest.fn(() => release.promise) + ); + graphileCache.set(entry.cacheKey, entry); + + let cleared = false; + const clearing = clearGraphileCache().then(() => { + cleared = true; + }); + await flushPromises(); + expect(graphileCache.size).toBe(0); + expect(cleared).toBe(false); + + release.resolve(undefined); + await clearing; + expect(cleared).toBe(true); + }); + + it('does not complete entry disposal before preset services are released', async () => { + const releasePresetServices = deferred(); + const entry = makeEntry( + 'delayed-preset-service', + jest.fn().mockResolvedValue(undefined), + jest.fn(() => releasePresetServices.promise) + ); + graphileCache.set(entry.cacheKey, entry); + + graphileCache.delete(entry.cacheKey); + const disposal = waitForEntryDisposal(entry); + let disposed = false; + void disposal.then(() => { + disposed = true; + }); + + await flushPromises(); + expect(disposed).toBe(false); + + releasePresetServices.resolve(undefined); + await disposal; + expect(disposed).toBe(true); + }); + + it('disposes a new generation after an older same-key generation fails', async () => { + const firstFailure = new Error('first generation release failed'); + const firstRelease = jest.fn().mockRejectedValue(firstFailure); + const secondRelease = jest.fn().mockResolvedValue(undefined); + const first = makeEntry('retry-by-generation', firstRelease); + const second = makeEntry('retry-by-generation', secondRelease); + + await expect(disposeUncachedEntry(first)).rejects.toBe(firstFailure); + await expect(disposeUncachedEntry(second)).resolves.toBeUndefined(); + + expect(firstRelease).toHaveBeenCalledTimes(1); + expect(secondRelease).toHaveBeenCalledTimes(1); + await expect(clearGraphileCache()).rejects.toBe(firstFailure); + }); +}); diff --git a/graphile/graphile-cache/src/__tests__/pg-service-release.test.ts b/graphile/graphile-cache/src/__tests__/pg-service-release.test.ts new file mode 100644 index 0000000000..01f44a9dfa --- /dev/null +++ b/graphile/graphile-cache/src/__tests__/pg-service-release.test.ts @@ -0,0 +1,71 @@ +import { type GetConnectionResult, getConnections } from 'pgsql-test'; +import { makePgService } from 'postgraphile/adaptors/pg'; + +import { createPresetServicesReleaser } from '../preset-services'; + +const waitFor = async (predicate: () => boolean): Promise => { + const deadline = Date.now() + 5000; + while (!predicate()) { + if (Date.now() >= deadline) throw new Error('Timed out waiting for subscriber cleanup'); + await new Promise(resolve => setTimeout(resolve, 5)); + } +}; + +describe('unpatched PostgreSQL service release', () => { + let fixture: GetConnectionResult; + + beforeAll(async () => { + fixture = await getConnections({}, []); + }); + beforeEach(async () => { + await fixture.pg.beforeEach(); + await fixture.db.beforeEach(); + }); + afterEach(async () => { + await fixture.db.afterEach(); + await fixture.pg.afterEach(); + }); + afterAll(async () => { + await fixture?.teardown(); + }); + + it('eventually returns the subscriber backend and removes its LISTEN channels', async () => { + const poolConfig = { ...fixture.pg.config, max: 1 }; + const pool = fixture.manager.getPool(poolConfig); + const service = makePgService({ pool, schemas: ['public'] }); + const release = createPresetServicesReleaser({ pgServices: [service] }); + try { + const iterator = await service.pgSubscriber!.subscribe('graphile_cache_release_test'); + let received = false; + const notification = iterator.next().then(result => { + expect(result.value).toBe('ready'); + received = true; + }); + // pg_notify must commit independently of the fixture's savepoint. + for (let attempt = 0; attempt < 100 && !received; attempt++) { + await fixture.pg.query("SELECT pg_notify('graphile_cache_release_test', 'ready')"); + await fixture.pg.publish(); + await new Promise(resolve => setTimeout(resolve, 10)); + } + expect(received).toBe(true); + await notification; + expect(pool.totalCount).toBe(1); + expect(pool.idleCount).toBe(0); + const before = await fixture.pg.query( + 'SELECT pid FROM pg_stat_activity WHERE datname = current_database() AND pid <> pg_backend_pid()' + ); + + await release(); + // The public call succeeding is the disposal contract. Pool return is a + // separate eventual observation, not a prerequisite for that success. + await waitFor(() => pool.idleCount === 1); + const channels = await pool.query('SELECT pg_listening_channels()'); + expect(channels.rows).toEqual([]); + const backend = await pool.query('SELECT pg_backend_pid() AS pid'); + expect(before.rows.map(row => row.pid)).toContain(backend.rows[0].pid); + expect(pool.totalCount).toBe(1); + } finally { + await release(); + } + }); +}); diff --git a/graphile/graphile-cache/src/__tests__/preset-services.test.ts b/graphile/graphile-cache/src/__tests__/preset-services.test.ts new file mode 100644 index 0000000000..dcc76edc81 --- /dev/null +++ b/graphile/graphile-cache/src/__tests__/preset-services.test.ts @@ -0,0 +1,69 @@ +import { createPresetServicesReleaser } from '../preset-services'; + +describe('preset service ownership', () => { + it('releases unique services in reverse order exactly once', async () => { + const events: string[] = []; + const first = { + release: jest.fn(async () => { + events.push('first'); + }), + }; + const second = { + release: jest.fn(async () => { + events.push('second'); + }), + }; + const release = createPresetServicesReleaser({ + pgServices: [first, second, first], + }); + + const releases = [release(), release(), release()]; + expect(releases[0]).toBe(releases[1]); + expect(releases[1]).toBe(releases[2]); + await Promise.all(releases); + + expect(events).toEqual(['second', 'first']); + expect(first.release).toHaveBeenCalledTimes(1); + expect(second.release).toHaveBeenCalledTimes(1); + }); + + it('continues releasing services and preserves the first error', async () => { + const firstFailure = new Error('second failed'); + const first = { release: jest.fn().mockResolvedValue(undefined) }; + const second = { release: jest.fn().mockRejectedValue(firstFailure) }; + const release = createPresetServicesReleaser({ + pgServices: [first, second], + }); + + await expect(release()).rejects.toBe(firstFailure); + expect(first.release).toHaveBeenCalledTimes(1); + expect(second.release).toHaveBeenCalledTimes(1); + }); + + it('accepts a void release without waiting for private background cleanup', async () => { + let finishCleanup!: () => void; + let cleaned = false; + const background = new Promise(resolve => { finishCleanup = resolve; }); + const service = { + release: jest.fn((): void => { + void background.then(() => { cleaned = true; }); + }), + }; + const release = createPresetServicesReleaser({ pgServices: [service] }); + + await release(); + expect(service.release).toHaveBeenCalledTimes(1); + expect(cleaned).toBe(false); + finishCleanup(); + await background; + expect(cleaned).toBe(true); + }); + + it('is a safe idempotent no-op when the preset has no services', async () => { + const release = createPresetServicesReleaser({}); + + const first = release(); + expect(release()).toBe(first); + await first; + }); +}); diff --git a/graphile/graphile-cache/src/build-identity.ts b/graphile/graphile-cache/src/build-identity.ts new file mode 100644 index 0000000000..6d356bab2f --- /dev/null +++ b/graphile/graphile-cache/src/build-identity.ts @@ -0,0 +1,234 @@ +import { errors } from '@constructive-io/errors'; +import { createHmac, randomBytes } from 'crypto'; + +/** A per-process key prevents cache fingerprints from exposing config values. */ +const fingerprintSecret = randomBytes(32); + +const referenceIds = new WeakMap(); +const symbolIds = new Map(); +let nextReferenceId = 1; + +export interface GraphileBuildReference { + readonly value: T; +} + +class GraphileBuildReferenceToken implements GraphileBuildReference { + constructor(readonly value: T) {} +} + +/** Compare a value by identity even when it is a plain object. */ +export const referenceGraphileBuildValue = ( + value: T +): GraphileBuildReference => Object.freeze(new GraphileBuildReferenceToken(value)); + +const invalidIdentity = (): never => { + throw errors.INTERNAL_FAILURE({ details: 'Invalid Graphile build identity' }); +}; + +const referenceId = (value: object | Function): number => { + let id = referenceIds.get(value as object); + if (id === undefined) { + id = nextReferenceId++; + referenceIds.set(value as object, id); + } + return id; +}; + +const symbolId = (value: symbol): number => { + let id = symbolIds.get(value); + if (id === undefined) { + id = nextReferenceId++; + symbolIds.set(value, id); + } + return id; +}; + +const numberValue = (value: number): string | number => { + if (Number.isNaN(value)) return 'NaN'; + if (value === Infinity) return '+Infinity'; + if (value === -Infinity) return '-Infinity'; + if (Object.is(value, -0)) return '-0'; + return value; +}; + +const isArrayIndex = (key: string, length: number): boolean => { + const index = Number(key); + return Number.isInteger(index) && index >= 0 && index < length && String(index) === key; +}; + +const canonicalObjectEntries = ( + value: object, + active: Set, + omit: (key: PropertyKey) => boolean = () => false +): Array<[unknown, unknown]> => { + const entries: Array<[unknown, unknown]> = []; + for (const key of Reflect.ownKeys(value)) { + if (omit(key)) continue; + const descriptor = Object.getOwnPropertyDescriptor(value, key); + if (!descriptor?.enumerable) continue; + if (!('value' in descriptor)) return invalidIdentity(); + const canonicalKey = typeof key === 'symbol' + ? ['symbol-key', symbolId(key)] + : ['string-key', key]; + entries.push([canonicalKey, canonicalize(descriptor.value, active)]); + } + entries.sort((left, right) => { + const leftKey = JSON.stringify(left[0]); + const rightKey = JSON.stringify(right[0]); + return leftKey < rightKey ? -1 : leftKey > rightKey ? 1 : 0; + }); + return entries; +}; + +const canonicalize = (value: unknown, active: Set): unknown => { + if (value === undefined) return ['undefined']; + if (value === null) return ['null']; + + switch (typeof value) { + case 'boolean': return ['boolean', value]; + case 'string': return ['string', value]; + case 'number': return ['number', numberValue(value)]; + case 'bigint': return ['bigint', value.toString()]; + case 'symbol': return ['symbol', symbolId(value)]; + case 'function': return ['function', referenceId(value)]; + case 'object': break; + default: return invalidIdentity(); + } + + const object = value as object; + if (value instanceof GraphileBuildReferenceToken) { + return ['explicit-reference', referenceId(value.value)]; + } + if (active.has(object)) return invalidIdentity(); + + if (Array.isArray(value)) { + active.add(object); + try { + const items = Array.from({ length: value.length }, (_, index) => + index in value ? canonicalize(value[index], active) : ['array-hole'] + ); + const extraEntries = canonicalObjectEntries(value, active, (key) => + key === 'length' || (typeof key === 'string' && isArrayIndex(key, value.length)) + ); + return ['array', value.length, items, extraEntries]; + } finally { + active.delete(object); + } + } + + if (Object.getPrototypeOf(value) === Date.prototype) { + const time = Date.prototype.getTime.call(value); + return ['date', Number.isNaN(time) ? 'Invalid Date' : time]; + } + + const prototype = Object.getPrototypeOf(value); + if (prototype !== Object.prototype && prototype !== null) { + return ['object-reference', referenceId(object)]; + } + + active.add(object); + try { + const entries = canonicalObjectEntries(value, active); + return [prototype === null ? 'null-prototype-object' : 'plain-object', entries]; + } finally { + active.delete(object); + } +}; + +/** + * Create a process-local, secret fingerprint for an exact Graphile build. + * Plain objects compare by sorted enumerable data properties, arrays retain + * order, and functions, symbols, and opaque instances compare by identity. + */ +export const createGraphileBuildCacheKey = ( + domain: 'server' | 'explorer', + snapshot: unknown +): string => { + try { + const canonical = JSON.stringify(['graphile-build', domain, canonicalize(snapshot, new Set())]); + const fingerprint = createHmac('sha256', fingerprintSecret).update(canonical).digest('hex'); + return `graphile:${domain}:${fingerprint}`; + } catch { + return invalidIdentity(); + } +}; + +const cloneSnapshotValue = (value: unknown, active: Set): unknown => { + if (value === null || typeof value !== 'object') return value; + + const object = value as object; + if (active.has(object)) return invalidIdentity(); + + if (Object.getPrototypeOf(value) === Date.prototype) { + const snapshot = new Date(Date.prototype.getTime.call(value)); + const mutators = [ + 'setTime', 'setMilliseconds', 'setUTCMilliseconds', 'setSeconds', 'setUTCSeconds', + 'setMinutes', 'setUTCMinutes', 'setHours', 'setUTCHours', 'setDate', 'setUTCDate', + 'setMonth', 'setUTCMonth', 'setFullYear', 'setUTCFullYear', 'setYear' + ]; + for (const name of mutators) { + Object.defineProperty(snapshot, name, { + value: invalidIdentity, + configurable: false, + writable: false + }); + } + return Object.freeze(snapshot); + } + + if (Array.isArray(value)) { + active.add(object); + try { + const result = new Array(value.length) as unknown[] & Record; + for (let index = 0; index < value.length; index++) { + if (index in value) result[index] = cloneSnapshotValue(value[index], active); + } + for (const key of Reflect.ownKeys(value)) { + if (key === 'length' || (typeof key === 'string' && isArrayIndex(key, value.length))) continue; + const descriptor = Object.getOwnPropertyDescriptor(value, key); + if (!descriptor?.enumerable) continue; + if (!('value' in descriptor)) return invalidIdentity(); + Object.defineProperty(result, key, { + value: cloneSnapshotValue(descriptor.value, active), + enumerable: true, + configurable: false, + writable: false + }); + } + return Object.freeze(result); + } finally { + active.delete(object); + } + } + + const prototype = Object.getPrototypeOf(value); + if (prototype !== Object.prototype && prototype !== null) return value; + + active.add(object); + try { + const result = Object.create(prototype) as Record; + for (const key of Reflect.ownKeys(value)) { + const descriptor = Object.getOwnPropertyDescriptor(value, key); + if (!descriptor?.enumerable) continue; + if (!('value' in descriptor)) return invalidIdentity(); + Object.defineProperty(result, key, { + value: cloneSnapshotValue(descriptor.value, active), + enumerable: true, + configurable: false, + writable: false + }); + } + return Object.freeze(result); + } finally { + active.delete(object); + } +}; + +/** Clone and freeze the plain data used by both fingerprinting and the build. */ +export const snapshotGraphileBuildValue = (value: T): Readonly => { + try { + return cloneSnapshotValue(value, new Set()) as Readonly; + } catch { + return invalidIdentity(); + } +}; diff --git a/graphile/graphile-cache/src/build-readiness.ts b/graphile/graphile-cache/src/build-readiness.ts new file mode 100644 index 0000000000..308d56f767 --- /dev/null +++ b/graphile/graphile-cache/src/build-readiness.ts @@ -0,0 +1,41 @@ +export interface GraphileBuildReadiness { + schemaResult: PromiseLike | unknown; + addTo(): PromiseLike | unknown; + ready(): PromiseLike | unknown; + release(): PromiseLike | unknown; + onReleaseError?(error: unknown): void; +} + +/** + * Resolve only after schema gathering and the HTTP adapter are ready. A failed + * generation reaches its release terminal state before the failure escapes. + */ +export const awaitGraphileBuildReadiness = async ( + build: GraphileBuildReadiness +): Promise => { + const schemaOutcome = Promise.resolve(build.schemaResult).then( + () => ({ ready: true as const }), + (error: unknown) => ({ ready: false as const, error }) + ); + try { + await build.addTo(); + const [schema] = await Promise.all([schemaOutcome, build.ready()]); + if ('error' in schema) throw schema.error; + } catch (error) { + // A failed adapter/readiness operation can race schema construction. Wait + // for schema work to settle before releasing the services it may still use. + await schemaOutcome; + try { + await build.release(); + } catch (releaseError) { + const failures = [error, releaseError]; + try { + build.onReleaseError?.(releaseError); + } catch (observerError) { + failures.push(observerError); + } + throw new AggregateError(failures, 'Graphile build and cleanup failed', { cause: error }); + } + throw error; + } +}; diff --git a/graphile/graphile-cache/src/create-instance.ts b/graphile/graphile-cache/src/create-instance.ts index a522701720..0ab4bb77f0 100644 --- a/graphile/graphile-cache/src/create-instance.ts +++ b/graphile/graphile-cache/src/create-instance.ts @@ -6,7 +6,9 @@ import { grafserv } from 'grafserv/express/v4'; import { RealtimeManager } from 'graphile-realtime-subscriptions'; import { postgraphile } from 'postgraphile'; -import type { GraphileCacheEntry } from './graphile-cache'; +import { awaitGraphileBuildReadiness } from './build-readiness'; +import { disposeUncachedEntry, type GraphileCacheEntry } from './graphile-cache'; +import { createPresetServicesReleaser } from './preset-services'; const log = new Logger('graphile-cache:create'); @@ -43,12 +45,45 @@ export const createGraphileInstance = async ( const { preset, cacheKey, enableRealtime = false } = opts; const pgl = postgraphile(preset); + const resolvedPreset = pgl.getResolvedPreset(); + const releasePresetServices = createPresetServicesReleaser(resolvedPreset); const serv = pgl.createServ(grafserv); const handler = express(); const httpServer = createServer(handler); - await serv.addTo(handler, httpServer); - await serv.ready(); + let failedBuildReleasePromise: Promise | null = null; + const releaseFailedBuild = (): Promise => { + if (failedBuildReleasePromise) return failedBuildReleasePromise; + failedBuildReleasePromise = (async () => { + const failures: unknown[] = []; + try { + await pgl.release(); + } catch (error) { + failures.push(error); + } + try { + await releasePresetServices(); + } catch (error) { + failures.push(error); + } + if (failures.length === 1) throw failures[0]; + if (failures.length > 1) throw new AggregateError(failures, 'Graphile service cleanup failed'); + })(); + return failedBuildReleasePromise; + }; + + await awaitGraphileBuildReadiness({ + schemaResult: pgl.getSchemaResult(), + addTo: () => serv.addTo(handler, httpServer), + ready: () => serv.ready(), + release: releaseFailedBuild, + onReleaseError: (releaseError) => { + log.error( + 'Graphile build cleanup failed', + { stage: 'failed-build-release' } + ); + } + }); const entry: GraphileCacheEntry = { pgl, @@ -57,6 +92,7 @@ export const createGraphileInstance = async ( httpServer, cacheKey, createdAt: Date.now(), + releasePresetServices }; if (enableRealtime) { @@ -64,15 +100,14 @@ export const createGraphileInstance = async ( // Extract PgSubscriber and pool from the resolved preset's pgServices. // The pool is the same instance managed by pg-cache (via getPgPool) // and threaded into the preset by makePgService({ pool, schemas }). - const resolvedPreset = pgl.getResolvedPreset(); const pgService = (resolvedPreset as any).pgServices?.[0]; const pgSubscriber = pgService?.pgSubscriber ?? null; const pool = pgService?.adaptorSettings?.pool ?? null; if (!pgSubscriber) { - log.warn(`PostGraphile[${cacheKey}] has no pgSubscriber — RealtimeManager will not be started`); + throw new Error('Realtime requires a PostgreSQL subscriber'); } else if (!pool) { - log.warn(`PostGraphile[${cacheKey}] has no pool in pgService — RealtimeManager will not be started`); + throw new Error('Realtime requires a PostgreSQL pool'); } else { const manager = new RealtimeManager({ pgSubscriber, @@ -81,12 +116,17 @@ export const createGraphileInstance = async ( schema: 'realtime_public', }); - await manager.start(); entry.realtimeManager = manager; + await manager.start(); log.info(`RealtimeManager started for PostGraphile[${cacheKey}]`); } } catch (err) { - log.error(`Failed to start RealtimeManager for PostGraphile[${cacheKey}]:`, err); + try { + await disposeUncachedEntry(entry); + } catch (cleanupError) { + throw new AggregateError([err, cleanupError], 'Realtime startup and cleanup failed', { cause: err }); + } + throw err; } } diff --git a/graphile/graphile-cache/src/graphile-cache.ts b/graphile/graphile-cache/src/graphile-cache.ts index 83782c6a21..611ae88d43 100644 --- a/graphile/graphile-cache/src/graphile-cache.ts +++ b/graphile/graphile-cache/src/graphile-cache.ts @@ -86,12 +86,24 @@ export interface GraphileCacheEntry { httpServer: HttpServer; cacheKey: string; createdAt: number; + /** Logical service that owns this build (for example an API key or explorer route). */ + serviceKey?: string; + /** Resolved database identifier, when the serving surface has one. */ + databaseId?: string | null; + /** pg-cache's physical pool key (currently the resolved database name). */ + poolKey?: string; + /** Idempotent release for pgServices owned by this exact preset generation. */ + releasePresetServices?: () => Promise; /** Optional RealtimeManager for cursor-tracked subscription delivery */ realtimeManager?: { stop(): Promise } | null; } -// Track disposed entries to prevent double-disposal -const disposedKeys = new Set(); +const disposalPromises = new WeakMap>(); +const activeDisposals = new Set>(); +// Retain failed generations until a bulk drain observes them. Successful work +// is removed immediately; exact-entry outcomes remain in the WeakMap above. +const disposalRecords = new Map>(); +let disposalSequence = 0; // Track keys that are being manually evicted for accurate eviction reason const manualEvictionKeys = new Set(); @@ -101,43 +113,96 @@ const manualEvictionKeys = new Set(); * * Properly releases resources by: * 1. Closing the HTTP server if listening - * 2. Releasing the PostGraphile instance (which internally releases grafserv) - * - * Uses disposedKeys set to prevent double-disposal when closeAllCaches() - * explicitly disposes entries and then clear() triggers the dispose callback. + * 2. Stopping the realtime manager + * 3. Releasing PostGraphile/Grafserv and preset services */ -const disposeEntry = async (entry: GraphileCacheEntry, key: string): Promise => { - // Prevent double-disposal - if (disposedKeys.has(key)) { - return; - } - disposedKeys.add(key); - +const releaseEntry = async ( + entry: GraphileCacheEntry, + key: string +): Promise => { log.debug(`Disposing PostGraphile[${key}]`); + const failures: unknown[] = []; try { - // Close HTTP server if it's listening if (entry.httpServer?.listening) { - await new Promise((resolve) => { - entry.httpServer.close(() => resolve()); + await new Promise((resolve, reject) => { + entry.httpServer.close((error) => error ? reject(error) : resolve()); }); } - // Stop RealtimeManager if present (before releasing PostGraphile) + } catch (error) { + failures.push(error); + } + try { if (entry.realtimeManager) { - try { - await entry.realtimeManager.stop(); - } catch (err) { - log.error(`Error stopping RealtimeManager for PostGraphile[${key}]:`, err); - } + await entry.realtimeManager.stop(); } - // Release PostGraphile instance (this also releases grafserv internally) - if (entry.pgl) { - await entry.pgl.release(); - } - } catch (err) { - log.error(`Error disposing PostGraphile[${key}]:`, err); - } finally { - disposedKeys.delete(key); + } catch (error) { + failures.push(error); + } + try { + await entry.pgl.release(); + } catch (error) { + failures.push(error); } + try { + await entry.releasePresetServices?.(); + } catch (error) { + failures.push(error); + } + if (failures.length === 1) throw failures[0]; + if (failures.length > 1) throw new AggregateError(failures, 'Graphile resource cleanup failed'); +}; + +/** + * Coalesce teardown by exact entry identity, not by its reusable cache key. + */ +const scheduleDisposal = ( + entry: GraphileCacheEntry, + key: string +): Promise => { + const existing = disposalPromises.get(entry); + if (existing) return existing; + + const pending = releaseEntry(entry, key); + + disposalPromises.set(entry, pending); + activeDisposals.add(pending); + const sequence = ++disposalSequence; + disposalRecords.set(sequence, pending); + void pending + .then(() => { disposalRecords.delete(sequence); }, () => { + // The original rejection remains observable through the owner APIs. + log.error('Graphile resource cleanup failed', { generation: sequence }); + }) + .finally(() => activeDisposals.delete(pending)); + return pending; +}; + +/** Dispose a generation that was built but never published in the cache. */ +export const disposeUncachedEntry = ( + entry: GraphileCacheEntry, + key = entry.cacheKey +): Promise => scheduleDisposal(entry, key); + +/** Await disposal calls, not background connection cleanup inside upstream. */ +export const waitForEntryDisposal = ( + entry: GraphileCacheEntry +): Promise => disposalPromises.get(entry) ?? Promise.resolve(); + +/** + * Await disposals scheduled before this call, including failures that already + * settled. Later generations belong to the next drain. Concurrent callers hold + * their own snapshot, so acknowledging a failure cannot hide it from a peer. + */ +export const waitForActiveDisposals = async (): Promise => { + const records = [...disposalRecords]; + const outcomes = await Promise.allSettled(records.map(([, pending]) => pending)); + const failures: unknown[] = []; + outcomes.forEach((outcome, index) => { + disposalRecords.delete(records[index][0]); + if (outcome.status === 'rejected') failures.push(outcome.reason); + }); + if (failures.length === 1) throw failures[0]; + if (failures.length > 1) throw new AggregateError(failures, 'Graphile disposal drain failed'); }; /** @@ -176,11 +241,7 @@ export const graphileCache = new LRUCache({ log.debug(`Evicting PostGraphile[${key}] (reason: ${reason})`); - // LRU dispose is synchronous, but v5 disposal is async - // Fire and forget the async cleanup - disposeEntry(entry, key).catch((err) => { - log.error(`Failed to dispose PostGraphile[${key}]:`, err); - }); + scheduleDisposal(entry, key); } }); @@ -227,24 +288,56 @@ export function clearMatchingEntries(pattern: RegExp): number { return cleared; } +const clearEntries = (matches: (key: string, entry: GraphileCacheEntry) => boolean): number => { + let cleared = 0; + for (const [key, entry] of graphileCache.entries()) { + if (!matches(key, entry)) continue; + manualEvictionKeys.add(key); + graphileCache.delete(key); + cleared++; + } + return cleared; +}; + +/** Clear all cached build variants owned by one logical service. */ +export const clearGraphileEntriesForService = (serviceKey: string): number => { + if (typeof serviceKey !== 'string' || serviceKey.length === 0) return 0; + return clearEntries((key, entry) => entry.serviceKey === serviceKey || (!entry.serviceKey && key === serviceKey)); +}; + +/** Clear all cached build variants for one resolved database identifier. */ +export const clearGraphileEntriesForDatabase = (databaseId: string): number => + clearEntries((_key, entry) => entry.databaseId === databaseId); + +/** Clear all cached build variants backed by one pg-cache pool key. */ +export const clearGraphileEntriesForPool = (poolKey: string): number => + clearEntries((_key, entry) => entry.poolKey === poolKey); + + // Register cleanup callback with pgCache // When a pg pool is disposed, clean up any graphile instances using it const unregister = pgCache.registerCleanupCallback((pgPoolKey: string) => { log.debug(`pgPool[${pgPoolKey}] disposed - checking graphile entries`); - - // Remove graphile entries that reference this pool key - graphileCache.forEach((entry, k) => { - if (entry.cacheKey.includes(pgPoolKey)) { - log.debug(`Removing graphileCache[${k}] due to pgPool[${pgPoolKey}] disposal`); - manualEvictionKeys.add(k); - graphileCache.delete(k); - } - }); + const cleared = clearGraphileEntriesForPool(pgPoolKey); + if (cleared > 0) log.debug(`Removed ${cleared} graphile entries for pgPool[${pgPoolKey}]`); }); // Enhanced close function that handles all caches const closePromise: { promise: Promise | null } = { promise: null }; +/** Clear all resident entries and await every exact-generation disposal. */ +export const clearGraphileCache = async (): Promise => { + for (const key of graphileCache.keys()) { + manualEvictionKeys.add(key); + } + graphileCache.clear(); + try { + await waitForActiveDisposals(); + } finally { + manualEvictionKeys.clear(); + } +}; + /** * Close all caches and release resources * @@ -263,30 +356,20 @@ export const closeAllCaches = async (verbose = false): Promise => { try { if (verbose) log.info('Closing all server caches...'); - // Collect all entries and dispose them properly - const entries = [...graphileCache.entries()]; - - // Mark all as manual evictions - for (const [key] of entries) { - manualEvictionKeys.add(key); + const failures: unknown[] = []; + try { + await clearGraphileCache(); + } catch (error) { + failures.push(error); } - - const disposePromises = entries.map(([key, entry]) => - disposeEntry(entry, key) - ); - - // Wait for all disposals to complete - await Promise.allSettled(disposePromises); - - // Clear the cache after disposal (dispose callback will no-op due to disposedKeys) - graphileCache.clear(); - - // Clear disposed keys tracking after full cleanup - disposedKeys.clear(); - manualEvictionKeys.clear(); - - // Close pg pools - await pgCache.close(); + // Attempt pool cleanup even when a generation's release rejected. + try { + await pgCache.close(); + } catch (error) { + failures.push(error); + } + if (failures.length === 1) throw failures[0]; + if (failures.length > 1) throw new AggregateError(failures, 'Server cache cleanup failed'); if (verbose) log.success('All caches disposed.'); } finally { diff --git a/graphile/graphile-cache/src/index.ts b/graphile/graphile-cache/src/index.ts index 9a845fafe3..e02ec2c6de 100644 --- a/graphile/graphile-cache/src/index.ts +++ b/graphile/graphile-cache/src/index.ts @@ -8,9 +8,14 @@ export { CacheEvictionEvent, // Cache stats CacheStats, + clearGraphileCache, + clearGraphileEntriesForDatabase, + clearGraphileEntriesForPool, + clearGraphileEntriesForService, // Clear matching entries clearMatchingEntries, closeAllCaches, + disposeUncachedEntry, // Eviction tracking EvictionReason, FIVE_MINUTES_MS, @@ -20,7 +25,16 @@ export { graphileCache, GraphileCacheEntry, // Time constants - ONE_HOUR_MS} from './graphile-cache'; + ONE_HOUR_MS, + waitForActiveDisposals, + waitForEntryDisposal} from './graphile-cache'; + +export { + createGraphileBuildCacheKey, + referenceGraphileBuildValue, + snapshotGraphileBuildValue +} from './build-identity'; +export type { GraphileBuildReference } from './build-identity'; // Factory for creating PostGraphile v5 instances export { createGraphileInstance } from './create-instance'; diff --git a/graphile/graphile-cache/src/preset-services.ts b/graphile/graphile-cache/src/preset-services.ts new file mode 100644 index 0000000000..5bb2109345 --- /dev/null +++ b/graphile/graphile-cache/src/preset-services.ts @@ -0,0 +1,33 @@ +interface ReleasablePresetService { + release?: () => void | PromiseLike; +} + +/** + * Own and release the unique pgServices for one resolved preset generation. + * Success means the public release calls returned successfully (including any + * promise they expose). Upstream may continue subscriber UNLISTEN/client return + * in the background; this does not promise that the pool is already idle. + */ +export const createPresetServicesReleaser = (resolvedPreset: { + pgServices?: readonly ReleasablePresetService[]; +}): (() => Promise) => { + const services = [...new Set(resolvedPreset.pgServices ?? [])]; + let releasePromise: Promise | null = null; + + return (): Promise => { + if (releasePromise) return releasePromise; + releasePromise = (async () => { + const failures: unknown[] = []; + for (const service of [...services].reverse()) { + try { + await service.release?.(); + } catch (error) { + failures.push(error); + } + } + if (failures.length === 1) throw failures[0]; + if (failures.length > 1) throw new AggregateError(failures, 'Graphile service cleanup failed'); + })(); + return releasePromise; + }; +}; diff --git a/graphql/explorer/package.json b/graphql/explorer/package.json index 288768f38f..f3eb7b8911 100644 --- a/graphql/explorer/package.json +++ b/graphql/explorer/package.json @@ -34,6 +34,8 @@ "test:watch": "jest --watch" }, "dependencies": { + "@constructive-io/errors": "workspace:^", + "@pgpmjs/logger": "workspace:^", "@constructive-io/graphql-env": "workspace:^", "@constructive-io/graphql-types": "workspace:^", "@constructive-io/url-domains": "workspace:^", diff --git a/graphql/explorer/src/server.ts b/graphql/explorer/src/server.ts index 26035d5e12..fff3d878d5 100644 --- a/graphql/explorer/src/server.ts +++ b/graphql/explorer/src/server.ts @@ -1,9 +1,19 @@ +import { normalizeError } from '@constructive-io/errors'; +import { Logger } from '@pgpmjs/logger'; import { getEnvOptions } from '@constructive-io/graphql-env'; import type { ConstructiveOptions } from '@constructive-io/graphql-types'; import { middleware as parseDomains } from '@constructive-io/url-domains'; import { cors, healthz, poweredBy } from '@pgpmjs/server-utils'; import express, { Express, NextFunction, Request, Response } from 'express'; -import { createGraphileInstance, graphileCache, GraphileCacheEntry } from 'graphile-cache'; +import { + clearGraphileEntriesForService, + createGraphileBuildCacheKey, + createGraphileInstance, + graphileCache, + GraphileCacheEntry, + referenceGraphileBuildValue, + snapshotGraphileBuildValue +} from 'graphile-cache'; import type { GraphileConfig } from 'graphile-config'; import { makePgService } from 'graphile-settings'; import { getPgPool } from 'pg-cache'; @@ -12,8 +22,20 @@ import { getPgEnvOptions } from 'pg-env'; import { printDatabases, printSchemas } from './render'; import { getGraphilePreset } from './settings'; +const log = new Logger('graphql-explorer'); +const respondError = (res: Response, error: unknown): void => { + if ((error as { code?: string })?.code === '3D000') { + res.status(404).send('Database not found'); + return; + } + const failure = normalizeError(error); + log.error('Explorer request refused', { code: failure.code }); + res.status(failure.http).json({ errors: [{ message: failure.message, extensions: failure.toExtensions() }] }); +}; + export const GraphQLExplorer = (rawOpts: ConstructiveOptions = {}): Express => { const opts = getEnvOptions(rawOpts); + const ownerIdentity = {}; const { pg, server } = opts; @@ -21,13 +43,6 @@ export const GraphQLExplorer = (rawOpts: ConstructiveOptions = {}): Express => { dbname: string, schemaname: string ): Promise => { - const key = `${dbname}.${schemaname}`; - - const cached = graphileCache.get(key); - if (cached) { - return cached; - } - const pgConfig = getPgEnvOptions({ ...pg, database: dbname, @@ -37,20 +52,45 @@ export const GraphQLExplorer = (rawOpts: ConstructiveOptions = {}): Express => { // properly, preventing leaked connections during database teardown. const pool = getPgPool(pgConfig); - const basePreset = getGraphilePreset(opts); - const preset: GraphileConfig.Preset = { - ...basePreset, - pgServices: [ - makePgService({ pool, schemas: [schemaname] }), - ], - grafserv: { + const serviceKey = `${dbname}.${schemaname}`; + const snapshot = snapshotGraphileBuildValue({ + ownerIdentity: referenceGraphileBuildValue(ownerIdentity), + serviceKey, + poolIdentity: referenceGraphileBuildValue(pool), + poolKey: pgConfig.database, + pgConfig, + databaseName: dbname, + databaseId: null, + apiId: null, + schemas: [schemaname], + role: pg.user ?? 'postgres', + surface: { graphqlPath: '/graphql', graphiqlPath: '/graphiql', graphiql: true, }, + explain: undefined, + enableRealtime: false, + }); + const key = createGraphileBuildCacheKey('explorer', snapshot); + const cached = graphileCache.get(key); + if (cached) return cached; + + const basePreset = getGraphilePreset(opts, snapshot.role); + const preset: GraphileConfig.Preset = { + ...basePreset, + pgServices: [ + makePgService({ pool, schemas: snapshot.schemas }), + ], + grafserv: snapshot.surface, }; const instance = await createGraphileInstance({ preset, cacheKey: key }); + Object.assign(instance, { + serviceKey, + databaseId: snapshot.databaseId, + poolKey: snapshot.poolKey, + } satisfies Partial); graphileCache.set(key, instance); return instance; }; @@ -89,12 +129,7 @@ export const GraphQLExplorer = (rawOpts: ConstructiveOptions = {}): Express => { ); return; } catch (e: any) { - if (e.message?.match(/does not exist/)) { - res.status(404).send('DB Not found'); - return; - } - console.error(e); - res.status(500).send('Something happened...'); + respondError(res, e); return; } } @@ -114,12 +149,7 @@ export const GraphQLExplorer = (rawOpts: ConstructiveOptions = {}): Express => { await pgPool.query('SELECT 1;'); } catch (e: any) { - if (e.message?.match(/does not exist/)) { - res.status(404).send('DB Not found'); - return; - } - console.error(e); - res.status(500).send('Something happened...'); + respondError(res, e); return; } } @@ -130,28 +160,22 @@ export const GraphQLExplorer = (rawOpts: ConstructiveOptions = {}): Express => { if (req.urlDomains?.subdomains.length === 2) { const [schemaName, dbName] = req.urlDomains.subdomains; try { + if (req.url === '/flush') { + clearGraphileEntriesForService(`${dbName}.${schemaName}`); + res.status(200).send('OK'); + return; + } const instance = await getGraphileInstanceObj(dbName, schemaName); instance.handler(req, res, next); return; } catch (e: any) { - res.status(500).send(e.message); + respondError(res, e); return; } } return next(); }); - app.use(async (req: Request, res: Response, next: NextFunction) => { - if (req.urlDomains?.subdomains.length === 2 && req.url === '/flush') { - const [schemaName, dbName] = req.urlDomains.subdomains; - const key = `${dbName}.${schemaName}`; - graphileCache.delete(key); - res.status(200).send('OK'); - return; - } - return next(); - }); - app.use(async (req: Request, res: Response, next: NextFunction) => { if (req.urlDomains?.subdomains.length === 0) { try { @@ -171,12 +195,7 @@ export const GraphQLExplorer = (rawOpts: ConstructiveOptions = {}): Express => { ); return; } catch (e: any) { - if (e.message?.match(/does not exist/)) { - res.status(404).send('DB Not found'); - return; - } - console.error(e); - res.status(500).send('Something happened...'); + respondError(res, e); return; } } diff --git a/graphql/explorer/src/settings.ts b/graphql/explorer/src/settings.ts index 4d2047f9f1..8890edfd70 100644 --- a/graphql/explorer/src/settings.ts +++ b/graphql/explorer/src/settings.ts @@ -8,14 +8,17 @@ import { ConstructivePreset } from 'graphile-settings'; * * This returns a v5 preset that can be extended with pgServices. */ -export const getGraphilePreset = (rawOpts: ConstructiveOptions): GraphileConfig.Preset => { - const opts = getEnvOptions(rawOpts); +export const getGraphilePreset = ( + rawOpts: ConstructiveOptions, + effectiveRole?: string +): GraphileConfig.Preset => { + const role = effectiveRole ?? getEnvOptions(rawOpts).pg?.user ?? 'postgres'; return { extends: [ConstructivePreset], grafast: { context: () => ({ - pgSettings: { role: opts.pg?.user ?? 'postgres' }, + pgSettings: { role }, }), }, }; diff --git a/graphql/server-test/__tests__/graphile-build.integration.test.ts b/graphql/server-test/__tests__/graphile-build.integration.test.ts new file mode 100644 index 0000000000..622cd77085 --- /dev/null +++ b/graphql/server-test/__tests__/graphile-build.integration.test.ts @@ -0,0 +1,391 @@ +/** + * Graphile build and cache lifecycle integration tests. + * + * These tests use the scoped routing fixture and a real HTTP server. The + * caller plugins only control the Graphile build boundary; they do not mock + * any Constructive cache or middleware internals. + */ + +import type { GraphileOptions } from '@constructive-io/graphql-types'; +import { clearGraphileCache, graphileCache } from 'graphile-cache'; +import path from 'path'; +import type supertest from 'supertest'; + +import { getConnections, seed } from '../src'; + +jest.setTimeout(120000); + +const sharedSeedRoot = path.join(__dirname, '..', '..', '..', '__fixtures__', 'seed'); +const shared = (...segments: string[]) => path.join(sharedSeedRoot, ...segments); +const pgpmWorkspace = path.join(sharedSeedRoot, '..', '..'); +const schemas = ['simple-pets-public', 'simple-pets-pets-public']; +const scopedMetaSchemas = [ + 'catalog_private', + 'routing_public', + 'apps_public', + 'metaschema_public', + 'metaschema_modules_public' +]; +const appHost = 'app.test.constructive.io'; +const flushToken = 'f17-flush-secret'; +const secretBuildMessage = 'f17-private-build-marker'; +const animalsQuery = '{ animals { nodes { name species } } }'; + +const scopedSeedAdapters = () => [ + seed.pgpm(pgpmWorkspace), + seed.sqlfile([ + shared('app-schemas', 'simple-pets', 'schema.sql'), + shared('scoped', 'test-data.sql'), + shared('app-schemas', 'simple-pets', 'test-data.sql') + ]) +]; + +/** Keep build failures from entering Graphile's upstream retry loop. */ +const graphileOptions = (plugin?: unknown): GraphileOptions => + ({ + preset: { + schema: { retryOnInitFail: false }, + ...(plugin ? { plugins: [plugin] } : {}) + } + } as unknown as GraphileOptions); + +interface BuildControl { + plugin: unknown; + entered: Promise; + release(): void; + buildCount(): number; +} + +interface AnimalsResponseBody { + data?: { + animals?: { + nodes?: Array<{ name: string }>; + }; + }; +} + +const createBuildControl = ( + onFirstBuild: () => Promise | void = async () => undefined +): BuildControl => { + let builds = 0; + let resolveEntered!: () => void; + let resolveGate!: () => void; + const entered = new Promise((resolve) => { + resolveEntered = resolve; + }); + const gate = new Promise((resolve) => { + resolveGate = resolve; + }); + + const plugin = { + name: 'F17GraphileBuildTestPlugin', + gather: { + namespace: 'f17GraphileBuildTest', + main: async () => { + builds += 1; + resolveEntered(); + await onFirstBuild(); + await gate; + } + } + }; + + return { + plugin, + entered, + release: () => resolveGate(), + buildCount: () => builds + }; +}; + +const createFailOnceBuildControl = (): BuildControl => { + let builds = 0; + let resolveEntered!: () => void; + let resolveGate!: () => void; + const entered = new Promise((resolve) => { + resolveEntered = resolve; + }); + const gate = new Promise((resolve) => { + resolveGate = resolve; + }); + + const plugin = { + name: 'F17GraphileBuildFailOnceTestPlugin', + gather: { + namespace: 'f17GraphileBuildFailOnceTest', + main: async () => { + builds += 1; + if (builds === 1) { + resolveEntered(); + await gate; + throw new Error(secretBuildMessage); + } + } + } + }; + + return { + plugin, + entered, + release: () => resolveGate(), + buildCount: () => builds + }; +}; + +const waitForSignal = async ( + signal: Promise, + label: string, + timeoutMs = 90000 +): Promise => { + await new Promise((resolve, reject) => { + const timer = setTimeout(() => { + reject(new Error(`Timed out waiting for ${label}`)); + }, timeoutMs); + signal.then( + () => { + clearTimeout(timer); + resolve(); + }, + (error: unknown) => { + clearTimeout(timer); + reject(error); + } + ); + }); +}; + +const nextTurn = (): Promise => + new Promise((resolve) => setImmediate(resolve)); + +// SuperTest starts the socket when `.then()` is called, but the request still +// has to traverse the real routing/context middleware before it can observe +// the in-flight build. Give that second request a bounded event-loop window +// while the first caller remains held by the plugin gate. +const allowConcurrentRequestToReachGraphile = async (): Promise => { + await nextTurn(); + await new Promise((resolve) => setTimeout(resolve, 500)); +}; + +describe('Graphile build and cache over the real scoped HTTP server', () => { + let request: supertest.Agent | undefined; + let teardown: (() => Promise) | undefined; + let transactionCleanup: (() => Promise) | undefined; + let releaseBuild: (() => void) | undefined; + const pendingRequests = new Set>(); + + const trackRequest = (promise: Promise): Promise => { + pendingRequests.add(promise); + promise.then( + () => pendingRequests.delete(promise), + () => pendingRequests.delete(promise) + ); + return promise; + }; + + const startServer = async (plugin?: unknown, token?: string): Promise => { + const connection = await getConnections( + { + schemas, + authRole: 'anonymous', + graphile: graphileOptions(plugin), + server: { + useRouting: true, + api: { + isPublic: true, + metaSchemas: scopedMetaSchemas, + ...(token ? { flushToken: token } : {}) + } + } + }, + scopedSeedAdapters() + ); + request = connection.request; + teardown = connection.teardown; + transactionCleanup = async () => { + try { + await connection.db.afterEach(); + } finally { + await connection.pg.afterEach(); + } + }; + await connection.pg.beforeEach(); + await connection.db.beforeEach(); + }; + + const postGraphQL = () => { + if (!request) throw new Error('Test server has not started'); + const test = request + .post('/graphql') + .set('Host', appHost) + .send({ query: animalsQuery }); + return trackRequest(test.then((response) => response)); + }; + + const flush = (method: 'get' | 'post', authorization?: string) => { + if (!request) throw new Error('Test server has not started'); + let req = (method === 'get' ? request.get('/flush') : request.post('/flush')) + .set('Host', appHost); + if (authorization !== undefined) req = req.set('Authorization', authorization); + return trackRequest(req.then((response) => response)); + }; + + const expectAnimals = (response: { status: number; body: AnimalsResponseBody }): void => { + expect(response.status).toBe(200); + const nodes = response.body.data?.animals?.nodes; + expect(nodes).toBeDefined(); + const names = nodes!.map( + (node: { name: string }) => node.name + ); + expect(names).toEqual( + expect.arrayContaining(['Buddy', 'Max', 'Whiskers', 'Mittens', 'Tweety']) + ); + }; + + beforeEach(async () => { + await clearGraphileCache(); + }); + + afterEach(async () => { + releaseBuild?.(); + releaseBuild = undefined; + const pending = [...pendingRequests]; + const currentTeardown = teardown; + const currentTransactionCleanup = transactionCleanup; + teardown = undefined; + transactionCleanup = undefined; + request = undefined; + try { + try { + await waitForSignal( + Promise.allSettled(pending).then((): void => undefined), + 'HTTP requests to finish', + 15000 + ); + } finally { + pendingRequests.clear(); + } + try { + // Drain schema handlers before pgsql-test rolls back and drops the + // fixture database. The server's close path also drains this cache, but + // the explicit ordering keeps failed tests from poisoning teardown. + await clearGraphileCache(); + } finally { + await currentTransactionCleanup?.(); + } + } finally { + await currentTeardown?.(); + } + }); + + it('serves the seeded schema through scoped routing', async () => { + await startServer(); + + expectAnimals(await postGraphQL()); + }); + + it('coalesces concurrent requests while a caller plugin defers schema build', async () => { + const control = createBuildControl(); + releaseBuild = control.release; + await startServer(control.plugin); + + const first = postGraphQL(); + await waitForSignal(control.entered, 'the deferred Graphile build'); + + const second = postGraphQL(); + await allowConcurrentRequestToReachGraphile(); + expect(control.buildCount()).toBe(1); + + control.release(); + releaseBuild = undefined; + const [firstResponse, secondResponse] = await Promise.all([first, second]); + expectAnimals(firstResponse); + expectAnimals(secondResponse); + expect(control.buildCount()).toBe(1); + }); + + it('maps a failed build to a safe HTTP error and retries after coalescing ends', async () => { + const control = createFailOnceBuildControl(); + releaseBuild = control.release; + await startServer(control.plugin); + + const first = postGraphQL(); + await waitForSignal(control.entered, 'the failing Graphile build'); + + const second = postGraphQL(); + await allowConcurrentRequestToReachGraphile(); + expect(control.buildCount()).toBe(1); + + control.release(); + releaseBuild = undefined; + const [firstResponse, secondResponse] = await Promise.all([first, second]); + for (const response of [firstResponse, secondResponse]) { + expect(response.status).toBe(500); + expect(response.body.errors[0].extensions).toMatchObject({ + code: 'INTERNAL_FAILURE', + http: 500 + }); + expect(JSON.stringify(response.body)).not.toContain(secretBuildMessage); + } + + // The failed generation must be removed from the in-flight map. A later + // request gets a fresh build, which this plugin allows to complete. + expectAnimals(await postGraphQL()); + expect(control.buildCount()).toBe(2); + }); + + it('clears the real service cache for authenticated POST and GET flushes', async () => { + const control = createBuildControl(async () => undefined); + releaseBuild = control.release; + control.release(); + await startServer(control.plugin, flushToken); + + expectAnimals(await postGraphQL()); + expect(control.buildCount()).toBe(1); + + const postFlush = await flush('post', `Bearer ${flushToken}`); + expect(postFlush.status).toBe(200); + expect(postFlush.text).toBe('OK'); + expect(graphileCache.size).toBe(0); + expectAnimals(await postGraphQL()); + expect(control.buildCount()).toBe(2); + + const getFlush = await flush('get', `Bearer ${flushToken}`); + expect(getFlush.status).toBe(200); + expect(getFlush.text).toBe('OK'); + expect(graphileCache.size).toBe(0); + }); + + it('rejects missing and wrong flush tokens without clearing the service cache', async () => { + const control = createBuildControl(async () => undefined); + releaseBuild = control.release; + control.release(); + await startServer(control.plugin, flushToken); + + expectAnimals(await postGraphQL()); + expect(control.buildCount()).toBe(1); + + expect((await flush('post')).status).toBe(401); + expect((await flush('get', 'Bearer wrong-token')).status).toBe(401); + expectAnimals(await postGraphQL()); + expect(control.buildCount()).toBe(1); + }); + + it('passes an unrelated health request without building Graphile', async () => { + const control = createBuildControl(async () => undefined); + releaseBuild = control.release; + await startServer(control.plugin, flushToken); + + if (!request) throw new Error('Test server has not started'); + const response = await trackRequest( + request.get('/healthz').set('Host', appHost).then((result) => result) + ); + expect(response.status).toBe(200); + expect(control.buildCount()).toBe(0); + }); + + it('keeps flush closed when no token is configured', async () => { + await startServer(); + + expect((await flush('post', `Bearer ${flushToken}`)).status).toBe(404); + }); +}); diff --git a/graphql/server-test/package.json b/graphql/server-test/package.json index 7aaace72ca..d42ad9ab24 100644 --- a/graphql/server-test/package.json +++ b/graphql/server-test/package.json @@ -39,6 +39,7 @@ "@types/supertest": "^7.2.1", "appstash": "^0.8.0", "gql-ast": "workspace:^", + "graphile-cache": "workspace:^", "inquirerer": "^4.9.4", "makage": "^0.8.0", "nested-obj": "^0.2.3" diff --git a/graphql/server/src/errors/__tests__/graphql-response.test.ts b/graphql/server/src/errors/__tests__/graphql-response.test.ts index ef4645afb5..f8db262aff 100644 --- a/graphql/server/src/errors/__tests__/graphql-response.test.ts +++ b/graphql/server/src/errors/__tests__/graphql-response.test.ts @@ -1,4 +1,4 @@ -import { errors } from '@constructive-io/errors'; +import { errors, normalizeError } from '@constructive-io/errors'; import type { Response } from 'express'; import { respondWithGraphQLError } from '../graphql-response'; @@ -10,6 +10,15 @@ const createMockResponse = () => { }; describe('respondWithGraphQLError', () => { + it('sends the normalized refusal status without internal cause or credentials', () => { + const { res, json } = createMockResponse(); + const failure = normalizeError(new Error('password=private tenant=hidden')); + respondWithGraphQLError(res, failure, { status: failure.http }); + expect(res.status).toHaveBeenCalledWith(500); + expect(json.mock.calls[0][0].errors[0].extensions.code).toBe('INTERNAL_FAILURE'); + expect(JSON.stringify(json.mock.calls)).not.toMatch(/private|hidden|cause|stack/); + }); + it('always emits a top-level message so clients never render an empty error', () => { const { res, json } = createMockResponse(); diff --git a/graphql/server/src/middleware/__tests__/flush.test.ts b/graphql/server/src/middleware/__tests__/flush.test.ts deleted file mode 100644 index a5a44d1671..0000000000 --- a/graphql/server/src/middleware/__tests__/flush.test.ts +++ /dev/null @@ -1,100 +0,0 @@ -import { ConstructiveOptions } from '@constructive-io/graphql-types'; -import { svcCache } from '@pgpmjs/server-utils'; -import { NextFunction, Request, Response } from 'express'; -import { graphileCache } from 'graphile-cache'; - -import { createFlushMiddleware } from '../flush'; - -const TOKEN = 'flush-secret-token'; - -const makeReq = (url: string, authorization?: string): Request => { - const req: any = { - url, - svc_key: 'tenant.example.com', - get: (name: string) => - name.toLowerCase() === 'authorization' ? authorization : undefined - }; - return req as Request; -}; - -const makeRes = (): Response & { statusCode?: number; body?: string } => { - const res: any = { - status(code: number) { - res.statusCode = code; - return res; - }, - send(body: string) { - res.body = body; - return res; - } - }; - return res; -}; - -const opts = (flushToken?: string): ConstructiveOptions => - ({ api: { ...(flushToken && { flushToken }) } } as ConstructiveOptions); - -describe('createFlushMiddleware', () => { - let next: NextFunction; - - beforeEach(() => { - graphileCache.set('tenant.example.com', { cached: true } as any); - svcCache.set('tenant.example.com', { cached: true } as any); - next = jest.fn(); - }); - - afterEach(() => { - graphileCache.delete('tenant.example.com'); - svcCache.delete('tenant.example.com'); - }); - - it('passes non-flush requests through', async () => { - const res = makeRes(); - await createFlushMiddleware(opts(TOKEN))(makeReq('/graphql'), res, next); - expect(next).toHaveBeenCalled(); - expect(res.statusCode).toBeUndefined(); - }); - - it('flushes the caches for a request carrying the token', async () => { - const res = makeRes(); - await createFlushMiddleware(opts(TOKEN))( - makeReq('/flush', `Bearer ${TOKEN}`), - res, - next - ); - expect(res.statusCode).toBe(200); - expect(graphileCache.get('tenant.example.com')).toBeUndefined(); - expect(svcCache.get('tenant.example.com')).toBeUndefined(); - }); - - it.each([ - ['no authorization header', undefined], - ['a wrong token', 'Bearer nope'], - ['a token of the same length', `Bearer ${'x'.repeat(TOKEN.length)}`], - ['the token without the bearer scheme', TOKEN], - ['a basic credential', `Basic ${TOKEN}`] - ])('rejects a flush request with %s', async (_label, authorization) => { - const res = makeRes(); - await createFlushMiddleware(opts(TOKEN))( - makeReq('/flush', authorization), - res, - next - ); - expect(res.statusCode).toBe(401); - expect(next).not.toHaveBeenCalled(); - expect(graphileCache.get('tenant.example.com')).toBeDefined(); - expect(svcCache.get('tenant.example.com')).toBeDefined(); - }); - - it('keeps the route closed when no token is configured', async () => { - const res = makeRes(); - await createFlushMiddleware(opts())( - makeReq('/flush', `Bearer ${TOKEN}`), - res, - next - ); - expect(res.statusCode).toBe(404); - expect(graphileCache.get('tenant.example.com')).toBeDefined(); - expect(svcCache.get('tenant.example.com')).toBeDefined(); - }); -}); diff --git a/graphql/server/src/middleware/__tests__/graphile-build-observability.test.ts b/graphql/server/src/middleware/__tests__/graphile-build-observability.test.ts new file mode 100644 index 0000000000..902bd1ef5f --- /dev/null +++ b/graphql/server/src/middleware/__tests__/graphile-build-observability.test.ts @@ -0,0 +1,15 @@ +import { getGraphileBuildStats, observeGraphileBuild, resetGraphileBuildStats } from '../observability/graphile-build-stats'; + +it('retains the original build error while recording only its safe canonical code', async () => { + resetGraphileBuildStats(); + const failure = new Error('private database password'); + await expect(observeGraphileBuild( + { cacheKey: 'build-key', serviceKey: 'service', databaseId: 'database' }, + async () => { throw failure; }, + { enabled: true } + )).rejects.toBe(failure); + const stats = getGraphileBuildStats(); + expect(stats.lastError).toBe('INTERNAL_FAILURE'); + expect(JSON.stringify(stats)).not.toContain(failure.message); + resetGraphileBuildStats(); +}); diff --git a/graphql/server/src/middleware/__tests__/graphile-build-snapshot.test.ts b/graphql/server/src/middleware/__tests__/graphile-build-snapshot.test.ts new file mode 100644 index 0000000000..4176151840 --- /dev/null +++ b/graphql/server/src/middleware/__tests__/graphile-build-snapshot.test.ts @@ -0,0 +1,94 @@ +import type { DatabaseSettings } from '@constructive-io/express-context'; +import type { Pool } from 'pg'; + +import { createGraphileBuildCacheKey } from 'graphile-cache'; + +import { createGraphileServerBuildSnapshot } from '../graphile-build-snapshot'; + +const owner = {}; +const pool = {} as Pool; +const maskError = (): void => {}; + +const makeInput = (overrides: Record = {}) => ({ + ownerIdentity: owner, + serviceKey: 'api:db-1:main', + pool, + databaseName: 'tenant_one', + databaseId: 'db-1', + apiId: 'api-1', + schemas: ['app', 'auth'], + anonRole: 'app_anon', + roleName: 'app_user', + introspectionRole: 'app_reader', + databaseSettings: { enableSearch: true, enableRealtime: false } as DatabaseSettings, + compute: { + modules: [{ + schemaName: 'compute', + definitionsTableName: 'definitions', + bindingsTableName: 'bindings', + invocationsSchemaName: 'private', + invocationsTableName: 'invocations', + invocationsEntityField: 'database_id' + }] + }, + explain: false, + maskError, + ...overrides +}); + +const key = (input: Record = makeInput()): string => + createGraphileBuildCacheKey('server', createGraphileServerBuildSnapshot(input as any)); + +describe('server Graphile build snapshot', () => { + it.each([ + ['service key', { serviceKey: 'api:db-2:main' }], + ['pool object', { pool: {} as Pool }], + ['database name', { databaseName: 'tenant_two' }], + ['database id', { databaseId: 'db-2' }], + ['API id', { apiId: 'api-2' }], + ['schema order', { schemas: ['auth', 'app'] }], + ['anonymous role', { anonRole: 'other_anon' }], + ['served role', { roleName: 'other_user' }], + ['introspection role', { introspectionRole: 'other_reader' }], + ['caller presets', { graphileOptions: { preset: { schema: { defaultBehavior: '-insert' } } } }], + ['Grafast cache settings', { graphileOptions: { grafastCache: { queryCacheMaxLength: 100 } } }], + ['plugin flags', { databaseSettings: { enableSearch: false, enableRealtime: false } }], + ['compute binding config', { + compute: { modules: [{ + schemaName: 'compute', + definitionsTableName: 'definitions', + bindingsTableName: 'other_bindings', + invocationsSchemaName: 'private', + invocationsTableName: 'invocations', + invocationsEntityField: 'database_id' + }] } + }], + ['explain option', { explain: true }], + ['mask-error closure', { maskError: (): string => 'different closure' }], + ['configuration owner', { ownerIdentity: {} }] + ])('includes the actual producer input for %s', (_label, override) => { + expect(key({ ...makeInput(), ...override })).not.toBe(key()); + }); + + it('keeps the snapshot and its fingerprint stable after the source inputs mutate', () => { + const input = makeInput(); + const snapshot = createGraphileServerBuildSnapshot(input as any); + const before = createGraphileBuildCacheKey('server', snapshot); + + input.schemas.push('private'); + (input.databaseSettings as DatabaseSettings).enableSearch = false; + input.compute.modules[0].bindingsTableName = 'changed'; + + expect(createGraphileBuildCacheKey('server', snapshot)).toBe(before); + expect(snapshot.schemas).toEqual(['app', 'auth']); + expect(snapshot.databaseSettings?.enableSearch).toBe(true); + expect(snapshot.computeModules[0].bindingsTableName).toBe('bindings'); + expect(Object.isFrozen(snapshot)).toBe(true); + expect(Object.isFrozen(snapshot.schemas)).toBe(true); + expect(Object.isFrozen(snapshot.computeModules[0])).toBe(true); + }); + + it('uses the same stable owner and physical pool references across snapshots', () => { + expect(key()).toBe(key()); + }); +}); diff --git a/graphql/server/src/middleware/__tests__/graphile-single-flight.test.ts b/graphql/server/src/middleware/__tests__/graphile-single-flight.test.ts deleted file mode 100644 index 5e276a9045..0000000000 --- a/graphql/server/src/middleware/__tests__/graphile-single-flight.test.ts +++ /dev/null @@ -1,122 +0,0 @@ -const mockCacheGet = jest.fn(); -const mockCacheSet = jest.fn(); -const mockCreateGraphileInstance = jest.fn(); -const mockGetPgPool = jest.fn(); -const mockMakeIntrospectionWiring = jest.fn(); -const mockCreateGrafastCacheLimitsPreset = jest.fn(); - -jest.mock('graphile-cache', () => ({ - createGraphileInstance: mockCreateGraphileInstance, - graphileCache: { - get: mockCacheGet, - set: mockCacheSet - } -})); - -jest.mock('graphile-settings', () => ({ - createConstructivePreset: jest.fn(() => ({})), - createGrafastCacheLimitsPreset: mockCreateGrafastCacheLimitsPreset -})); - -jest.mock('pg-cache', () => ({ - getPgPool: mockGetPgPool -})); - -jest.mock('../graphile-introspection', () => ({ - makeIntrospectionWiring: mockMakeIntrospectionWiring -})); - -import type { Request, Response } from 'express'; - -import { - clearInFlightMap, - getInFlightCount, - graphile -} from '../graphile'; - -const makeRequest = (): Request => - ({ - requestId: 'request-id', - svc_key: 'service-key', - api: { - dbname: 'database', - anonRole: 'anonymous', - roleName: 'authenticated', - schema: ['app_public'], - databaseId: 'database-id' - }, - get: jest.fn((): undefined => undefined) - }) as unknown as Request; - -describe('graphile single-flight handler creation', () => { - beforeEach(() => { - clearInFlightMap(); - mockCacheGet.mockReset().mockReturnValue(undefined); - mockCacheSet.mockReset(); - mockGetPgPool.mockReset().mockReturnValue({}); - mockCreateGraphileInstance.mockReset().mockResolvedValue({ - handler: jest.fn() - }); - mockMakeIntrospectionWiring.mockReset(); - mockCreateGrafastCacheLimitsPreset.mockReset().mockReturnValue({ - plugins: [{ name: 'GrafastCacheLimitsPlugin', version: '1.0.0' }] - }); - }); - - it('coalesces requests while asynchronous preset wiring is pending', async () => { - let releaseWiring!: () => void; - const wiringReady = new Promise((resolve) => { - releaseWiring = resolve; - }); - mockMakeIntrospectionWiring.mockImplementation(async () => { - await wiringReady; - return { presets: [], pgService: {} }; - }); - - const middleware = graphile({ - graphile: { - grafastCache: { queryCacheMaxLength: 16 }, - extends: [ - { - gather: { - pgScopedIntrospection: { main: true } - } - } - ], - preset: { - gather: { - pgScopedIntrospection: { main: false } - } - } - } - } as any); - const response = { headersSent: false } as unknown as Response; - const next = jest.fn(); - const first = middleware(makeRequest(), response, next); - - // Let the first request enter the registered creation promise before the - // second request checks the in-flight map. - await Promise.resolve(); - expect(getInFlightCount()).toBe(1); - - const second = middleware(makeRequest(), response, next); - expect(getInFlightCount()).toBe(1); - expect(mockMakeIntrospectionWiring).toHaveBeenCalledTimes(1); - - releaseWiring(); - await Promise.all([first, second]); - - expect(mockCreateGraphileInstance).toHaveBeenCalledTimes(1); - expect(mockCreateGrafastCacheLimitsPreset).toHaveBeenCalledWith({ - queryCacheMaxLength: 16 - }); - expect(mockCreateGraphileInstance.mock.calls[0][0].preset.extends).toContainEqual({ - plugins: [{ name: 'GrafastCacheLimitsPlugin', version: '1.0.0' }] - }); - expect(mockCacheSet).toHaveBeenCalledTimes(1); - expect(getInFlightCount()).toBe(0); - expect(mockCreateGraphileInstance.mock.calls[0][0].preset.gather).toEqual({ - pgScopedIntrospection: { main: false } - }); - }); -}); diff --git a/graphql/server/src/middleware/flush.ts b/graphql/server/src/middleware/flush.ts index 635ad2657e..e9ed69c060 100644 --- a/graphql/server/src/middleware/flush.ts +++ b/graphql/server/src/middleware/flush.ts @@ -1,11 +1,15 @@ import './types'; // for Request type +import { errors } from '@constructive-io/errors'; import { ConstructiveOptions } from '@constructive-io/graphql-types'; import { Logger } from '@pgpmjs/logger'; import { svcCache } from '@pgpmjs/server-utils'; import { createHash, timingSafeEqual } from 'crypto'; import { NextFunction, Request, RequestHandler, Response } from 'express'; -import { graphileCache } from 'graphile-cache'; +import { + clearGraphileEntriesForDatabase, + clearGraphileEntriesForService +} from 'graphile-cache'; import { getPgPool } from 'pg-cache'; import { getRoutingSchema, isValidSchemaName } from './routing'; @@ -61,8 +65,13 @@ export const createFlushMiddleware = ( return; } - graphileCache.delete((req as any).svc_key); - svcCache.delete((req as any).svc_key); + const serviceKey = req.constructive?.serviceKey; + if (!serviceKey) { + next(errors.INTERNAL_FAILURE({ details: 'Missing request context' })); + return; + } + clearGraphileEntriesForService(serviceKey); + svcCache.delete(serviceKey); res.status(200).send('OK'); }; }; @@ -77,14 +86,12 @@ export const flushService = async ( const api = new RegExp(`^api:${databaseId}:.*`); const schemata = new RegExp(`^schemata:${databaseId}:.*`); const meta = new RegExp(`^metaschema:api:${databaseId}`); + clearGraphileEntriesForDatabase(databaseId); if (!opts.api.isPublic) { - graphileCache.forEach((_, k: string) => { - if (api.test(k) || schemata.test(k) || meta.test(k)) { - graphileCache.delete(k); - svcCache.delete(k); - } - }); + for (const key of svcCache.keys()) { + if (api.test(key) || schemata.test(key) || meta.test(key)) svcCache.delete(key); + } } const routingSchema = getRoutingSchema(opts); @@ -104,7 +111,7 @@ export const flushService = async ( for (const row of svc.rows) { const key: string | undefined = row.hostname || undefined; if (key) { - graphileCache.delete(key); + clearGraphileEntriesForService(key); svcCache.delete(key); } } diff --git a/graphql/server/src/middleware/graphile-build-snapshot.ts b/graphql/server/src/middleware/graphile-build-snapshot.ts new file mode 100644 index 0000000000..599c20b90f --- /dev/null +++ b/graphql/server/src/middleware/graphile-build-snapshot.ts @@ -0,0 +1,87 @@ +import type { ConstructiveOptions } from '@constructive-io/graphql-types'; +import type { DatabaseSettings, ComputeConfig, ComputeModuleConfig } from '@constructive-io/express-context'; +import type { Pool } from 'pg'; + +import { referenceGraphileBuildValue, snapshotGraphileBuildValue } from 'graphile-cache'; +import type { GraphileBuildReference } from 'graphile-cache'; + +export interface GraphileServerBuildSnapshot { + ownerIdentity: GraphileBuildReference; + serviceKey: string; + poolIdentity: GraphileBuildReference; + poolKey: string; + databaseName: string; + databaseId: string | null; + apiId?: string; + schemas: string[]; + anonRole: string; + roleName: string; + introspectionRole?: string; + databaseSettings?: DatabaseSettings; + graphileOptions?: ConstructiveOptions['graphile']; + computeModules: ComputeModuleConfig[]; + explain: boolean; + grafserv: { + graphqlPath: string; + graphiqlPath: string; + graphiql: boolean; + graphiqlOnGraphQLGET: boolean; + maskError: (...args: any[]) => any; + }; +} + +interface GraphileServerBuildSnapshotInput { + ownerIdentity: object; + serviceKey: string; + pool: Pool; + databaseName: string; + databaseId?: string | null; + apiId?: string; + schemas: string[]; + anonRole: string; + roleName: string; + introspectionRole?: string; + databaseSettings?: DatabaseSettings; + graphileOptions?: ConstructiveOptions['graphile']; + compute?: ComputeConfig; + explain: boolean; + maskError: (...args: any[]) => any; +} + +export const createGraphileServerBuildSnapshot = ( + input: GraphileServerBuildSnapshotInput +): Readonly => { + const computeModules = (input.compute?.modules ?? []).map((module) => ({ + schemaName: module.schemaName, + definitionsTableName: module.definitionsTableName, + bindingsTableName: module.bindingsTableName, + invocationsSchemaName: module.invocationsSchemaName, + invocationsTableName: module.invocationsTableName, + invocationsEntityField: module.invocationsEntityField + })); + + return snapshotGraphileBuildValue({ + ownerIdentity: referenceGraphileBuildValue(input.ownerIdentity), + serviceKey: input.serviceKey, + poolIdentity: referenceGraphileBuildValue(input.pool), + poolKey: input.databaseName, + databaseName: input.databaseName, + databaseId: input.databaseId ?? null, + apiId: input.apiId, + schemas: input.schemas, + anonRole: input.anonRole, + roleName: input.roleName, + introspectionRole: input.introspectionRole, + databaseSettings: input.databaseSettings, + graphileOptions: input.graphileOptions, + computeModules, + explain: input.explain, + grafserv: { + graphqlPath: '/graphql', + graphiqlPath: '/graphiql', + graphiql: true, + graphiqlOnGraphQLGET: false, + maskError: input.maskError + } + }); +}; diff --git a/graphql/server/src/middleware/graphile.ts b/graphql/server/src/middleware/graphile.ts index e888d0acff..b3d6cce4dc 100644 --- a/graphql/server/src/middleware/graphile.ts +++ b/graphql/server/src/middleware/graphile.ts @@ -1,34 +1,30 @@ import './types'; // for Request type -import { errors } from '@constructive-io/errors'; -import type { ComputeConfig } from '@constructive-io/express-context'; -import { DEFAULT_REQUEST_PROTECTION, protectionPgSettings } from '@constructive-io/express-context'; +import { errors, normalizeError } from '@constructive-io/errors'; import type { ConstructiveOptions } from '@constructive-io/graphql-types'; import { getNodeEnv } from '@pgpmjs/env'; import { Logger } from '@pgpmjs/logger'; import type { NextFunction, Request, RequestHandler, Response } from 'express'; -import { createGraphileInstance, graphileCache,type GraphileCacheEntry } from 'graphile-cache'; +import { + createGraphileBuildCacheKey, + createGraphileInstance, + graphileCache, + snapshotGraphileBuildValue, + type GraphileCacheEntry +} from 'graphile-cache'; import type { GraphileConfig } from 'graphile-config'; import { createFunctionBindingsPlugin } from 'graphile-function-bindings'; -import { - createConstructivePreset, - createGrafastCacheLimitsPreset -} from 'graphile-settings'; -import { getPgPool } from 'pg-cache'; -import { getPgEnvOptions } from 'pg-env'; +import { createConstructivePreset, createGrafastCacheLimitsPreset } from 'graphile-settings'; import { isGraphqlObservabilityEnabled } from '../diagnostics/observability'; -import { HandlerCreationError } from '../errors/api-errors'; import { respondWithGraphQLError } from '../errors/graphql-response'; import { AuthCookiePlugin } from '../plugins/auth-cookie-plugin'; import { createErrorEventsPlugin } from '../plugins/error-events-plugin'; import { RequestProtectionPlugin } from '../plugins/request-protection-plugin'; -import type { DatabaseSettings } from '../types'; import { makeIntrospectionWiring } from './graphile-introspection'; import { maskError } from './mask-error'; import { observeGraphileBuild } from './observability/graphile-build-stats'; - -const isDev = (): boolean => getNodeEnv() === 'development'; +import { createGraphileServerBuildSnapshot } from './graphile-build-snapshot'; // ============================================================================= // Single-Flight Pattern: In-Flight Tracking @@ -65,7 +61,6 @@ export function clearInFlightMap(): void { } const log = new Logger('graphile'); -const reqLabel = (req: Request): string => (req.requestId ? `[${req.requestId}]` : '[req]'); /** * Build a PostGraphile v5 preset for a tenant. @@ -76,29 +71,20 @@ const reqLabel = (req: Request): string => (req.requestId ? `[${req.requestId}]` * (everything on except aggregates). */ const buildPreset = async ( - pool: import('pg').Pool, - schemas: string[], - anonRole: string, - roleName: string, - introspectionRole: string | undefined, - graphileOptions: ConstructiveOptions['graphile'], - databaseSettings?: DatabaseSettings, - apiId?: string, - compute?: ComputeConfig + snapshot: ReturnType, + pool: import('pg').Pool ): Promise => { - const introspection = await makeIntrospectionWiring(pool, schemas, graphileOptions, undefined, introspectionRole); - const configuredPreset = graphileOptions?.preset ?? {}; - const grafastCachePreset = createGrafastCacheLimitsPreset(graphileOptions?.grafastCache); + const introspection = await makeIntrospectionWiring(pool, snapshot.schemas, snapshot.graphileOptions, undefined, snapshot.introspectionRole); + const configuredPreset = snapshot.graphileOptions?.preset ?? {}; + const grafastCachePreset = createGrafastCacheLimitsPreset(snapshot.graphileOptions?.grafastCache); return { ...configuredPreset, extends: [ - createConstructivePreset(databaseSettings), - ...(graphileOptions?.extends ?? []), + createConstructivePreset(snapshot.databaseSettings), + ...(snapshot.graphileOptions?.extends ?? []), ...(configuredPreset.extends ?? []), ...introspection.presets, - ...(Object.keys(grafastCachePreset).length > 0 - ? [grafastCachePreset] - : []) + ...(Object.keys(grafastCachePreset).length > 0 ? [grafastCachePreset] : []) ], plugins: [ ...(configuredPreset.plugins ?? []), @@ -109,11 +95,11 @@ const buildPreset = async ( // database — all schema/table names come from the constructive // metaschema (express-context compute module loader); the plugin has // no fallbacks or discovery of its own. - ...(apiId && compute?.modules.length + ...(snapshot.apiId && snapshot.computeModules.length ? [ createFunctionBindingsPlugin({ - apiId, - modules: compute.modules.map((m) => ({ + apiId: snapshot.apiId, + modules: snapshot.computeModules.map((m) => ({ computeSchema: m.schemaName, bindingsTable: m.bindingsTableName, definitionsTable: m.definitionsTableName, @@ -126,154 +112,14 @@ const buildPreset = async ( : []) ], pgServices: [introspection.pgService], - grafserv: { - graphqlPath: '/graphql', - graphiqlPath: '/graphiql', - graphiql: true, - graphiqlOnGraphQLGET: false, - maskError - }, + grafserv: snapshot.grafserv, grafast: { - explain: process.env.NODE_ENV === 'development', + explain: snapshot.explain, context: (requestContext: Partial) => { - // In grafserv/express/v4, the request is available at requestContext.expressv4.req const req = (requestContext as { expressv4?: { req?: Request } })?.expressv4?.req; - const context: Record = {}; - - // Timeouts travel with the transaction as GUCs, so they bound the work - // this request can do inside PostgreSQL whatever the plan turns out to - // be. Resolved per request (not baked into the cached preset) so a - // tenant lowering a timeout takes effect on the next request. - const timeouts = protectionPgSettings(req?.requestProtection ?? DEFAULT_REQUEST_PROTECTION); - - if (req) { - if (req.databaseId) { - context['jwt.claims.database_id'] = req.databaseId; - } - // API provenance — which API surface this request arrived through. - // Derived server-side by resolving the hostname through the scoped - // routing plane (resolve_route -> api_id); never taken from - // client-supplied headers, body, or token payload. - if (req.api?.apiId) { - context['jwt.claims.api_id'] = req.api.apiId; - } - if (req.clientIp) { - context['jwt.claims.ip_address'] = req.clientIp; - } - if (req.get('origin')) { - context['jwt.claims.origin'] = req.get('origin') as string; - } - if (req.get('User-Agent')) { - context['jwt.claims.user_agent'] = req.get('User-Agent') as string; - } - if (req.deviceToken) { - context['jwt.claims.device_token'] = req.deviceToken; - } - - if (req.token?.user_id) { - const pgSettings: Record = { - ...timeouts, - role: roleName, - 'jwt.claims.token_id': req.token.id, - 'jwt.claims.user_id': req.token.user_id, - ...context - }; - - if (req.token.session_id) { - pgSettings['jwt.claims.session_id'] = req.token.session_id; - } - if (req.token.root_session_id) { - pgSettings['jwt.claims.root_session_id'] = req.token.root_session_id; - } - if (req.token.parent_session_id) { - pgSettings['jwt.claims.parent_session_id'] = req.token.parent_session_id; - } - if (req.token.intent) { - pgSettings['jwt.claims.intent'] = req.token.intent; - } - - // Propagate credential metadata as JWT claims so PG functions - // can read them via current_setting('jwt.claims.access_level') etc. - if (req.token.access_level) { - pgSettings['jwt.claims.access_level'] = req.token.access_level; - } - if (req.token.kind) { - pgSettings['jwt.claims.kind'] = req.token.kind; - } - - // Principal identity — always set; equals user_id for human sessions - pgSettings['jwt.claims.principal_id'] = req.token.principal_id || req.token.user_id; - - // Enforce read-only transactions for read_only credentials - if (req.token.access_level === 'read_only') { - pgSettings['default_transaction_read_only'] = 'on'; - } - - if (req.requestId) { - pgSettings['request.id'] = req.requestId; - } - - return { pgSettings }; - } - - // Private (in-cluster) surface: there is no token — identity - // arrives on the trusted internal X-* headers stamped by the - // dispatching worker/sync gateway (the same vocabulary as - // X-Database-Id above). Map it into per-request claims so writes - // made through this surface carry actor attribution. Never applied - // on the public surface, where client-supplied identity headers - // must not assert identity. - const headerActorId = req.get('X-Actor-Id'); - if (req.api?.isPublic === false && headerActorId) { - const pgSettings: Record = { - ...timeouts, - role: roleName, - 'jwt.claims.user_id': headerActorId, - 'jwt.claims.principal_id': headerActorId, - ...context - }; - // The entity pair travels together: the tenant's writers reject an - // entity id whose type they cannot interpret (`ENTITY_TYPE_REQUIRED`). - const headerEntityId = req.get('X-Entity-Id'); - const headerEntityType = req.get('X-Entity-Type'); - if (headerEntityId) { - pgSettings['jwt.claims.entity_id'] = headerEntityId; - } - if (headerEntityType) { - pgSettings['jwt.claims.entity_type'] = headerEntityType; - } - const headerOrganizationId = req.get('X-Organization-Id'); - if (headerOrganizationId) { - pgSettings['jwt.claims.organization_id'] = headerOrganizationId; - } - if (req.requestId) { - pgSettings['request.id'] = req.requestId; - } - return { pgSettings }; - } - } - - // No actor to name, so the tenant database the request addresses carries - // the attribution — the same rule the sync gateway applies to a request - // that arrives without a credential. Without it the tenant's own writers - // refuse the work an anonymous request legitimately does - // (`ATTRIBUTION_REQUIRED`), so a public mutation cannot enqueue a job. - const anonSettings: Record = { - ...timeouts, - role: anonRole, - ...context - }; - if (req?.databaseId) { - anonSettings['jwt.claims.entity_id'] = req.databaseId; - anonSettings['jwt.claims.entity_type'] = 'database'; - } - if (req?.requestId) { - anonSettings['request.id'] = req.requestId; - } - - return { - pgSettings: anonSettings - }; + const context = req?.constructive; + if (!context) throw errors.INTERNAL_FAILURE({ details: 'Missing request context' }); + return { pgSettings: context.pgSettings }; } } }; @@ -281,147 +127,110 @@ const buildPreset = async ( export const graphile = (opts: ConstructiveOptions): RequestHandler => { const observabilityEnabled = isGraphqlObservabilityEnabled(opts.server?.host); + const ownerIdentity = {}; return async (req: Request, res: Response, next: NextFunction) => { - const label = reqLabel(req); try { - const api = req.api; - if (!api) { - log.error(`${label} Missing API info`); - respondWithGraphQLError(res, errors.INTERNAL_FAILURE({ details: 'Missing API info' })); - return; + const context = req.constructive; + if (!context) { + throw errors.INTERNAL_FAILURE({ details: 'Missing request context' }); } - const key = req.svc_key; - if (!key) { - log.error(`${label} Missing service cache key`); - respondWithGraphQLError( - res, - errors.INTERNAL_FAILURE({ details: 'Missing service cache key' }) - ); - return; + const api = context.api; + const serviceKey = context.serviceKey; + if (!serviceKey) { + throw errors.INTERNAL_FAILURE({ details: 'Missing service cache key' }); } const { dbname, anonRole, roleName, schema } = api; - const schemaLabel = schema?.join(',') || 'unknown'; + const schemas = [...(schema ?? [])]; + const pool = context.pool; + const databaseId = context.databaseId; + const apiId = api.apiId; + const databaseSettings = api.databaseSettings + ? snapshotGraphileBuildValue(api.databaseSettings) + : undefined; + const introspectionRole = opts.api?.introspectionRole; + const explain = getNodeEnv() === 'development'; + const snapshotInput = { + ownerIdentity, + serviceKey, + pool, + databaseName: dbname, + databaseId, + apiId, + schemas, + anonRole, + roleName, + introspectionRole, + databaseSettings, + graphileOptions: snapshotGraphileBuildValue(opts.graphile), + explain, + maskError + }; + // Resolve the request-owned compute snapshot before cache lookup because + // its provisioned bindings change the exact Graphile schema. + const compute = apiId ? await context.useModule('compute') : undefined; + const snapshot = createGraphileServerBuildSnapshot({ ...snapshotInput, compute }); + const key = createGraphileBuildCacheKey('server', snapshot); - // ========================================================================= - // Phase A: Cache Check (fast path) - // ========================================================================= const cached = graphileCache.get(key); if (cached) { - log.debug(`${label} PostGraphile cache hit key=${key} db=${dbname} schemas=${schemaLabel}`); + log.debug('Graphile cache hit', { requestId: context.requestId }); return cached.handler(req, res, next); } + log.debug('Graphile cache miss', { requestId: context.requestId }); - log.debug(`${label} PostGraphile cache miss key=${key} db=${dbname} schemas=${schemaLabel}`); - - // ========================================================================= - // Phase B: In-Flight Check (single-flight coalescing) - // ========================================================================= const inFlight = creating.get(key); if (inFlight) { - log.debug(`${label} Coalescing request for PostGraphile[${key}] - waiting for in-flight creation`); - try { - const instance = await inFlight; - return instance.handler(req, res, next); - } catch (error) { - log.warn(`${label} Coalesced request failed for PostGraphile[${key}], retrying`); - // Fall through to Phase C to retry creation - } + const instance = await inFlight; + return instance.handler(req, res, next); } - - // ========================================================================= - // Phase C: Create New Handler (first request for this key) - // ========================================================================= - - // Re-check cache after coalesced request failure (another retry may have succeeded) const recheckedCache = graphileCache.get(key); - if (recheckedCache) { - log.debug(`${label} PostGraphile cache hit on re-check key=${key}`); - return recheckedCache.handler(req, res, next); - } - - // Re-check in-flight map (another retry may have started creation) + if (recheckedCache) return recheckedCache.handler(req, res, next); const retryInFlight = creating.get(key); if (retryInFlight) { - log.debug(`${label} Re-coalescing request for PostGraphile[${key}]`); const retryInstance = await retryInFlight; return retryInstance.handler(req, res, next); } - log.info( - `${label} Building PostGraphile v5 handler key=${key} db=${dbname} schemas=${schemaLabel} role=${roleName} anon=${anonRole}` - ); - - const pgConfig = getPgEnvOptions({ - ...opts.pg, - database: dbname - }); - - // Route through pg-cache so the pool is tracked and can be cleaned up - // properly, preventing leaked connections during database teardown. - const pool = getPgPool(pgConfig); - - // Register the creation promise before any asynchronous preset work so - // concurrent requests coalesce while optional plugin modules load. - const creationPromise = (async () => { - const compute = api.apiId ? await req.constructive?.useModule('compute') : undefined; - const preset = await buildPreset( - pool, - schema || [], - anonRole, - roleName, - opts.api?.introspectionRole, - opts.graphile, - api.databaseSettings, - api.apiId, - compute - ); + const creationPromise = Promise.resolve().then(async () => { + const preset = await buildPreset(snapshot, pool); return observeGraphileBuild( - { - cacheKey: key, - serviceKey: key, - databaseId: api.databaseId ?? null - }, + { cacheKey: key, serviceKey, databaseId }, () => createGraphileInstance({ preset, cacheKey: key, - enableRealtime: api.databaseSettings?.enableRealtime + enableRealtime: snapshot.databaseSettings?.enableRealtime }), { enabled: observabilityEnabled } ); - })(); + }); creating.set(key, creationPromise); try { const instance = await creationPromise; + Object.assign(instance, { + serviceKey, + databaseId, + poolKey: snapshot.poolKey + } satisfies Partial); graphileCache.set(key, instance); - log.info(`${label} Cached PostGraphile v5 handler key=${key} db=${dbname}`); + log.debug('Graphile handler ready', { requestId: context.requestId }); return instance.handler(req, res, next); - } catch (error) { - log.error(`${label} Failed to create PostGraphile[${key}]:`, error); - throw new HandlerCreationError( - `Failed to create handler for ${key}: ${error instanceof Error ? error.message : String(error)}`, - { - cacheKey: key, - cause: error instanceof Error ? error.message : String(error) - } - ); } finally { // Always clean up in-flight tracker creating.delete(key); } - } catch (e: any) { - log.error(`${label} PostGraphile middleware error`, e); + } catch (error) { + const failure = normalizeError(error); + log.error('Graphile request refused', { + requestId: req.constructive?.requestId, + code: failure.code + }); if (!res.headersSent) { - respondWithGraphQLError( - res, - errors.INTERNAL_FAILURE({ - details: isDev() ? e?.message ?? String(e) : 'An unexpected error occurred' - }) - ); + respondWithGraphQLError(res, failure, { status: failure.http }); return; } - next(e); + next(failure); } }; }; diff --git a/graphql/server/src/middleware/observability/graphile-build-stats.ts b/graphql/server/src/middleware/observability/graphile-build-stats.ts index f8825ac011..0b6e0ddd7c 100644 --- a/graphql/server/src/middleware/observability/graphile-build-stats.ts +++ b/graphql/server/src/middleware/observability/graphile-build-stats.ts @@ -1,3 +1,5 @@ +import { normalizeError } from '@constructive-io/errors'; + export interface GraphileBuildEvent { type: 'start' | 'success' | 'failure'; cacheKey: string; @@ -137,7 +139,7 @@ const recordBuildFailure = ( buildStats.maxMs = Math.max(buildStats.maxMs, durationMs); buildStats.lastMs = durationMs; buildStats.lastFinishedAt = new Date().toISOString(); - buildStats.lastError = error instanceof Error ? error.message : String(error); + buildStats.lastError = normalizeError(error).code; buildStats.lastServiceKey = context.serviceKey; buildStats.lastDatabaseId = context.databaseId; @@ -147,7 +149,7 @@ const recordBuildFailure = ( serviceKey: context.serviceKey, databaseId: context.databaseId, durationMs, - error: error instanceof Error ? error.message : String(error), + error: normalizeError(error).code, timestamp: new Date().toISOString(), }); }; diff --git a/graphql/server/src/server.ts b/graphql/server/src/server.ts index af82eeb42c..07f9122712 100644 --- a/graphql/server/src/server.ts +++ b/graphql/server/src/server.ts @@ -225,8 +225,8 @@ class Server { // REST function invocation routes (POST /fn/:alias, GET /fn/invocations/:id) app.use(createFnRouter()); - app.use(graphile(effectiveOpts)); app.use(createFlushMiddleware(effectiveOpts)); + app.use(graphile(effectiveOpts)); // Error handling - MUST be LAST app.use(notFoundHandler); // Catches unmatched routes (404) diff --git a/packages/errors/__tests__/cause.test.ts b/packages/errors/__tests__/cause.test.ts new file mode 100644 index 0000000000..d22e5dd3d8 --- /dev/null +++ b/packages/errors/__tests__/cause.test.ts @@ -0,0 +1,38 @@ +import { ConstructiveError, errors, normalizeError } from '../src'; + +it('retains the original internal cause without serializing it into extensions', () => { + const cause = new Error('private database credential'); + const error = new ConstructiveError({ + code: 'INTERNAL_FAILURE', message: 'Internal failure', errorClass: 'internal', http: 500, cause + }); + expect(error.cause).toBe(cause); + expect(JSON.stringify(error.toExtensions())).not.toContain(cause.message); + expect(JSON.stringify(error)).not.toContain(cause.message); +}); + +it('maps unknown errors to a safe internal result with the original cause', () => { + const original = new Error('password=private tenant=secret'); + const normalized = normalizeError(original); + expect(normalized.code).toBe('INTERNAL_FAILURE'); + expect(normalized.http).toBe(500); + expect(normalized.cause).toBe(original); + expect(normalized.message).not.toContain('private'); + expect(JSON.stringify(normalized.toExtensions())).not.toContain('secret'); +}); + +it('removes overridden messages and context from canonical internal failures', () => { + const original = errors.INTERNAL_FAILURE({ details: 'password=private' }, 'secret stack'); + const normalized = normalizeError(original); + expect(normalized.cause).toBe(original); + expect(normalized.message).not.toContain('secret'); + expect(normalized.toExtensions()).not.toHaveProperty('context'); +}); + +it('preserves a registered refusal status and code without exposing a message override', () => { + const original = errors.FORBIDDEN({}, 'private implementation detail'); + const normalized = normalizeError(original); + expect(normalized.code).toBe('FORBIDDEN'); + expect(normalized.http).toBe(403); + expect(normalized.cause).toBe(original); + expect(normalized.message).not.toContain('private'); +}); diff --git a/packages/errors/src/error.ts b/packages/errors/src/error.ts index 98f9269f15..68860d4c27 100644 --- a/packages/errors/src/error.ts +++ b/packages/errors/src/error.ts @@ -6,6 +6,8 @@ export interface ConstructiveErrorArgs { errorClass: ErrorClass; http: number; context?: ErrorContext; + /** Internal cause; never included in GraphQL extensions. */ + cause?: unknown; } /** @@ -22,7 +24,7 @@ export class ConstructiveError extends Error { readonly context?: ErrorContext; constructor(args: ConstructiveErrorArgs) { - super(args.message); + super(args.message, { cause: args.cause }); this.name = 'ConstructiveError'; this.code = args.code; this.errorClass = args.errorClass; diff --git a/packages/errors/src/http.ts b/packages/errors/src/http.ts index d61895c9e0..d08ac866ab 100644 --- a/packages/errors/src/http.ts +++ b/packages/errors/src/http.ts @@ -1,3 +1,6 @@ +import { ConstructiveError } from './error'; +import { errors } from './factory'; +import { format } from './format'; import { getDefinition } from './registry'; /** Status used when a code carries no mapping. */ @@ -61,3 +64,20 @@ export function httpStatusFor(code: string | null | undefined): HttpStatusResolu return { status: UNMAPPED_HTTP_STATUS, mapped: false }; } + +/** Preserve the internal cause while exposing only registry-owned diagnostics. */ +export function normalizeError(error: unknown): ConstructiveError { + const definition = error instanceof ConstructiveError ? getDefinition(error.code) : undefined; + const status = error instanceof ConstructiveError ? httpStatusFor(error.code).status : 500; + const fallback = errors.INTERNAL_FAILURE({ details: 'An unexpected error occurred' }); + return new ConstructiveError({ + code: definition?.code ?? fallback.code, + message: definition && definition.code !== fallback.code + ? format(definition.code, error instanceof ConstructiveError && definition.class === 'public' ? error.context : {}) + : fallback.message, + errorClass: definition?.class ?? fallback.errorClass, + http: status, + context: definition?.class === 'public' && error instanceof ConstructiveError ? error.context : undefined, + cause: error + }); +} diff --git a/packages/express-context/__tests__/pg-settings.test.ts b/packages/express-context/__tests__/pg-settings.test.ts index 14b9cafb4c..28cedc45dc 100644 --- a/packages/express-context/__tests__/pg-settings.test.ts +++ b/packages/express-context/__tests__/pg-settings.test.ts @@ -1,4 +1,5 @@ import { buildPgSettings } from '../src/pg-settings'; +import { DEFAULT_REQUEST_PROTECTION } from '../src/request-protection'; import type { ApiStructure, ConstructiveAPIToken } from '../src/types'; const api: ApiStructure = { @@ -84,3 +85,140 @@ describe('buildPgSettings — agent-auth claims (intent, session lineage)', () = expect(settings).not.toHaveProperty('jwt.claims.intent'); }); }); + +describe('buildPgSettings — Graphile request context parity', () => { + it('preserves authenticated claims and request metadata', () => { + const token: ConstructiveAPIToken = { + id: 'token-1', + user_id: 'user-1', + principal_id: 'principal-1', + session_id: 'session-1', + root_session_id: 'root-session-1', + parent_session_id: 'parent-session-1', + intent: 'agent:run', + access_level: 'read_only', + kind: 'api_token' + }; + + const settings = buildPgSettings({ + api, + token, + requestId: 'request-1', + clientIp: '192.0.2.1', + origin: 'https://app.example.test', + userAgent: 'test-agent', + deviceToken: 'device-1' + }); + + expect(settings).toMatchObject({ + role: 'authenticated', + 'jwt.claims.token_id': 'token-1', + 'jwt.claims.user_id': 'user-1', + 'jwt.claims.principal_id': 'principal-1', + 'jwt.claims.session_id': 'session-1', + 'jwt.claims.root_session_id': 'root-session-1', + 'jwt.claims.parent_session_id': 'parent-session-1', + 'jwt.claims.intent': 'agent:run', + 'jwt.claims.access_level': 'read_only', + 'jwt.claims.kind': 'api_token', + 'jwt.claims.api_id': api.apiId, + 'jwt.claims.database_id': api.databaseId, + 'jwt.claims.ip_address': '192.0.2.1', + 'jwt.claims.origin': 'https://app.example.test', + 'jwt.claims.user_agent': 'test-agent', + 'jwt.claims.device_token': 'device-1', + 'request.id': 'request-1', + default_transaction_read_only: 'on' + }); + expect(buildPgSettings({ + api, + token: { ...token, principal_id: undefined }, + requestId: 'request-1' + })['jwt.claims.principal_id']).toBe('user-1'); + }); + + it('uses the database as anonymous attribution', () => { + const settings = buildPgSettings({ api, token: null, requestId: 'request-2' }); + + expect(settings).toMatchObject({ + role: 'anonymous', + 'jwt.claims.entity_id': api.databaseId, + 'jwt.claims.entity_type': 'database' + }); + expect(settings['jwt.claims.user_id']).toBeUndefined(); + }); + + it('honors actor metadata on private APIs and ignores the same spoof on public APIs', () => { + const headers = { + actorId: 'actor-1', + entityId: 'entity-1', + entityType: 'organization', + organizationId: 'organization-1' + }; + const privateSettings = buildPgSettings({ + api: { ...api, isPublic: false }, + token: null, + requestId: 'request-3', + headers + }); + const publicSettings = buildPgSettings({ + api, + token: null, + requestId: 'request-4', + headers + }); + + expect(privateSettings).toMatchObject({ + role: 'authenticated', + 'jwt.claims.user_id': 'actor-1', + 'jwt.claims.principal_id': 'actor-1', + 'jwt.claims.entity_id': 'entity-1', + 'jwt.claims.entity_type': 'organization', + 'jwt.claims.organization_id': 'organization-1' + }); + const actorOnlySettings = buildPgSettings({ + api: { ...api, isPublic: false }, + token: null, + requestId: 'request-3b', + headers: { actorId: 'actor-1' } + }); + expect(actorOnlySettings['jwt.claims.entity_id']).toBeUndefined(); + expect(actorOnlySettings['jwt.claims.entity_type']).toBeUndefined(); + expect(publicSettings).toMatchObject({ + role: 'anonymous', + 'jwt.claims.entity_id': api.databaseId, + 'jwt.claims.entity_type': 'database' + }); + expect(publicSettings['jwt.claims.user_id']).toBeUndefined(); + expect(publicSettings['jwt.claims.organization_id']).toBeUndefined(); + }); + + it('applies bounded request timeouts and read-only transactions', () => { + const requestProtection = { + ...DEFAULT_REQUEST_PROTECTION, + statementTimeoutMs: 1200, + idleInTransactionTimeoutMs: 2300, + lockTimeoutMs: 340, + }; + const readOnly = buildPgSettings({ + api, + token: { user_id: 'user-1', access_level: 'read_only' }, + requestId: 'request-5', + requestProtection + }); + const writable = buildPgSettings({ + api, + token: { user_id: 'user-1' }, + requestId: 'request-6', + requestProtection + }); + + expect(readOnly).toMatchObject({ + statement_timeout: '1200', + idle_in_transaction_session_timeout: '2300', + lock_timeout: '340', + default_transaction_read_only: 'on' + }); + expect(writable.default_transaction_read_only).toBeUndefined(); + }); +}); diff --git a/packages/express-context/src/context.ts b/packages/express-context/src/context.ts index 82d87de9d3..6bb64d554a 100644 --- a/packages/express-context/src/context.ts +++ b/packages/express-context/src/context.ts @@ -25,8 +25,15 @@ import type { LoaderRegistry } from './loaders/registry'; import type { LoaderContext } from './loaders/types'; import { withPgClient as withPgClientFn } from './pg-client'; import { buildPgSettings } from './pg-settings'; +import type { RequestProtection } from './request-protection'; import type { BillingConfig, BuiltinModuleMap, ConstructiveContext, InferenceLogConfig, LlmConfig } from './types'; +type ContextRequest = Request & { + deviceToken?: string; + requestProtection?: RequestProtection; + svc_key?: string; +}; + export interface ContextMiddlewareOptions { /** Base PG options for pool creation (host, port, user, password) */ pg?: PgpmOptions['pg']; @@ -70,14 +77,25 @@ export function buildContext( const api = req.api; if (!api) return null; + const contextReq = req as ContextRequest; const token = req.token ?? null; const requestId = req.requestId || ''; - const pgSettings = buildPgSettings({ + const buildCurrentPgSettings = () => buildPgSettings({ api, token, requestId, - clientIp: req.clientIp + clientIp: contextReq.clientIp, + origin: contextReq.get('origin'), + userAgent: contextReq.get('User-Agent'), + deviceToken: contextReq.deviceToken, + requestProtection: contextReq.requestProtection, + headers: { + actorId: contextReq.get('X-Actor-Id'), + entityId: contextReq.get('X-Entity-Id'), + entityType: contextReq.get('X-Entity-Type'), + organizationId: contextReq.get('X-Organization-Id') + } }); const tenantPool: Pool = getPgPool({ @@ -100,7 +118,7 @@ export function buildContext( } const withPgClient = (fn: (client: any) => Promise) => - withPgClientFn(tenantPool, pgSettings, fn); + withPgClientFn(tenantPool, buildCurrentPgSettings(), fn); const useModule = createUseModule(opts.loaders, loaderCtx); // Lazy-initialized billing client (cached per request) @@ -111,7 +129,12 @@ export function buildContext( return { api, token, - pgSettings, + serviceKey: contextReq.svc_key, + get pgSettings() { + // Request protection is resolved after this middleware, so compute the + // settings at read time and keep withPgClient on the same owner path. + return buildCurrentPgSettings(); + }, databaseId: api.databaseId ?? null, userId: token?.user_id ?? null, requestId, diff --git a/packages/express-context/src/pg-settings.ts b/packages/express-context/src/pg-settings.ts index 92dfa89e77..a3fb642ebe 100644 --- a/packages/express-context/src/pg-settings.ts +++ b/packages/express-context/src/pg-settings.ts @@ -10,8 +10,21 @@ * across the PostGraphile server, LLM sidecar, or any Express service. */ +import { + DEFAULT_REQUEST_PROTECTION, + protectionPgSettings, + type RequestProtection +} from './request-protection'; import type { ApiStructure, ConstructiveAPIToken } from './types'; +/** Trusted identity headers available only on private, server-to-server APIs. */ +export interface PgSettingsHeaders { + actorId?: string; + entityId?: string; + entityType?: string; + organizationId?: string; +} + export interface PgSettingsInput { /** Resolved API config (provides role names, database_id) */ api: ApiStructure; @@ -21,6 +34,16 @@ export interface PgSettingsInput { requestId: string; /** Client IP address (from request-ip middleware) */ clientIp?: string; + /** Origin header captured by the server. */ + origin?: string; + /** User-Agent header captured by the server. */ + userAgent?: string; + /** Trusted device cookie resolved by authentication middleware. */ + deviceToken?: string; + /** Per-request protection bounds, resolved and clamped by the platform. */ + requestProtection?: RequestProtection; + /** Identity headers captured from the request; trusted only for private APIs. */ + headers?: PgSettingsHeaders; } /** @@ -30,15 +53,55 @@ export interface PgSettingsInput { * making them available to RLS policies and SQL functions. */ export function buildPgSettings(input: PgSettingsInput): Record { - const { api, token, requestId, clientIp } = input; - const settings: Record = {}; + const { + api, + token, + requestId, + clientIp, + origin, + userAgent, + deviceToken, + requestProtection, + headers + } = input; + const settings: Record = { + ...protectionPgSettings(requestProtection ?? DEFAULT_REQUEST_PROTECTION) + }; - // Role: from token (authenticated) or api (anonymous fallback) + // Role: from token (authenticated), trusted private actor, or the API's + // anonymous fallback. The private branch must be selected before the + // anonymous database attribution is added. + // Internal actor attribution is accepted only on a server-to-server API. + // Public requests may carry these headers, but they cannot assert identity. if (token?.user_id) { settings['role'] = api.roleName || 'authenticated'; settings['jwt.claims.user_id'] = token.user_id; + + if (token.id) { + settings['jwt.claims.token_id'] = token.id; + } + } else if (api.isPublic === false && headers?.actorId) { + settings['role'] = api.roleName || 'authenticated'; + settings['jwt.claims.user_id'] = headers.actorId; + settings['jwt.claims.principal_id'] = headers.actorId; + if (headers.entityId) { + settings['jwt.claims.entity_id'] = headers.entityId; + } + if (headers.entityType) { + settings['jwt.claims.entity_type'] = headers.entityType; + } + if (headers.organizationId) { + settings['jwt.claims.organization_id'] = headers.organizationId; + } } else { settings['role'] = api.anonRole || 'anonymous'; + + // Preserve database attribution for anonymous requests. This is the + // tenant's own identity and is unrelated to client-supplied headers. + if (api.databaseId) { + settings['jwt.claims.entity_id'] = api.databaseId; + settings['jwt.claims.entity_type'] = 'database'; + } } // Session claims @@ -59,9 +122,18 @@ export function buildPgSettings(input: PgSettingsInput): Record settings['jwt.claims.intent'] = token.intent; } + // Credential metadata is available to tenant procedures for authorization + // and audit decisions, while the role remains the API's resolved role. + if (token?.access_level) { + settings['jwt.claims.access_level'] = token.access_level; + } + if (token?.kind) { + settings['jwt.claims.kind'] = token.kind; + } + // Principal identity (service accounts / bots) - if (token?.principal_id) { - settings['jwt.claims.principal_id'] = token.principal_id; + if (token?.user_id) { + settings['jwt.claims.principal_id'] = token.principal_id || token.user_id; } // Database context @@ -85,5 +157,19 @@ export function buildPgSettings(input: PgSettingsInput): Record settings['jwt.claims.ip_address'] = clientIp; } + if (origin) { + settings['jwt.claims.origin'] = origin; + } + if (userAgent) { + settings['jwt.claims.user_agent'] = userAgent; + } + if (deviceToken) { + settings['jwt.claims.device_token'] = deviceToken; + } + + if (token?.access_level === 'read_only') { + settings['default_transaction_read_only'] = 'on'; + } + return settings; } diff --git a/packages/express-context/src/types.ts b/packages/express-context/src/types.ts index c1a1170fb5..e8dbff6b3a 100644 --- a/packages/express-context/src/types.ts +++ b/packages/express-context/src/types.ts @@ -303,6 +303,8 @@ export interface ConstructiveContext { api: ApiStructure; /** Authenticated token (null for anonymous requests) */ token: ConstructiveAPIToken | null; + /** Resolved service cache key for the Graphile/runtime route, when present. */ + serviceKey?: string; /** pgSettings for SET LOCAL in tenant transactions */ pgSettings: Record; /** Database UUID from the API resolver */ diff --git a/packages/perf-harness/README.md b/packages/perf-harness/README.md index 5c6451bd41..9964c25569 100644 --- a/packages/perf-harness/README.md +++ b/packages/perf-harness/README.md @@ -118,3 +118,49 @@ package; `cache:analyze` validates local raw reports and produces both summaries Generated results are not committed. The [PR #1746 description](https://github.com/constructive-io/constructive/pull/1746) records the measured results and tradeoffs; the suite README contains reproduction instructions. + +## Connection lifecycle stability suite + +The TypeScript entry points in `src/connection-lifecycle` exercise the unpatched +PostGraphile service and the real `graphile-cache` lifecycle. Build the workspace +(or `pnpm --filter @constructive-io/perf-harness... build`), then run: + +```sh +node packages/perf-harness/dist/connection-lifecycle/suite.js \ + --database-url 'postgresql://postgres:password@localhost:5432/postgres' \ + --output /tmp/graphile-connection-lifecycle.json +``` + +Use a local PostgreSQL test administrator URL without query parameters. Its +named database is the administrative connection database; `pgsql-test` creates +and tears down a separate test database per worker. The suite defaults to one +fresh-process repetition, one concurrent lifecycle, a two-connection pool, and +`--scale 0.01`: 20 cycles for each service-only case and three cycles for each +Graphile case (72 lifecycles total). This is a low-resource functional check. + +For a separate stress run on a suitable machine, pass `--scale 1 --repetitions 3 +--concurrency 8`: 2,000 cycles per service-only case, 256 per Graphile case, and a +16-connection pool (21,072 lifecycles total). Concurrency accepts 1–16; the pool +limit is twice concurrency, capped at 16. Each worker retains the harness's +five-minute deadline. + +Cases cover idle services, confirmed LISTEN subscriptions, release during +subscription startup, Graphile build/dispose, subscribed build/dispose, failed +schema builds, and replacement of generations using the same cache key. Runtime +GraphQL queries verify the built schema against a real table. Every 64 cycles +(and at the end), the worker separately observes pool quiescence and checks every +remaining backend for LISTEN channels. It also verifies that `pg_stat_activity` +returns to its pre-run baseline after idle expiry, before fixture teardown or +process exit can hide a leak. + +A successful public release call is counted immediately; it does not imply that +upstream background UNLISTEN/client return has completed. Reports distinguish +pending connections at return, peak pool size/checkout/waiting counts, cleanup +settlement latency, quiescent checkpoints, and retained client event listeners. +Listener accumulation is reported separately from checked-out connection leaks. +For this suite `buildMs` measures the whole churn loop, including its validation +checkpoints, rather than one +schema build. The failed-build case expects its injected schema errors. Network +partitions, backend termination, and production workload distributions are not +modeled. Keep generated JSON reports outside the repository; commit only the +TypeScript suite and its documentation. diff --git a/packages/perf-harness/package.json b/packages/perf-harness/package.json index 0230425f5d..4ec7a61b5a 100644 --- a/packages/perf-harness/package.json +++ b/packages/perf-harness/package.json @@ -37,6 +37,7 @@ "devDependencies": { "@types/node": "^22.19.11", "@types/pg": "^8.20.4", + "graphile-cache": "workspace:^", "makage": "^0.8.0", "pg-cache": "workspace:^", "pgsql-test": "workspace:^" diff --git a/packages/perf-harness/src/connection-lifecycle/suite.ts b/packages/perf-harness/src/connection-lifecycle/suite.ts new file mode 100644 index 0000000000..8fff587b65 --- /dev/null +++ b/packages/perf-harness/src/connection-lifecycle/suite.ts @@ -0,0 +1,37 @@ +import { resolve } from 'node:path'; + +import { parseValueArgs } from '../process'; +import { runBenchmarkSuite, writeJsonAtomically } from '../run'; +import type { BenchmarkSuiteDefinition } from '../types'; + +const main = async (): Promise => { + const { values } = parseValueArgs(process.argv.slice(2)); + const databaseUrl = values.get('database-url'); + const output = values.get('output'); + if (!databaseUrl || !output) throw new Error('--database-url and --output are required'); + const repetitions = Number(values.get('repetitions') ?? 1); + const scale = Number(values.get('scale') ?? 0.01); + const concurrency = Number(values.get('concurrency') ?? 1); + if (!Number.isSafeInteger(concurrency) || concurrency < 1 || concurrency > 16) { + throw new Error('concurrency must be between 1 and 16'); + } + if (!Number.isSafeInteger(repetitions) || repetitions < 1 || repetitions > 50 || !Number.isFinite(scale) || scale <= 0) { + throw new Error('repetitions must be 1–50 and scale must be positive'); + } + const suite: BenchmarkSuiteDefinition = { + name: 'unpatched-graphile-connection-lifecycle', + cases: ['idle', 'subscribed', 'startup-race', 'cache', 'cache-subscribed', 'failed-build', 'replacement'].map(mode => ({ + name: mode, + workerConfig: { mode, cycles: Math.max(1, Math.ceil((['idle', 'subscribed', 'startup-race'].includes(mode) ? 2000 : 256) * scale)), concurrency }, + ...(['cache', 'cache-subscribed', 'replacement'].includes(mode) ? { expectedSchemaGroup: 'lifecycle' } : {}), + })), + }; + const report = await runBenchmarkSuite(suite, { + databaseUrl, repetitions, seed: 1747, order: null, workerTimeoutMs: 300_000, + }, resolve(__dirname, 'worker.js')); + await writeJsonAtomically(output, report); + process.stdout.write(`${JSON.stringify({ output, validation: report.validation })}\n`); + if (report.validation.errors.length) process.exitCode = 1; +}; + +void main().catch(error => { console.error(error); process.exitCode = 1; }); diff --git a/packages/perf-harness/src/connection-lifecycle/worker.ts b/packages/perf-harness/src/connection-lifecycle/worker.ts new file mode 100644 index 0000000000..38d666edec --- /dev/null +++ b/packages/perf-harness/src/connection-lifecycle/worker.ts @@ -0,0 +1,271 @@ +import assert from 'node:assert/strict'; +import { createHash } from 'node:crypto'; +import { performance } from 'node:perf_hooks'; +import { setTimeout as delay } from 'node:timers/promises'; + +import { + clearGraphileCache, + createGraphileInstance, + disposeUncachedEntry, + graphileCache, + type GraphileCacheEntry, + waitForActiveDisposals, +} from 'graphile-cache'; +import { lexicographicSortSchema, printSchema } from 'graphql'; +import type { Pool } from 'pg'; +import { getConnections } from 'pgsql-test'; +import { makePgService } from 'postgraphile/adaptors/pg'; +import { grafast } from 'postgraphile/grafast'; +import { PostGraphileAmberPreset } from 'postgraphile/presets/amber'; + +import { measureBenchmarkCase } from '../metrics'; +import { parseWorkerProcessArgs, redactSecret, writeWorkerResult } from '../process'; +import type { JsonValue, SuccessfulWorkerResult } from '../types'; + +const modes = ['idle', 'subscribed', 'startup-race', 'cache', 'cache-subscribed', 'failed-build', 'replacement'] as const; +type Mode = typeof modes[number]; +interface Config { mode: Mode; cycles: number; concurrency: number } + +const parseConfig = (value: JsonValue): Config => { + const config = value as unknown as Config; + assert(config && modes.includes(config.mode), 'unknown lifecycle mode'); + assert(Number.isSafeInteger(config.cycles) && config.cycles > 0, 'cycles must be positive'); + assert(Number.isSafeInteger(config.concurrency) && config.concurrency > 0 && config.concurrency <= 16, + 'concurrency must be between 1 and 16'); + return config; +}; + +const waitUntil = async (predicate: () => boolean | Promise, label: string): Promise => { + const start = performance.now(); + while (!await predicate()) { + if (performance.now() - start > 5000) throw new Error(`Timed out: ${label}`); + await delay(2); + } + return performance.now() - start; +}; + +const run = async (): Promise => { + const { databaseUrl, envelope } = parseWorkerProcessArgs(process.argv.slice(2)); + const config = parseConfig(envelope.workerConfig); + // pgsql-test owns the isolated database, clients, pool, and final teardown. + const url = new URL(databaseUrl); + assert(['postgres:', 'postgresql:'].includes(url.protocol) && !url.search, + 'use a local PostgreSQL URL without query parameters'); + const fixture = await getConnections({ + pg: { host: url.hostname, port: Number(url.port || 5432), + user: decodeURIComponent(url.username), password: decodeURIComponent(url.password) }, + db: { rootDb: decodeURIComponent(url.pathname.slice(1)) || 'postgres' }, + }, []); + const poolConfig = { + ...fixture.pg.config, max: Math.min(16, config.concurrency * 2), + connectionTimeoutMillis: 5000, idleTimeoutMillis: 1000, + }; + const pool = fixture.manager.getPool(poolConfig); + let result: SuccessfulWorkerResult | undefined; + let primaryError: unknown; + const cleanupErrors: unknown[] = []; + let created = 0; + let released = 0; + let pendingAtReturn = 0; + let expectedBuildFailures = 0; + let peakTotal = 0; + let peakCheckedOut = 0; + let peakWaiting = 0; + let maxSettleMs = 0; + let peakNotificationListeners = 0; + let peakErrorListeners = 0; + let listenerWarnings = 0; + const onWarning = (warning: Error): void => { + if (warning.name === 'MaxListenersExceededWarning') listenerWarnings++; + }; + process.on('warning', onWarning); + let schemaHash = ''; + let schemaTypeCount = 0; + const checkpoints: JsonValue[] = []; + const releaseTimes: number[] = []; + const poolErrors: string[] = []; + pool.on('error', error => poolErrors.push(error.message)); + const sample = (): void => { + peakTotal = Math.max(peakTotal, pool.totalCount); + peakCheckedOut = Math.max(peakCheckedOut, pool.totalCount - pool.idleCount); + peakWaiting = Math.max(peakWaiting, pool.waitingCount); + }; + const sampler = setInterval(sample, 2); + const newService = () => { + created++; + const service = makePgService({ pool, schemas: ['cperf_lifecycle'] }); + const upstreamRelease = service.release.bind(service); + // Measure the actual public call in every path, including failed builds + // and disposal started by cache replacement. Do not infer cleanup counts. + service.release = () => releaseCall(upstreamRelease); + return service; + }; + const releaseCall = async (release: () => void | PromiseLike): Promise => { + const start = performance.now(); + await release(); + releaseTimes.push(performance.now() - start); + released++; + if (pool.totalCount !== pool.idleCount) pendingAtReturn++; + sample(); + }; + const subscribe = async (service: ReturnType, id: number): Promise => { + const topic = `cperf_${id}`; + const iterator = await service.pgSubscriber!.subscribe(topic); + if (config.mode === 'startup-race') return; + let received = false; + const next = iterator.next().then(value => { + assert.equal(value.value, 'ready'); + received = true; + }); + const deadline = performance.now() + 5000; + while (!received) { + assert(performance.now() < deadline, 'LISTEN readiness timed out'); + await fixture.pg.query('SELECT pg_notify($1, $2)', [topic, 'ready']); + await delay(2); + } + await next; + }; + const verifyChannels = async (target: Pool): Promise => { + // Hold all acquired clients until every idle backend has been inspected. + const clients = await Promise.all(Array.from({ length: target.totalCount }, () => target.connect())); + try { + for (const client of clients) { + peakNotificationListeners = Math.max(peakNotificationListeners, client.listenerCount('notification')); + peakErrorListeners = Math.max(peakErrorListeners, client.listenerCount('error')); + } + const results = await Promise.all(clients.map(client => client.query('SELECT pg_listening_channels()'))); + return results.reduce((count, result) => count + result.rowCount!, 0); + } finally { + for (const client of clients) client.release(); + } + }; + + try { + // Fixture setup is committed so independent Graphile connections can see it. + await fixture.pg.query('CREATE SCHEMA cperf_lifecycle; CREATE TABLE cperf_lifecycle.item (id int PRIMARY KEY, value int NOT NULL); INSERT INTO cperf_lifecycle.item VALUES (1, 42)'); + await fixture.db.query('SELECT 1'); + const baseline = await fixture.pg.query('SELECT count(*)::int AS n FROM pg_stat_activity WHERE datname = current_database()'); + result = await measureBenchmarkCase(envelope.caseName, async () => { + for (let offset = 0; offset < config.cycles; offset += config.concurrency) { + await Promise.all(Array.from({ length: Math.min(config.concurrency, config.cycles - offset) }, async (_, index) => { + const id = offset + index; + const service = newService(); + if (config.mode === 'idle' || config.mode === 'subscribed' || config.mode === 'startup-race') { + try { + if (config.mode !== 'idle') await subscribe(service, id); + } finally { + await service.release(); + } + return; + } + const failure = new Error('intentional lifecycle build failure'); + const entry = await createGraphileInstance({ + cacheKey: `cperf_${index}`, + preset: { + extends: [PostGraphileAmberPreset], + pgServices: [service], + grafserv: { graphqlPath: '/graphql' }, + ...(config.mode === 'failed-build' ? { + plugins: [{ name: 'LifecycleFailurePlugin', version: '1.0.0', schema: { + hooks: { GraphQLSchema: () => { throw failure; } }, + } }], + } : {}), + }, + }).catch((error): GraphileCacheEntry | null => { + if (config.mode !== 'failed-build' || error !== failure) throw error; + expectedBuildFailures++; + return null; + }); + if (!entry) return; + try { + const { schema } = await entry.pgl.getSchemaResult(); + const execution = await grafast({ schema, source: '{ allItems { nodes { value } } }', + resolvedPreset: entry.pgl.getResolvedPreset(), requestContext: {} }); + assert('data' in execution, `runtime query failed: ${JSON.stringify(execution)}`); + assert.equal(execution.errors, undefined); + assert.deepEqual(JSON.parse(JSON.stringify(execution.data)), { allItems: { nodes: [{ value: 42 }] } }); + schemaHash = createHash('sha256').update(printSchema(lexicographicSortSchema(schema))).digest('hex'); + schemaTypeCount = Object.keys(schema.getTypeMap()).length; + if (config.mode === 'cache-subscribed' || config.mode === 'replacement') await subscribe(service, id); + if (config.mode === 'replacement') { + // Reuse each key across generations while the old teardown runs. + graphileCache.set(entry.cacheKey, entry); + } else { + await disposeUncachedEntry(entry); + } + } catch (error) { + await disposeUncachedEntry(entry); + throw error; + } + })); + sample(); + // Observe a long churn window before quiescence, not after every release. + if ((offset + config.concurrency) % 64 === 0 || offset + config.concurrency >= config.cycles) { + if (config.mode === 'replacement') { + await clearGraphileCache(); + } + await waitForActiveDisposals(); + maxSettleMs = Math.max(maxSettleMs, await waitUntil( + () => pool.totalCount === pool.idleCount && pool.waitingCount === 0, 'pool quiescence' + )); + const channels = await verifyChannels(pool); + assert.equal(channels, 0, 'LISTEN channels remain after quiescence'); + const activity = await fixture.pg.query('SELECT count(*)::int AS n FROM pg_stat_activity WHERE datname = current_database()'); + checkpoints.push({ cycles: Math.min(offset + config.concurrency, config.cycles), + total: pool.totalCount, idle: pool.idleCount, waiting: pool.waitingCount, + backends: activity.rows[0].n, channels, peakNotificationListeners, peakErrorListeners }); + } + } + return baseline.rows[0].n as number; + }, async baselineBackends => { + assert.equal(released, created); + assert.equal(pool.totalCount - pool.idleCount, 0); + assert.deepEqual(poolErrors, []); + assert.equal(graphileCache.size, 0); + if (config.mode === 'failed-build') assert.equal(expectedBuildFailures, config.cycles); + const probe = await pool.query('SELECT 1 AS ok'); + assert.equal(probe.rows[0].ok, 1); + await waitUntil(() => pool.totalCount === 0, 'idle connection expiry'); + let finalBackends = -1; + const backendSettleMs = await waitUntil(async () => { + const activity = await fixture.pg.query('SELECT count(*)::int AS n FROM pg_stat_activity WHERE datname = current_database()'); + finalBackends = activity.rows[0].n; + return finalBackends === baselineBackends; + }, 'database backend baseline after idle expiry'); + releaseTimes.sort((a, b) => a - b); + return { + schemaHash: schemaHash || createHash('sha256').update('SELECT 1 AS ok').digest('hex'), + schemaTypeCount, runtimeVerified: true, + metadata: { ...config, poolMax: poolConfig.max, idleTimeoutMs: poolConfig.idleTimeoutMillis, + created, released, expectedBuildFailures, pendingAtReturn, + peakTotal, peakCheckedOut, peakWaiting, maxSettleMs, baselineBackends, + finalBackends, backendSettleMs, releaseSamples: releaseTimes.length, + releaseP99Ms: releaseTimes[Math.floor(releaseTimes.length * 0.99)], + checkpoints, poolErrors, peakNotificationListeners, peakErrorListeners, listenerWarnings }, + }; + }); + } catch (error) { + primaryError = error; + process.stderr.write(`Lifecycle failure: ${redactSecret(String(error), databaseUrl)}\n`); + process.stderr.write(`${JSON.stringify({ created, released, total: pool.totalCount, + idle: pool.idleCount, waiting: pool.waitingCount, checkpoints })}\n`); + } finally { + clearInterval(sampler); + process.off('warning', onWarning); + try { await clearGraphileCache(); } catch (error) { cleanupErrors.push(error); } + try { await fixture.teardown(); } catch (error) { cleanupErrors.push(error); } + } + if (primaryError || cleanupErrors.length) { + throw new AggregateError([...(primaryError ? [primaryError] : []), ...cleanupErrors], 'Lifecycle benchmark failed'); + } + writeWorkerResult(result!); +}; + +void run().catch(error => { + const { databaseUrl, envelope } = parseWorkerProcessArgs(process.argv.slice(2)); + const message = error instanceof AggregateError + ? error.errors.map(item => String(item?.stack ?? item)).join('\n') : String(error?.stack ?? error); + writeWorkerResult({ status: 'error', pid: process.pid, caseName: envelope.caseName, + error: redactSecret(message, databaseUrl) }); + process.exitCode = 1; +}); diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 78fffc290b..1a061e6242 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -541,6 +541,9 @@ importers: graphile/graphile-cache: dependencies: + '@constructive-io/errors': + specifier: workspace:^ + version: link:../../packages/errors/dist 12factor-env: specifier: workspace:^ version: link:../../packages/12factor-env/dist @@ -575,6 +578,9 @@ importers: nodemon: specifier: ^3.1.14 version: 3.1.14 + pgsql-test: + specifier: workspace:^ + version: link:../../postgres/pgsql-test/dist ts-node: specifier: ^10.9.2 version: 10.9.2(@types/node@25.9.1)(typescript@6.0.3) @@ -1834,6 +1840,12 @@ importers: graphql/explorer: dependencies: + '@constructive-io/errors': + specifier: workspace:^ + version: link:../../packages/errors/dist + '@pgpmjs/logger': + specifier: workspace:^ + version: link:../../pgpm/logger/dist '@constructive-io/graphql-env': specifier: workspace:^ version: link:../env/dist @@ -2306,6 +2318,9 @@ importers: specifier: ^7.0.0 version: 7.2.2 devDependencies: + graphile-cache: + specifier: workspace:^ + version: link:../../graphile/graphile-cache/dist '@0no-co/graphql.web': specifier: ^1.3.3 version: 1.3.3(graphql@16.13.0) @@ -2833,6 +2848,9 @@ importers: '@types/pg': specifier: ^8.20.4 version: 8.20.4 + graphile-cache: + specifier: workspace:^ + version: link:../../graphile/graphile-cache/dist makage: specifier: ^0.8.0 version: 0.8.0