diff --git a/.github/workflows/run-tests.yaml b/.github/workflows/run-tests.yaml index 93da3d912f..849fad656b 100644 --- a/.github/workflows/run-tests.yaml +++ b/.github/workflows/run-tests.yaml @@ -113,7 +113,7 @@ jobs: - batch: graphql packages: 'graphql/query graphql/codegen' - batch: graphile-unit - packages: 'graphile/graphile-plugin-utils graphile/graphile-cache graphile/graphile-realtime-subscriptions graphile/graphile-sql-expression-validator graphile/graphile-upload-plugin graphile/graphile-storage-registry' + packages: 'graphile/graphile-plugin-utils graphile/graphile-realtime-subscriptions graphile/graphile-sql-expression-validator graphile/graphile-upload-plugin graphile/graphile-storage-registry' - batch: agentic packages: 'agentic/protocol agentic/agentic-kit agentic/agent agentic/harness agentic/permissions agentic/chat agentic/cli agentic/db-tools agentic/pi agentic/dsh agentic/react agentic/agentic-server agentic/anthropic agentic/openai agentic/ollama agentic/run-log agentic/metering agentic/agent-conversation agentic/pi-host agentic/run-log-client agentic/run-log-gate agentic/machine-protocol agentic/machine-runner agentic/agent-cli' - batch: pgpm-unit @@ -246,7 +246,7 @@ jobs: - batch: pg-graphql packages: 'graphile/graphile-search graphile/graphile-function-source 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/__tests__/graphile-cache.test.ts b/graphile/graphile-cache/__tests__/graphile-cache.test.ts index b51ba0c017..be53ca7c09 100644 --- a/graphile/graphile-cache/__tests__/graphile-cache.test.ts +++ b/graphile/graphile-cache/__tests__/graphile-cache.test.ts @@ -5,6 +5,7 @@ import type { GraphileCacheEntry } from '../src'; import { cacheGraphileInstance, closeAllCaches, + disposeUncachedEntry, getCacheStats, graphileCache, GraphileCacheBudgetError} from '../src'; @@ -50,12 +51,16 @@ describe('graphile cache heap budget', () => { const small = entry('small', 2 * MiB); await cacheGraphileInstance('small', small.entry); const huge = entry('huge', 11 * MiB); + const releasePresetServices = jest.fn(async () => undefined); + huge.entry.releasePresetServices = releasePresetServices; await expect(cacheGraphileInstance('huge', huge.entry)).rejects.toThrow( GraphileCacheBudgetError ); + await disposeUncachedEntry(huge.entry); expect(huge.release).toHaveBeenCalledTimes(1); + expect(releasePresetServices).toHaveBeenCalledTimes(1); expect(graphileCache.has('huge')).toBe(false); expect(graphileCache.has('small')).toBe(true); }); 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 61c7d48a7f..4e058c4e12 100644 --- a/graphile/graphile-cache/package.json +++ b/graphile/graphile-cache/package.json @@ -42,6 +42,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-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..ec1c76a31d --- /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, + type GraphileCacheEntry, + waitForActiveDisposals +} from '../graphile-cache'; + +const entry = (key: string, release: () => Promise): GraphileCacheEntry => ({ + cacheKey: key, createdAt: Date.now(), heapBytes: 1024 * 1024, 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..2ae2d59aa2 --- /dev/null +++ b/graphile/graphile-cache/src/__tests__/disposal-lifecycle.test.ts @@ -0,0 +1,178 @@ +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(), + heapBytes: 1024 * 1024, + 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-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 acdd87bc32..0b91ef77bd 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, MIN_ENTRY_HEAP_BYTES } from './graphile-cache'; +import { awaitGraphileBuildReadiness } from './build-readiness'; +import { disposeUncachedEntry, type GraphileCacheEntry, MIN_ENTRY_HEAP_BYTES } from './graphile-cache'; +import { createPresetServicesReleaser } from './preset-services'; const log = new Logger('graphile-cache:create'); @@ -48,13 +50,46 @@ export const createGraphileInstance = async ( const heapBefore = process.memoryUsage().heapUsed; 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(); - await pgl.getSchema(); + 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: () => { + log.error( + 'Graphile build cleanup failed', + { stage: 'failed-build-release' } + ); + } + }); + const heapBytes = Math.max( process.memoryUsage().heapUsed - heapBefore, MIN_ENTRY_HEAP_BYTES @@ -68,6 +103,7 @@ export const createGraphileInstance = async ( cacheKey, createdAt: Date.now(), heapBytes, + releasePresetServices }; if (enableRealtime) { @@ -75,15 +111,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, @@ -92,12 +127,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 91c42c0d1a..0f510b065a 100644 --- a/graphile/graphile-cache/src/graphile-cache.ts +++ b/graphile/graphile-cache/src/graphile-cache.ts @@ -103,57 +103,117 @@ export interface GraphileCacheEntry { cacheKey: string; createdAt: number; heapBytes: number; + /** Idempotent release for pgServices owned by this exact preset generation. */ + releasePresetServices?: () => Promise; /** Optional RealtimeManager for cursor-tracked subscription delivery */ realtimeManager?: { stop(): Promise } | null; } +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(); -// One disposal per entry: closeAllCaches() and the LRU dispose callback both -// dispose the same entry, and PostGraphile throws if released twice. -const disposals = new WeakMap>(); - /** * Dispose a PostGraphile v5 cache entry * * Properly releases resources by: * 1. Closing the HTTP server if listening - * 2. Releasing the PostGraphile instance (which internally releases grafserv) + * 2. Stopping the realtime manager + * 3. Releasing PostGraphile/Grafserv and preset services */ -const releaseEntry = async (entry: GraphileCacheEntry, key: string): Promise => { +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); - } - } - // Release PostGraphile instance (this also releases grafserv internally) - if (entry.pgl) { - await entry.pgl.release(); + await entry.realtimeManager.stop(); } - } catch (err) { - log.error(`Error disposing PostGraphile[${key}]:`, err); + } 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'); }; -const disposeEntry = (entry: GraphileCacheEntry, key: string): Promise => { - let disposal = disposals.get(entry); - if (!disposal) { - disposal = releaseEntry(entry, key); - disposals.set(entry, disposal); - } - return disposal; +/** + * 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'); }; /** @@ -194,11 +254,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); } }); @@ -225,7 +281,7 @@ export async function cacheGraphileInstance( entry: GraphileCacheEntry ): Promise { if (entry.heapBytes > graphileCache.maxEntrySize) { - await disposeEntry(entry, key); + await scheduleDisposal(entry, key); throw new GraphileCacheBudgetError(key, entry.heapBytes, graphileCache.maxEntrySize); } graphileCache.set(key, entry); @@ -296,6 +352,19 @@ const unregister = pgCache.registerCleanupCallback((pgPoolKey: string) => { // 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 * @@ -314,22 +383,20 @@ export const closeAllCaches = async (verbose = false): Promise => { try { if (verbose) log.info('Closing all server caches...'); - 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); } - - // clear() starts each entry's disposal; wait for those same disposals. - graphileCache.clear(); - await Promise.allSettled( - entries.map(([key, entry]) => disposeEntry(entry, key)) - ); - 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 43b6e861e7..07ae83a43e 100644 --- a/graphile/graphile-cache/src/index.ts +++ b/graphile/graphile-cache/src/index.ts @@ -9,9 +9,11 @@ export { cacheGraphileInstance, // Cache stats CacheStats, + clearGraphileCache, // Clear matching entries clearMatchingEntries, closeAllCaches, + disposeUncachedEntry, // Eviction tracking EvictionReason, FIVE_MINUTES_MS, @@ -23,7 +25,9 @@ export { GraphileCacheEntry, MIN_ENTRY_HEAP_BYTES, // Time constants - ONE_HOUR_MS} from './graphile-cache'; + ONE_HOUR_MS, + waitForActiveDisposals, + waitForEntryDisposal} from './graphile-cache'; // 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/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 950d404e92..b6d9230a59 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 4307eb718e..8a121c31f0 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -614,6 +614,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) @@ -2928,6 +2931,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