diff --git a/.github/workflows/run-tests.yaml b/.github/workflows/run-tests.yaml index 7a89b17250..5a4b52fdf1 100644 --- a/.github/workflows/run-tests.yaml +++ b/.github/workflows/run-tests.yaml @@ -109,7 +109,7 @@ jobs: - batch: packages-core packages: 'packages/url-domains packages/coerce packages/csrf packages/oauth packages/12factor-env packages/orm packages/express-context packages/errors packages/llm-env packages/node-type-registry packages/query-spec packages/server-utils packages/site-deploy examples/site-deploy-ssg postgres/pg-cache postgres/pg-env' - batch: packages-services - packages: 'packages/postmaster packages/smtppostmaster packages/csv-to-pg packages/cli postgres/pgsql-client postgres/pg-ast' + packages: 'packages/postmaster packages/smtppostmaster packages/csv-to-pg packages/cli packages/perf-harness postgres/pgsql-client postgres/pg-ast' - batch: graphql packages: 'graphql/query graphql/codegen' - batch: graphile-unit diff --git a/graphile/graphile-realtime-subscriptions/README.md b/graphile/graphile-realtime-subscriptions/README.md index b546a5067c..f6f8ac2d7e 100644 --- a/graphile/graphile-realtime-subscriptions/README.md +++ b/graphile/graphile-realtime-subscriptions/README.md @@ -30,6 +30,31 @@ const preset = { 4. The subscription re-queries the source table with RLS enforced 5. The client receives `{ event, row }` where `row` reflects the current state +## Generation-Scoped Delivery + +`GenerationScopedRealtimeSubscriber` wraps a shared Grafast notification +source with an exact topic allowlist. Database notifications still fan out to +every generation that leased that topic, while `publish()` sends cursor +catch-up events only to subscriptions owned by that one Graphile generation. +The facade uses fixed bounded queues, fails a slow subscription on overflow, +and awaits its source iterators and source lease during `release()`. + +`RealtimeManager` accepts this explicit publisher capability. A transitional +`createPgSubscriberPublisher()` adapter retains compatibility with the current +`@dataplan/pg` subscriber, keeping its private emitter access out of the +manager. New shared-listener integrations should use the generation-scoped +facade and provide `allowedSourceSchemas` so cursor events cannot cross +generation boundaries. Omitting the schema allowlist is supported only by the +deprecated `pgSubscriber` adapter for existing callers. + +`RealtimeTopicCollector` receives the plugin's physical schema/table +descriptors during build and rejects missing, empty, changed, malformed, or +foreign topic sets. `ActivatableGenerationScopedRealtimeSubscriber` gives +PostGraphile a stable subscriber identity before schema construction, but +fails every subscribe/publish call until the validated exact-topic source is +installed. This two-phase boundary prevents an instance from serving while its +shared listener is incomplete. + ## Subscription Modes ### Phase 3a (current) diff --git a/graphile/graphile-realtime-subscriptions/__tests__/cursor-tracker.test.ts b/graphile/graphile-realtime-subscriptions/__tests__/cursor-tracker.test.ts index 05a3c9b50a..f3a5fe720d 100644 --- a/graphile/graphile-realtime-subscriptions/__tests__/cursor-tracker.test.ts +++ b/graphile/graphile-realtime-subscriptions/__tests__/cursor-tracker.test.ts @@ -21,6 +21,7 @@ jest.mock('@pgpmjs/logger', () => ({ import { CursorTracker, + CursorTrackerStartAbortedError, DEFAULT_BATCH_LIMIT, DEFAULT_HEARTBEAT_INTERVAL_MS, DEFAULT_POLL_INTERVAL_MS, @@ -51,6 +52,16 @@ function createChangeLogEntry(overrides: Partial = {}): ChangeLo }; } +function deferred() { + let resolve!: (value: T | PromiseLike) => void; + let reject!: (reason?: unknown) => void; + const promise = new Promise((resolvePromise, rejectPromise) => { + resolve = resolvePromise; + reject = rejectPromise; + }); + return { promise, resolve, reject }; +} + // --- Tests --- describe('CursorTracker defaults', () => { @@ -152,6 +163,58 @@ describe('CursorTracker.start()', () => { await tracker.stop(); }); + + it('fails readiness and rolls back when listener registration fails', async () => { + const error = new Error('touch denied'); + const pool: Queryable = { query: jest.fn().mockRejectedValue(error) }; + const onError = jest.fn(); + const tracker = new CursorTracker({ pool, onError }); + + await expect(tracker.start()).rejects.toBe(error); + + expect(tracker.isRunning).toBe(false); + expect(onError).toHaveBeenCalledWith(error); + expect((pool.query as jest.Mock).mock.calls).toHaveLength(1); + }); + + it('fails readiness and cleans up when the initial drain fails', async () => { + const error = new Error('drain denied'); + const pool: Queryable = { + query: jest.fn().mockImplementation(async (sql: string) => { + if (sql.includes('drain_changes')) throw error; + return { rows: [] }; + }) + }; + const tracker = new CursorTracker({ nodeId: 'strict-node', pool }); + + await expect(tracker.start()).rejects.toBe(error); + + expect(tracker.isRunning).toBe(false); + expect(pool.query).toHaveBeenCalledWith( + expect.stringContaining('cleanup_ephemeral'), + ['strict-node'] + ); + }); + + it('preserves startup and rollback failures when both operations fail', async () => { + const drainError = new Error('drain denied'); + const cleanupError = new Error('cleanup denied'); + const pool: Queryable = { + query: jest.fn().mockImplementation(async (sql: string) => { + if (sql.includes('drain_changes')) throw drainError; + if (sql.includes('cleanup_ephemeral')) throw cleanupError; + return { rows: [] }; + }) + }; + const tracker = new CursorTracker({ nodeId: 'rollback-node', pool }); + + const starting = tracker.start(); + await expect(starting).rejects.toBeInstanceOf(AggregateError); + await expect(starting).rejects.toMatchObject({ + errors: [drainError, cleanupError] + }); + expect(tracker.isRunning).toBe(false); + }); }); describe('CursorTracker.stop()', () => { @@ -191,6 +254,28 @@ describe('CursorTracker.stop()', () => { expect(tracker.isRunning).toBe(false); }); + it('surfaces cleanup failure and permits an explicit retry', async () => { + const cleanupError = new Error('cleanup failed'); + let failCleanup = true; + const pool = createMockPool(); + pool.query.mockImplementation(async (sql: string) => { + if (failCleanup && sql.includes('cleanup_ephemeral')) throw cleanupError; + return { rows: [] }; + }); + const onError = jest.fn(); + const tracker = new CursorTracker({ pool, onError }); + await tracker.start(); + + await expect(tracker.stop()).rejects.toBe(cleanupError); + expect(tracker.isRunning).toBe(false); + expect(onError).toHaveBeenCalledWith(cleanupError); + + failCleanup = false; + await expect(tracker.stop()).resolves.toBeUndefined(); + expect(pool.query.mock.calls.filter(([sql]) => sql.includes('cleanup_ephemeral'))) + .toHaveLength(2); + }); + it('is idempotent (calling stop twice does not double-cleanup)', async () => { const mockPool = createMockPool(); const tracker = new CursorTracker({ @@ -223,6 +308,112 @@ describe('CursorTracker.stop()', () => { expect(clearSpy).toHaveBeenCalledTimes(2); clearSpy.mockRestore(); }); + + it('waits for an active poll and suppresses its dispatch after stop begins', async () => { + const pool = createMockPool(); + const onChanges = jest.fn(); + const tracker = new CursorTracker({ + nodeId: 'poll-stop-node', + pool, + onChanges, + }); + await tracker.start(); + + const poll = deferred<{ rows: { drain_changes: ChangeLogEntry }[] }>(); + pool.query.mockImplementation((sql: string) => { + if (sql.includes('drain_changes')) return poll.promise; + return Promise.resolve({ rows: [] }); + }); + pool.query.mockClear(); + + const activeDrain = tracker.drain(); + const stopping = tracker.stop(); + let stopped = false; + void stopping.then(() => { + stopped = true; + }); + await Promise.resolve(); + + expect(stopped).toBe(false); + expect(pool.query.mock.calls.some(([sql]) => sql.includes('cleanup_ephemeral'))).toBe(false); + + const entry = createChangeLogEntry(); + poll.resolve({ rows: [{ drain_changes: entry }] }); + await expect(activeDrain).resolves.toEqual([entry]); + await stopping; + + expect(onChanges).not.toHaveBeenCalled(); + expect(pool.query).toHaveBeenCalledWith( + expect.stringContaining('cleanup_ephemeral'), + ['poll-stop-node'] + ); + }); + + it('waits for an active heartbeat before cleaning up the listener', async () => { + const pool = createMockPool(); + const tracker = new CursorTracker({ + nodeId: 'heartbeat-stop-node', + pool, + }); + await tracker.start(); + + const heartbeat = deferred<{ rows: never[] }>(); + pool.query.mockImplementation((sql: string) => { + if (sql.includes('touch_listener')) return heartbeat.promise; + return Promise.resolve({ rows: [] }); + }); + pool.query.mockClear(); + + const activeHeartbeat = tracker.touchListener(); + const stopping = tracker.stop(); + let stopped = false; + void stopping.then(() => { + stopped = true; + }); + await Promise.resolve(); + + expect(stopped).toBe(false); + expect(pool.query.mock.calls.some(([sql]) => sql.includes('cleanup_ephemeral'))).toBe(false); + + heartbeat.resolve({ rows: [] }); + await activeHeartbeat; + await stopping; + + expect(pool.query).toHaveBeenCalledWith( + expect.stringContaining('cleanup_ephemeral'), + ['heartbeat-stop-node'] + ); + }); + + it('aborts startup deterministically when stop wins the registration race', async () => { + const registration = deferred<{ rows: never[] }>(); + const pool: jest.Mocked = { + query: jest.fn().mockImplementation((sql: string) => { + if (sql.includes('touch_listener')) return registration.promise; + return Promise.resolve({ rows: [] }); + }), + }; + const tracker = new CursorTracker({ + nodeId: 'start-stop-node', + pool, + }); + + const starting = tracker.start(); + const startResult = expect(starting).rejects.toBeInstanceOf(CursorTrackerStartAbortedError); + await Promise.resolve(); + await Promise.resolve(); + expect(pool.query.mock.calls.some(([sql]) => sql.includes('touch_listener'))).toBe(true); + + const stopping = tracker.stop(); + registration.resolve({ rows: [] }); + + await startResult; + await stopping; + + expect(tracker.isRunning).toBe(false); + expect(pool.query.mock.calls.some(([sql]) => sql.includes('drain_changes'))).toBe(false); + expect(pool.query.mock.calls.filter(([sql]) => sql.includes('cleanup_ephemeral'))).toHaveLength(1); + }); }); describe('CursorTracker.drain()', () => { @@ -465,7 +656,7 @@ describe('CursorTracker error handling', () => { })); }); - it('cleanup_ephemeral error calls onError without throwing', async () => { + it('cleanup_ephemeral error calls onError and rejects', async () => { const failingPool: Queryable = { query: jest.fn().mockRejectedValue(new Error('cleanup failed')), }; @@ -476,7 +667,7 @@ describe('CursorTracker error handling', () => { onError, }); - await tracker.cleanupEphemeral(); + await expect(tracker.cleanupEphemeral()).rejects.toThrow('cleanup failed'); expect(onError).toHaveBeenCalledWith(expect.objectContaining({ message: 'cleanup failed', diff --git a/graphile/graphile-realtime-subscriptions/__tests__/generation-subscriber.test.ts b/graphile/graphile-realtime-subscriptions/__tests__/generation-subscriber.test.ts new file mode 100644 index 0000000000..9b0f285f97 --- /dev/null +++ b/graphile/graphile-realtime-subscriptions/__tests__/generation-subscriber.test.ts @@ -0,0 +1,350 @@ +import type { GrafastSubscriber } from 'grafast'; + +import { + ActivatableGenerationScopedRealtimeSubscriber, + GENERATION_SUBSCRIBER_QUEUE_CAPACITY, + GenerationScopedRealtimeSubscriber, + RealtimeGenerationNotActiveError, + RealtimeGenerationOverflowError, + RealtimeGenerationSourceEndedError, + RealtimeGenerationTopicError +} from '../src/generation-subscriber'; + +interface Deferred { + promise: Promise; + resolve(value: T | PromiseLike): void; + reject(error: unknown): void; +} + +const deferred = (): Deferred => { + let resolve!: (value: T | PromiseLike) => void; + let reject!: (error: unknown) => void; + const promise = new Promise((resolvePromise, rejectPromise) => { + resolve = resolvePromise; + reject = rejectPromise; + }); + return { promise, resolve, reject }; +}; + +class ManualIterator implements AsyncIterableIterator { + private readonly buffered: string[] = []; + private readonly waiting: Deferred>[] = []; + private failure: Error | null = null; + private done = false; + readonly returnMock = jest.fn(async (): Promise> => { + this.complete(); + return { done: true, value: undefined }; + }); + + [Symbol.asyncIterator](): AsyncIterableIterator { + return this; + } + + next(): Promise> { + const value = this.buffered.shift(); + if (value !== undefined) return Promise.resolve({ done: false, value }); + if (this.failure) return Promise.reject(this.failure); + if (this.done) return Promise.resolve({ done: true, value: undefined }); + const result = deferred>(); + this.waiting.push(result); + return result.promise; + } + + return(): Promise> { + return this.returnMock(); + } + + throw(error?: unknown): Promise> { + const failure = error instanceof Error ? error : new Error(String(error)); + this.fail(failure); + return Promise.reject(failure); + } + + push(value: string): void { + const waiter = this.waiting.shift(); + if (waiter) waiter.resolve({ done: false, value }); + else this.buffered.push(value); + } + + fail(error: Error): void { + this.failure = error; + for (const waiter of this.waiting.splice(0)) waiter.reject(error); + } + + complete(): void { + this.done = true; + for (const waiter of this.waiting.splice(0)) { + waiter.resolve({ done: true, value: undefined }); + } + } +} + +class ManualSource implements GrafastSubscriber> { + readonly streams = new Map>(); + readonly release = jest.fn(async (): Promise => {}); + + subscribe(topic: string): AsyncIterableIterator { + const stream = new ManualIterator(); + let streams = this.streams.get(topic); + if (!streams) { + streams = new Set(); + this.streams.set(topic, streams); + } + streams.add(stream); + return stream; + } + + publish(topic: string, payload: string): void { + for (const stream of this.streams.get(topic) ?? []) stream.push(payload); + } + + fail(topic: string, error: Error): void { + for (const stream of this.streams.get(topic) ?? []) stream.fail(error); + } + + complete(topic: string): void { + for (const stream of this.streams.get(topic) ?? []) stream.complete(); + } +} + +const flushMicrotasks = async (): Promise => { + for (let index = 0; index < 8; index++) await Promise.resolve(); +}; + +describe('GenerationScopedRealtimeSubscriber', () => { + it('merges database notifications with generation-local cursor publications', async () => { + const source = new ManualSource(); + const facade = new GenerationScopedRealtimeSubscriber({ + source, + allowedTopics: ['realtime:tenant_a.contacts'] + }); + const stream = facade.subscribe('realtime:tenant_a.contacts'); + await flushMicrotasks(); + + source.publish('realtime:tenant_a.contacts', 'INSERT:db-row'); + await expect(stream.next()).resolves.toMatchObject({ value: 'INSERT:db-row' }); + + facade.publish('realtime:tenant_a.contacts', 'UPDATE:cursor-row'); + await expect(stream.next()).resolves.toMatchObject({ value: 'UPDATE:cursor-row' }); + await facade.release(); + expect(source.release).toHaveBeenCalledTimes(1); + }); + + it('enforces exact allowlists rather than prefixes', async () => { + const source = new ManualSource(); + const facade = new GenerationScopedRealtimeSubscriber({ + source, + allowedTopics: ['realtime:tenant.contacts'] + }); + + expect(() => facade.subscribe('realtime:tenant.contacts.private')) + .toThrow(RealtimeGenerationTopicError); + expect(() => facade.publish('realtime:tenant', 'INSERT:wrong')) + .toThrow(RealtimeGenerationTopicError); + await facade.release(); + }); + + it('keeps cursor publications inside their Graphile generation', async () => { + const source = new ManualSource(); + const first = new GenerationScopedRealtimeSubscriber({ + source, + allowedTopics: ['realtime:shared.contacts'], + releaseSourceOnRelease: false + }); + const second = new GenerationScopedRealtimeSubscriber({ + source, + allowedTopics: ['realtime:shared.contacts'], + releaseSourceOnRelease: false + }); + const firstStream = first.subscribe('realtime:shared.contacts'); + const secondStream = second.subscribe('realtime:shared.contacts'); + await flushMicrotasks(); + + first.publish('realtime:shared.contacts', 'INSERT:first-cursor'); + await expect(firstStream.next()).resolves.toMatchObject({ + value: 'INSERT:first-cursor' + }); + + source.publish('realtime:shared.contacts', 'UPDATE:database'); + await expect(firstStream.next()).resolves.toMatchObject({ value: 'UPDATE:database' }); + await expect(secondStream.next()).resolves.toMatchObject({ value: 'UPDATE:database' }); + await Promise.all([first.release(), second.release()]); + }); + + it('fails an overflowing local subscriber without poisoning its peers', async () => { + const source = new ManualSource(); + const facade = new GenerationScopedRealtimeSubscriber({ + source, + allowedTopics: ['realtime:events'] + }); + const slow = facade.subscribe('realtime:events'); + + for (let index = 0; index <= GENERATION_SUBSCRIBER_QUEUE_CAPACITY; index++) { + facade.publish('realtime:events', `INSERT:${index}`); + } + await expect(slow.next()).rejects.toBeInstanceOf(RealtimeGenerationOverflowError); + + const healthy = facade.subscribe('realtime:events'); + facade.publish('realtime:events', 'INSERT:healthy'); + await expect(healthy.next()).resolves.toMatchObject({ value: 'INSERT:healthy' }); + await facade.release(); + }); + + it('retains background subscription teardown failures for release', async () => { + const teardownError = new Error('source iterator release failed'); + const source = new ManualSource(); + const facade = new GenerationScopedRealtimeSubscriber({ + source, + allowedTopics: ['realtime:events'] + }); + const slow = facade.subscribe('realtime:events'); + await flushMicrotasks(); + [...source.streams.get('realtime:events')!][0] + .returnMock.mockRejectedValue(teardownError); + + for (let index = 0; index <= GENERATION_SUBSCRIBER_QUEUE_CAPACITY; index++) { + facade.publish('realtime:events', `INSERT:${index}`); + } + await expect(slow.next()).rejects.toBeInstanceOf(RealtimeGenerationOverflowError); + await flushMicrotasks(); + + await expect(facade.release()).rejects.toBe(teardownError); + expect(source.release).toHaveBeenCalledTimes(1); + }); + + it('propagates source failure and unexpected completion', async () => { + const source = new ManualSource(); + const facade = new GenerationScopedRealtimeSubscriber({ + source, + allowedTopics: ['a', 'b'] + }); + const failed = facade.subscribe('a'); + const ended = facade.subscribe('b'); + await flushMicrotasks(); + + source.fail('a', new Error('listener failed')); + source.complete('b'); + + await expect(failed.next()).rejects.toThrow('listener failed'); + await expect(ended.next()).rejects.toBeInstanceOf( + RealtimeGenerationSourceEndedError + ); + await facade.release(); + }); + + it('makes release idempotent and awaits stream and source teardown', async () => { + const source = new ManualSource(); + const streamReleased = deferred>(); + const sourceReleased = deferred(); + const facade = new GenerationScopedRealtimeSubscriber({ + source, + allowedTopics: ['a'] + }); + facade.subscribe('a'); + await flushMicrotasks(); + const sourceStream = [...source.streams.get('a')!][0]; + sourceStream.returnMock.mockImplementation(async () => streamReleased.promise); + source.release.mockImplementation(async () => sourceReleased.promise); + + const first = facade.release(); + const second = facade.release(); + expect(first).toBe(second); + await flushMicrotasks(); + expect(source.release).not.toHaveBeenCalled(); + + streamReleased.resolve({ done: true, value: undefined }); + await flushMicrotasks(); + expect(source.release).toHaveBeenCalledTimes(1); + + let settled = false; + void first.then(() => { + settled = true; + }); + await flushMicrotasks(); + expect(settled).toBe(false); + sourceReleased.resolve(); + await first; + expect(settled).toBe(true); + }); + + it('attempts every teardown and aggregates release failures', async () => { + const firstError = new Error('first stream release failed'); + const secondError = new Error('second stream release failed'); + const sourceError = new Error('source release failed'); + const source = new ManualSource(); + const facade = new GenerationScopedRealtimeSubscriber({ + source, + allowedTopics: ['a', 'b'] + }); + facade.subscribe('a'); + facade.subscribe('b'); + await flushMicrotasks(); + [...source.streams.get('a')!][0].returnMock.mockRejectedValue(firstError); + [...source.streams.get('b')!][0].returnMock.mockRejectedValue(secondError); + source.release.mockRejectedValue(sourceError); + + const releasing = facade.release(); + await expect(releasing).rejects.toBeInstanceOf(AggregateError); + await expect(releasing).rejects.toMatchObject({ + errors: [firstError, secondError, sourceError] + }); + expect(source.release).toHaveBeenCalledTimes(1); + }); +}); + +describe('ActivatableGenerationScopedRealtimeSubscriber', () => { + it('fails closed before activation and owns an activated source exactly once', async () => { + const source = new ManualSource(); + const facade = new ActivatableGenerationScopedRealtimeSubscriber(); + + expect(() => facade.subscribe('realtime:tenant_a.contacts')) + .toThrow(RealtimeGenerationNotActiveError); + await facade.activate({ + source, + allowedTopics: ['realtime:tenant_a.contacts'] + }); + + const stream = facade.subscribe('realtime:tenant_a.contacts'); + await flushMicrotasks(); + source.publish('realtime:tenant_a.contacts', 'INSERT:row-a'); + await expect(stream.next()).resolves.toMatchObject({ value: 'INSERT:row-a' }); + + const first = facade.release(); + const second = facade.release(); + expect(first).toBe(second); + await first; + expect(source.release).toHaveBeenCalledTimes(1); + }); + + it('releases a rejected second activation source', async () => { + const firstSource = new ManualSource(); + const secondSource = new ManualSource(); + const facade = new ActivatableGenerationScopedRealtimeSubscriber(); + await facade.activate({ source: firstSource, allowedTopics: ['a'] }); + + await expect(facade.activate({ source: secondSource, allowedTopics: ['a'] })) + .rejects.toMatchObject({ code: 'REALTIME_GENERATION_ALREADY_ACTIVE' }); + expect(secondSource.release).toHaveBeenCalledTimes(1); + await facade.release(); + expect(firstSource.release).toHaveBeenCalledTimes(1); + }); + + it('preserves activation and rejected-source release failures', async () => { + const firstSource = new ManualSource(); + const secondSource = new ManualSource(); + const releaseError = new Error('rejected source release failed'); + secondSource.release.mockRejectedValue(releaseError); + const facade = new ActivatableGenerationScopedRealtimeSubscriber(); + await facade.activate({ source: firstSource, allowedTopics: ['a'] }); + + const activating = facade.activate({ source: secondSource, allowedTopics: ['a'] }); + await expect(activating).rejects.toBeInstanceOf(AggregateError); + await expect(activating).rejects.toMatchObject({ + errors: [ + expect.objectContaining({ code: 'REALTIME_GENERATION_ALREADY_ACTIVE' }), + releaseError + ] + }); + await facade.release(); + }); +}); diff --git a/graphile/graphile-realtime-subscriptions/__tests__/plugin.test.ts b/graphile/graphile-realtime-subscriptions/__tests__/plugin.test.ts index 7f6669bbf5..8e7a0a5003 100644 --- a/graphile/graphile-realtime-subscriptions/__tests__/plugin.test.ts +++ b/graphile/graphile-realtime-subscriptions/__tests__/plugin.test.ts @@ -235,6 +235,38 @@ describe('createRealtimeSubscriptionsPlugin', () => { }); describe('table discovery', () => { + it('reports sorted credential-free physical topic descriptors during build', () => { + const onTopicsDiscovered = jest.fn(); + createRealtimeSubscriptionsPlugin({ onTopicsDiscovered }); + + const zeta = createMockCodec('zeta', { + realtime: true, + schemaName: 'tenant_a', + }); + const alpha = createMockCodec('alpha', { + realtime: true, + schemaName: 'tenant_a', + }); + capturedFactory!(createMockBuild({ + zeta: createMockResource('zeta', zeta), + alpha: createMockResource('alpha', alpha), + })); + + expect(onTopicsDiscovered).toHaveBeenCalledTimes(1); + expect(onTopicsDiscovered).toHaveBeenCalledWith([ + { topic: 'realtime:tenant_a.alpha', schema: 'tenant_a', table: 'alpha' }, + { topic: 'realtime:tenant_a.zeta', schema: 'tenant_a', table: 'zeta' }, + ]); + }); + + it('reports an explicit empty topic set', () => { + const onTopicsDiscovered = jest.fn(); + createRealtimeSubscriptionsPlugin({ onTopicsDiscovered }); + capturedFactory!(createMockBuild({})); + + expect(onTopicsDiscovered).toHaveBeenCalledWith([]); + }); + it('discovers tables with @realtime tag', () => { createRealtimeSubscriptionsPlugin(); diff --git a/graphile/graphile-realtime-subscriptions/__tests__/realtime-manager.test.ts b/graphile/graphile-realtime-subscriptions/__tests__/realtime-manager.test.ts index c0d10650e0..bb617c2f99 100644 --- a/graphile/graphile-realtime-subscriptions/__tests__/realtime-manager.test.ts +++ b/graphile/graphile-realtime-subscriptions/__tests__/realtime-manager.test.ts @@ -1,7 +1,14 @@ import { EventEmitter } from 'events'; -import { RealtimeManager } from '../src/realtime-manager'; -import { entryToChannel,entryToNotifyPayload, extractRowId } from '../src/realtime-manager'; +import { + entryToChannel, + entryToNotifyPayload, + extractRowId, + RealtimeManager, + RealtimeSourceSchemaConfigurationError, + RealtimeSourceSchemaViolationError, + RealtimeSubscriberUnavailableError +} from '../src/realtime-manager'; import type { ChangeLogEntry, Queryable } from '../src/types'; // --------------------------------------------------------------------------- @@ -34,6 +41,20 @@ function createMockPgSubscriber() { return { eventEmitter, subscribe: jest.fn() }; } +function deferred() { + let resolve!: (value: T | PromiseLike) => void; + let reject!: (reason?: unknown) => void; + const promise = new Promise((resolvePromise, rejectPromise) => { + resolve = resolvePromise; + reject = rejectPromise; + }); + return { promise, resolve, reject }; +} + +async function flushMicrotasks(): Promise { + for (let i = 0; i < 6; i++) await Promise.resolve(); +} + // --------------------------------------------------------------------------- // Unit tests: helper functions // --------------------------------------------------------------------------- @@ -129,6 +150,7 @@ describe('RealtimeManager', () => { return new RealtimeManager({ pgSubscriber: mockSubscriber, pool: mockPool, + allowedSourceSchemas: ['public', 'billing'], nodeId: 'test-manager-node', pollIntervalMs: 1000, heartbeatIntervalMs: 5000, @@ -175,6 +197,127 @@ describe('RealtimeManager', () => { ); }); + it('fails startup before registration when the subscriber emitter is unavailable', async () => { + const manager = createManager({ pgSubscriber: {} }); + + await expect(manager.start()).rejects.toBeInstanceOf( + RealtimeSubscriberUnavailableError + ); + + expect(manager.isRunning).toBe(false); + expect(mockPool.query).not.toHaveBeenCalled(); + }); + + it('uses an explicit publisher without inspecting PgSubscriber internals', async () => { + const publish = jest.fn(); + const opaqueSubscriber = Object.defineProperty({}, 'eventEmitter', { + get() { + throw new Error('private field accessed'); + } + }); + mockPool.query.mockImplementation(async (sql: string) => { + if (sql.includes('drain_changes')) { + return { + rows: [{ + drain_changes: makeEntry({ payload_after: { id: 'cursor-row' } }) + }] + }; + } + return { rows: [] }; + }); + const manager = createManager({ + publisher: { publish }, + pgSubscriber: opaqueSubscriber + }); + + await manager.start(); + expect(publish).toHaveBeenCalledWith( + 'realtime:public.contact', + 'INSERT:cursor-row' + ); + await manager.stop(); + }); + + it('fails the generation when the explicit publisher rejects delivery', async () => { + const failure = new Error('generation released'); + const fatalErrors: Error[] = []; + mockPool.query.mockImplementation(async (sql: string) => { + if (sql.includes('drain_changes')) { + return { rows: [{ drain_changes: makeEntry() }] }; + } + return { rows: [] }; + }); + const manager = createManager({ + publisher: { + publish() { + throw failure; + } + }, + onFatalError: (error: Error) => fatalErrors.push(error) + }); + + await expect(manager.start()).rejects.toBe(failure); + expect(fatalErrors).toEqual([failure]); + expect(manager.isRunning).toBe(false); + }); + + it('preflights every cursor topic before publishing any row in the batch', async () => { + const publish = jest.fn(); + const topicFailure = new Error('topic outside generation'); + mockPool.query.mockImplementation(async (sql: string) => { + if (sql.includes('drain_changes')) { + return { + rows: [ + { drain_changes: makeEntry({ source_table: 'contact' }) }, + { drain_changes: makeEntry({ source_table: 'private_table' }) } + ] + }; + } + return { rows: [] }; + }); + const manager = createManager({ + publisher: { + assertTopics(topics: readonly string[]) { + if (topics.includes('realtime:public.private_table')) throw topicFailure; + }, + publish + } + }); + + await expect(manager.start()).rejects.toBe(topicFailure); + expect(publish).not.toHaveBeenCalled(); + }); + + it('fails startup before registration when no source schema is allowed', async () => { + const manager = createManager({ allowedSourceSchemas: [] }); + + await expect(manager.start()).rejects.toBeInstanceOf( + RealtimeSourceSchemaConfigurationError + ); + + expect(manager.isRunning).toBe(false); + expect(mockPool.query).not.toHaveBeenCalled(); + }); + + it('requires a source-schema allowlist for an explicit publisher', async () => { + const manager = createManager({ + publisher: { publish: jest.fn() }, + allowedSourceSchemas: undefined + }); + + await expect(manager.start()).rejects.toBeInstanceOf( + RealtimeSourceSchemaConfigurationError + ); + expect(mockPool.query).not.toHaveBeenCalled(); + }); + + it('preserves the deprecated pgSubscriber path when no allowlist is supplied', async () => { + const manager = createManager({ allowedSourceSchemas: undefined }); + + await expect(manager.start()).resolves.toBeUndefined(); + await manager.stop(); + }); + it('is idempotent for start', async () => { const manager = createManager(); await manager.start(); @@ -190,7 +333,184 @@ describe('RealtimeManager', () => { await manager.stop(); // should be no-op }); + it('fails a running generation when periodic cursor polling fails', async () => { + const failure = new Error('periodic drain failed'); + const errors: Error[] = []; + const fatalErrors: Error[] = []; + let rejectDrain = false; + mockPool.query.mockImplementation(async (sql: string) => { + if (rejectDrain && sql.includes('drain_changes')) throw failure; + return { rows: [] }; + }); + const manager = createManager({ + onError: (error: Error) => errors.push(error), + onFatalError: (error: Error) => fatalErrors.push(error) + }); + + await manager.start(); + rejectDrain = true; + await jest.advanceTimersByTimeAsync(1000); + await flushMicrotasks(); + await manager.stop(); + + expect(errors).toEqual([failure]); + expect(fatalErrors).toEqual([failure]); + expect(manager.isRunning).toBe(false); + expect(mockPool.query).toHaveBeenCalledWith( + expect.stringContaining('cleanup_ephemeral'), + ['test-manager-node'] + ); + }); + + it('fails a running generation when its periodic heartbeat fails', async () => { + const failure = new Error('periodic heartbeat failed'); + const errors: Error[] = []; + const fatalErrors: Error[] = []; + let rejectHeartbeat = false; + mockPool.query.mockImplementation(async (sql: string) => { + if (rejectHeartbeat && sql.includes('touch_listener')) throw failure; + return { rows: [] }; + }); + const manager = createManager({ + onError: (error: Error) => errors.push(error), + onFatalError: (error: Error) => fatalErrors.push(error) + }); + + await manager.start(); + rejectHeartbeat = true; + await jest.advanceTimersByTimeAsync(5000); + await flushMicrotasks(); + await manager.stop(); + + expect(errors).toEqual([failure]); + expect(fatalErrors).toEqual([failure]); + expect(manager.isRunning).toBe(false); + expect(mockPool.query).toHaveBeenCalledWith( + expect.stringContaining('cleanup_ephemeral'), + ['test-manager-node'] + ); + }); + + it('does not dispatch a deferred startup drain after stop begins', async () => { + const entry = makeEntry({ payload_after: { id: 'late-row' } }); + const drain = deferred<{ rows: { drain_changes: ChangeLogEntry }[] }>(); + const emitted: string[] = []; + mockSubscriber.eventEmitter.on('realtime:public.contact', (payload: string) => { + emitted.push(payload); + }); + mockPool.query.mockImplementation((sql: string) => { + if (sql.includes('drain_changes')) return drain.promise; + return Promise.resolve({ rows: [] }); + }); + + const manager = createManager(); + const starting = manager.start(); + const startResult = expect(starting).rejects.toMatchObject({ + code: 'CURSOR_TRACKER_START_ABORTED', + }); + await flushMicrotasks(); + expect(mockPool.query.mock.calls.some(([sql]) => sql.includes('drain_changes'))).toBe(true); + + const stopping = manager.stop(); + drain.resolve({ rows: [{ drain_changes: entry }] }); + + await startResult; + await stopping; + + expect(emitted).toEqual([]); + expect(manager.isRunning).toBe(false); + expect(mockPool.query).toHaveBeenCalledWith( + expect.stringContaining('cleanup_ephemeral'), + ['test-manager-node'] + ); + }); + describe('event dispatching', () => { + it('rejects a mixed batch atomically when it contains a foreign source schema', async () => { + const emitted: string[] = []; + const errors: Error[] = []; + const fatalErrors: Error[] = []; + mockSubscriber.eventEmitter.on('realtime:public.contact', (payload: string) => { + emitted.push(payload); + }); + const entries = [ + makeEntry({ payload_after: { id: 'allowed-row' } }), + makeEntry({ + source_schema: 'tenant_b', + payload_after: { id: 'foreign-row' } + }) + ]; + mockPool.query.mockImplementation(async (sql: string) => { + if (sql.includes('drain_changes')) { + return { rows: entries.map((entry) => ({ drain_changes: entry })) }; + } + return { rows: [] }; + }); + + const manager = createManager({ + allowedSourceSchemas: ['public'], + onError: (error: Error) => errors.push(error), + onFatalError: (error: Error) => fatalErrors.push(error) + }); + + await expect(manager.start()).rejects.toBeInstanceOf( + RealtimeSourceSchemaViolationError + ); + await manager.stop(); + + expect(emitted).toEqual([]); + expect(errors).toHaveLength(1); + expect(errors[0]).toMatchObject({ + code: 'REALTIME_SOURCE_SCHEMA_VIOLATION', + sourceSchema: 'tenant_b', + allowedSourceSchemas: ['public'] + }); + expect(fatalErrors).toEqual([errors[0]]); + expect(manager.isRunning).toBe(false); + }); + + it('stops a running manager before a foreign periodic batch can emit', async () => { + const errors: Error[] = []; + const fatalErrors: Error[] = []; + const emitted: string[] = []; + mockSubscriber.eventEmitter.on('realtime:public.contact', (payload: string) => { + emitted.push(payload); + }); + const manager = createManager({ + allowedSourceSchemas: ['public'], + onError: (error: Error) => errors.push(error), + onFatalError: (error: Error) => fatalErrors.push(error) + }); + await manager.start(); + mockPool.query.mockImplementation(async (sql: string) => { + if (sql.includes('drain_changes')) { + return { + rows: [{ + drain_changes: makeEntry({ + source_schema: 'tenant_b', + payload_after: { id: 'foreign-periodic-row' } + }) + }] + }; + } + return { rows: [] }; + }); + + await jest.advanceTimersByTimeAsync(1000); + await flushMicrotasks(); + await manager.stop(); + + expect(emitted).toEqual([]); + expect(errors).toHaveLength(1); + expect(errors[0]).toBeInstanceOf(RealtimeSourceSchemaViolationError); + expect(fatalErrors).toEqual([errors[0]]); + expect(manager.isRunning).toBe(false); + expect(mockPool.query).toHaveBeenCalledWith( + expect.stringContaining('cleanup_ephemeral'), + ['test-manager-node'] + ); + }); + it('emits cursor-tracked events on PgSubscriber eventEmitter', async () => { const emitted: { channel: string; payload: string }[] = []; mockSubscriber.eventEmitter.on('realtime:public.contact', (payload: string) => { @@ -290,7 +610,7 @@ describe('RealtimeManager', () => { }); describe('error handling', () => { - it('calls onError when drain fails', async () => { + it('fails startup and rolls back readiness when the initial drain fails', async () => { const errors: Error[] = []; mockPool.query.mockImplementation(async (sql: string) => { @@ -301,30 +621,12 @@ describe('RealtimeManager', () => { }); const manager = createManager({ onError: (err: Error) => errors.push(err) }); - await manager.start(); + await expect(manager.start()).rejects.toThrow('drain failed'); expect(errors).toHaveLength(1); expect(errors[0].message).toBe('drain failed'); - - await manager.stop(); + expect(manager.isRunning).toBe(false); }); - it('handles missing eventEmitter gracefully', async () => { - const entries: ChangeLogEntry[] = [ - makeEntry({ operation: 'INSERT', payload_after: { id: 'row-x' } }), - ]; - - mockPool.query.mockImplementation(async (sql: string) => { - if (typeof sql === 'string' && sql.includes('drain_changes')) { - return { rows: entries.map((e) => ({ drain_changes: e })) }; - } - return { rows: [] }; - }); - - // pgSubscriber without eventEmitter — should not crash - const manager = createManager({ pgSubscriber: {} }); - await manager.start(); - await manager.stop(); - }); }); }); diff --git a/graphile/graphile-realtime-subscriptions/__tests__/topic-collector.test.ts b/graphile/graphile-realtime-subscriptions/__tests__/topic-collector.test.ts new file mode 100644 index 0000000000..f38af65514 --- /dev/null +++ b/graphile/graphile-realtime-subscriptions/__tests__/topic-collector.test.ts @@ -0,0 +1,66 @@ +import { + RealtimeTopicCollector, + RealtimeTopicDiscoveryError +} from '../src/topic-collector'; + +describe('RealtimeTopicCollector', () => { + it('returns sorted exact physical topics for allowed schemas', () => { + const collector = new RealtimeTopicCollector(); + collector.collect([ + { topic: 'realtime:tenant_a.z', schema: 'tenant_a', table: 'z' }, + { topic: 'realtime:tenant_a.a', schema: 'tenant_a', table: 'a' } + ]); + + expect(collector.exactTopics(['tenant_a'])).toEqual([ + 'realtime:tenant_a.a', + 'realtime:tenant_a.z' + ]); + }); + + it.each([ + { + descriptors: [], + schemas: ['tenant_a'], + code: 'REALTIME_TOPIC_DISCOVERY_EMPTY' + }, + { + descriptors: [ + { topic: 'realtime:tenant_b.items', schema: 'tenant_b', table: 'items' } + ], + schemas: ['tenant_a'], + code: 'REALTIME_TOPIC_DISCOVERY_FOREIGN' + }, + { + descriptors: [ + { topic: 'realtime:tenant.a.items', schema: 'tenant.a', table: 'items' } + ], + schemas: ['tenant.a'], + code: 'REALTIME_TOPIC_DISCOVERY_INVALID' + } + ])('fails closed for $code', ({ descriptors, schemas, code }) => { + const collector = new RealtimeTopicCollector(); + expect(() => { + collector.collect(descriptors); + collector.exactTopics(schemas); + }).toThrow(expect.objectContaining({ + code + }) as RealtimeTopicDiscoveryError); + }); + + it('rejects missing discovery and post-discovery topic drift', () => { + const missing = new RealtimeTopicCollector(); + expect(() => missing.exactTopics(['tenant_a'])).toThrow(expect.objectContaining({ + code: 'REALTIME_TOPIC_DISCOVERY_MISSING' + }) as RealtimeTopicDiscoveryError); + + const changed = new RealtimeTopicCollector(); + changed.collect([ + { topic: 'realtime:tenant_a.items', schema: 'tenant_a', table: 'items' } + ]); + expect(() => changed.collect([ + { topic: 'realtime:tenant_a.users', schema: 'tenant_a', table: 'users' } + ])).toThrow(expect.objectContaining({ + code: 'REALTIME_TOPIC_DISCOVERY_CHANGED' + }) as RealtimeTopicDiscoveryError); + }); +}); diff --git a/graphile/graphile-realtime-subscriptions/src/cursor-tracker.ts b/graphile/graphile-realtime-subscriptions/src/cursor-tracker.ts index ab1f1204b9..6899050977 100644 --- a/graphile/graphile-realtime-subscriptions/src/cursor-tracker.ts +++ b/graphile/graphile-realtime-subscriptions/src/cursor-tracker.ts @@ -30,6 +30,17 @@ const DEFAULT_HEARTBEAT_INTERVAL_MS = 30000; const DEFAULT_BATCH_LIMIT = 500; const DEFAULT_SCHEMA = 'realtime_public'; +type CursorTrackerState = 'stopped' | 'starting' | 'running' | 'stopping'; + +export class CursorTrackerStartAbortedError extends Error { + readonly code = 'CURSOR_TRACKER_START_ABORTED'; + + constructor() { + super('CursorTracker was stopped before startup completed'); + this.name = 'CursorTrackerStartAbortedError'; + } +} + export class CursorTracker { readonly nodeId: string; @@ -43,8 +54,13 @@ export class CursorTracker { private pollTimer: ReturnType | null = null; private heartbeatTimer: ReturnType | null = null; - private running = false; - private draining = false; + private state: CursorTrackerState = 'stopped'; + private generation = 0; + private registered = false; + private startPromise: Promise | null = null; + private stopPromise: Promise | null = null; + private activeDrain: Promise | null = null; + private activeHeartbeat: Promise | null = null; constructor(options: CursorTrackerOptions) { this.nodeId = options.nodeId ?? randomUUID(); @@ -59,32 +75,122 @@ export class CursorTracker { }); } - async start(): Promise { - if (this.running) return; - this.running = true; + start(): Promise { + if (this.state === 'running') return Promise.resolve(); + if (this.state === 'starting') return this.startPromise!; + if (this.state === 'stopping') { + return (this.stopPromise ?? Promise.resolve()).then(() => this.start()); + } + + const generation = ++this.generation; + this.state = 'starting'; + const pending = this.startInternal(generation); + this.startPromise = pending; + void pending.then( + () => { + if (this.startPromise === pending) this.startPromise = null; + }, + () => { + if (this.startPromise === pending) this.startPromise = null; + } + ); + return pending; + } + private async startInternal(generation: number): Promise { log.info(`Starting cursor tracker: node=${this.nodeId}, schema=${this.schema}`); + try { + // A manual operation may have started while the tracker was stopped. + // Readiness must execute its own strict registration and drain rather + // than coalescing onto a non-strict operation. + await this.waitForActiveWork(); + this.assertStartCurrent(generation); + + // Startup is a readiness boundary: the instance must not become resident + // when the runtime role cannot register or drain the configured schema. + await this.touchListenerInternal(true); + this.registered = true; + this.assertStartCurrent(generation); - await this.touchListener(); + // A caller can request a manual drain while registration is in flight. + // Let it settle, then run the strict readiness drain ourselves so a + // best-effort call can never satisfy the startup boundary. + await this.waitForActiveWork(); + this.assertStartCurrent(generation); + await this.drainInternal(true, generation); + this.assertStartCurrent(generation); - // Initial drain immediately after registration - await this.drain(); + this.state = 'running'; - this.pollTimer = setInterval(() => { - void this.drain(); - }, this.pollIntervalMs); + this.pollTimer = setInterval(() => { + void this.drain(); + }, this.pollIntervalMs); + this.pollTimer.unref?.(); - this.heartbeatTimer = setInterval(() => { - void this.touchListener(); - }, this.heartbeatIntervalMs); + this.heartbeatTimer = setInterval(() => { + void this.touchListener(); + }, this.heartbeatIntervalMs); + this.heartbeatTimer.unref?.(); + } catch (reason) { + this.clearTimers(); + const error = this.toError(reason); + let cleanupError: Error | null = null; + if (this.registered) { + try { + await this.cleanupEphemeralInternal(true); + this.registered = false; + } catch (cleanupReason) { + cleanupError = this.toError(cleanupReason); + } + } + if (this.state === 'starting') this.state = 'stopped'; + if (cleanupError) { + throw new AggregateError( + [error, cleanupError], + 'CursorTracker startup and rollback both failed' + ); + } + throw error; + } } - async stop(): Promise { - if (!this.running) return; - this.running = false; + stop(): Promise { + if (this.state === 'stopped' && !this.registered) return Promise.resolve(); + if (this.state === 'stopping') return this.stopPromise!; + + const startInFlight = this.startPromise; + ++this.generation; + this.state = 'stopping'; + this.clearTimers(); log.info(`Stopping cursor tracker: node=${this.nodeId}`); + const pending = this.stopInternal(startInFlight); + this.stopPromise = pending; + void pending.then( + () => { + if (this.stopPromise === pending) this.stopPromise = null; + }, + () => { + if (this.stopPromise === pending) this.stopPromise = null; + } + ); + return pending; + } + + private async stopInternal(startInFlight: Promise | null): Promise { + try { + if (startInFlight) await Promise.allSettled([startInFlight]); + await this.waitForActiveWork(); + if (this.registered) { + await this.cleanupEphemeralInternal(true); + this.registered = false; + } + } finally { + this.state = 'stopped'; + } + } + private clearTimers(): void { if (this.pollTimer) { clearInterval(this.pollTimer); this.pollTimer = null; @@ -94,14 +200,39 @@ export class CursorTracker { clearInterval(this.heartbeatTimer); this.heartbeatTimer = null; } + } - await this.cleanupEphemeral(); + drain(): Promise { + if (this.state === 'stopping') return Promise.resolve([]); + const dispatchGeneration = this.state === 'starting' || this.state === 'running' + ? this.generation + : undefined; + return this.drainInternal(false, dispatchGeneration); } - async drain(): Promise { - if (this.draining) return []; - this.draining = true; + private drainInternal( + throwOnError: boolean, + dispatchGeneration?: number + ): Promise { + if (this.activeDrain) return Promise.resolve([]); + const pending = this.executeDrain(throwOnError, dispatchGeneration); + this.activeDrain = pending; + void pending.then( + () => { + if (this.activeDrain === pending) this.activeDrain = null; + }, + () => { + if (this.activeDrain === pending) this.activeDrain = null; + } + ); + return pending; + } + + private async executeDrain( + throwOnError: boolean, + dispatchGeneration?: number + ): Promise { try { const sql = `SELECT * FROM ${this.quoteIdent(this.schema)}.drain_changes($1, $2)`; const result = await this.pool.query<{ drain_changes: ChangeLogEntry }>( @@ -110,41 +241,100 @@ export class CursorTracker { ); const entries = result.rows.map((row) => row.drain_changes); - if (entries.length > 0) { + if (entries.length > 0 && this.mayDispatch(dispatchGeneration)) { log.info(`Drained ${entries.length} change(s) for node=${this.nodeId}`); this.onChanges(entries); } return entries; - } catch (err) { - this.onError(err instanceof Error ? err : new Error(String(err))); + } catch (reason) { + const error = this.toError(reason); + this.reportError(error); + if (throwOnError) throw error; return []; - } finally { - this.draining = false; } } - async touchListener(): Promise { + touchListener(): Promise { + if (this.state === 'stopping') return Promise.resolve(); + return this.touchListenerInternal(false); + } + + private touchListenerInternal(throwOnError: boolean): Promise { + if (this.activeHeartbeat) return this.activeHeartbeat; + const pending = this.executeTouchListener(throwOnError); + this.activeHeartbeat = pending; + void pending.then( + () => { + if (this.activeHeartbeat === pending) this.activeHeartbeat = null; + }, + () => { + if (this.activeHeartbeat === pending) this.activeHeartbeat = null; + } + ); + return pending; + } + + private async executeTouchListener(throwOnError: boolean): Promise { try { const sql = `SELECT ${this.quoteIdent(this.schema)}.touch_listener($1)`; await this.pool.query(sql, [this.nodeId]); - } catch (err) { - this.onError(err instanceof Error ? err : new Error(String(err))); + } catch (reason) { + const error = this.toError(reason); + this.reportError(error); + if (throwOnError) throw error; } } async cleanupEphemeral(): Promise { + await this.cleanupEphemeralInternal(true); + } + + private async cleanupEphemeralInternal(throwOnError: boolean): Promise { try { const sql = `SELECT ${this.quoteIdent(this.schema)}.cleanup_ephemeral($1)`; await this.pool.query(sql, [this.nodeId]); log.info(`Cleaned up ephemeral subscriptions for node=${this.nodeId}`); - } catch (err) { - this.onError(err instanceof Error ? err : new Error(String(err))); + } catch (reason) { + const error = this.toError(reason); + this.reportError(error); + if (throwOnError) throw error; } } get isRunning(): boolean { - return this.running; + return this.state === 'running'; + } + + private assertStartCurrent(generation: number): void { + if (this.state !== 'starting' || this.generation !== generation) { + throw new CursorTrackerStartAbortedError(); + } + } + + private mayDispatch(generation: number | undefined): boolean { + if (generation === undefined) return this.state !== 'stopping'; + return this.generation === generation + && (this.state === 'starting' || this.state === 'running'); + } + + private async waitForActiveWork(): Promise { + const active: Promise[] = []; + if (this.activeDrain) active.push(this.activeDrain); + if (this.activeHeartbeat) active.push(this.activeHeartbeat); + if (active.length > 0) await Promise.allSettled(active); + } + + private reportError(error: Error): void { + try { + this.onError(error); + } catch (callbackError) { + log.error(`CursorTracker error callback failed: ${String(callbackError)}`); + } + } + + private toError(reason: unknown): Error { + return reason instanceof Error ? reason : new Error(String(reason)); } private quoteIdent(identifier: string): string { diff --git a/graphile/graphile-realtime-subscriptions/src/generation-subscriber.ts b/graphile/graphile-realtime-subscriptions/src/generation-subscriber.ts new file mode 100644 index 0000000000..7e53001d3a --- /dev/null +++ b/graphile/graphile-realtime-subscriptions/src/generation-subscriber.ts @@ -0,0 +1,471 @@ +import type { GrafastSubscriber } from 'grafast'; + +import type { RealtimePublisher } from './types'; + +export const GENERATION_SUBSCRIBER_QUEUE_CAPACITY = 256; +export const REALTIME_GENERATION_TOPIC_ERROR_CODE = 'REALTIME_GENERATION_TOPIC_INVALID'; +export const REALTIME_GENERATION_RELEASED_ERROR_CODE = 'REALTIME_GENERATION_RELEASED'; +export const REALTIME_GENERATION_OVERFLOW_ERROR_CODE = 'REALTIME_GENERATION_OVERFLOW'; +export const REALTIME_GENERATION_SOURCE_ENDED_ERROR_CODE = 'REALTIME_GENERATION_SOURCE_ENDED'; +export const REALTIME_GENERATION_NOT_ACTIVE_ERROR_CODE = 'REALTIME_GENERATION_NOT_ACTIVE'; +export const REALTIME_GENERATION_ALREADY_ACTIVE_ERROR_CODE = 'REALTIME_GENERATION_ALREADY_ACTIVE'; + +type RealtimeTopicMap = Record; + +export interface GenerationScopedRealtimeSubscriberOptions< + TTopics extends RealtimeTopicMap +> { + /** Shared database notification source owned by this generation facade. */ + source: GrafastSubscriber; + /** Exact topics compiled into this Graphile generation. */ + allowedTopics: readonly (keyof TTopics & string)[]; + /** Defaults to true; set false only when lifecycle ownership lives elsewhere. */ + releaseSourceOnRelease?: boolean; +} + +interface Deferred { + promise: Promise; + resolve(value: T): void; + reject(error: unknown): void; +} + +const deferred = (): Deferred => { + let resolve!: (value: T) => void; + let reject!: (error: unknown) => void; + const promise = new Promise((resolvePromise, rejectPromise) => { + resolve = resolvePromise; + reject = rejectPromise; + }); + return { promise, resolve, reject }; +}; + +const toError = (reason: unknown): Error => ( + reason instanceof Error ? reason : new Error(String(reason)) +); + +export class RealtimeGenerationTopicError extends Error { + readonly code = REALTIME_GENERATION_TOPIC_ERROR_CODE; + + constructor(readonly topic: unknown) { + super(`Realtime topic ${JSON.stringify(topic)} is outside this generation's allowlist`); + this.name = 'RealtimeGenerationTopicError'; + } +} + +export class RealtimeGenerationReleasedError extends Error { + readonly code = REALTIME_GENERATION_RELEASED_ERROR_CODE; + + constructor() { + super('Realtime generation subscriber has been released'); + this.name = 'RealtimeGenerationReleasedError'; + } +} + +export class RealtimeGenerationOverflowError extends Error { + readonly code = REALTIME_GENERATION_OVERFLOW_ERROR_CODE; + + constructor( + readonly topic: string, + readonly capacity: number + ) { + super( + `Realtime generation queue for ${JSON.stringify(topic)} exceeded its ` + + `fixed capacity of ${capacity}` + ); + this.name = 'RealtimeGenerationOverflowError'; + } +} + +export class RealtimeGenerationSourceEndedError extends Error { + readonly code = REALTIME_GENERATION_SOURCE_ENDED_ERROR_CODE; + + constructor(readonly topic: string) { + super(`Realtime source for ${JSON.stringify(topic)} ended unexpectedly`); + this.name = 'RealtimeGenerationSourceEndedError'; + } +} + +export class RealtimeGenerationNotActiveError extends Error { + readonly code = REALTIME_GENERATION_NOT_ACTIVE_ERROR_CODE; + + constructor() { + super('Realtime generation subscriber has not been activated'); + this.name = 'RealtimeGenerationNotActiveError'; + } +} + +export class RealtimeGenerationAlreadyActiveError extends Error { + readonly code = REALTIME_GENERATION_ALREADY_ACTIVE_ERROR_CODE; + + constructor() { + super('Realtime generation subscriber has already been activated'); + this.name = 'RealtimeGenerationAlreadyActiveError'; + } +} + +class LocalQueue { + private readonly buffered: T[] = []; + private readonly waiting: Deferred>[] = []; + private terminal: 'open' | 'complete' | 'failed' = 'open'; + private failure: Error | null = null; + + constructor( + private readonly topic: string, + private readonly capacity: number + ) {} + + next(): Promise> { + if (this.buffered.length > 0) { + return Promise.resolve({ done: false, value: this.buffered.shift()! }); + } + if (this.terminal === 'failed') return Promise.reject(this.failure); + if (this.terminal === 'complete') { + return Promise.resolve({ done: true, value: undefined }); + } + const result = deferred>(); + this.waiting.push(result); + return result.promise; + } + + push(value: T): RealtimeGenerationOverflowError | null { + if (this.terminal !== 'open') return null; + const waiter = this.waiting.shift(); + if (waiter) { + waiter.resolve({ done: false, value }); + return null; + } + if (this.buffered.length >= this.capacity) { + const error = new RealtimeGenerationOverflowError(this.topic, this.capacity); + this.fail(error); + return error; + } + this.buffered.push(value); + return null; + } + + complete(): void { + if (this.terminal !== 'open') return; + this.terminal = 'complete'; + this.buffered.length = 0; + for (const waiter of this.waiting.splice(0)) { + waiter.resolve({ done: true, value: undefined }); + } + } + + fail(error: Error): void { + if (this.terminal !== 'open') return; + this.terminal = 'failed'; + this.failure = error; + this.buffered.length = 0; + for (const waiter of this.waiting.splice(0)) waiter.reject(error); + } +} + +class GenerationSubscription implements AsyncIterableIterator { + private readonly queue: LocalQueue; + private readonly sourceIteratorPromise: Promise>; + private sourceReturnPromise: Promise | null = null; + private stopped = false; + private stopPromise: Promise | null = null; + + constructor( + readonly topic: string, + source: GrafastSubscriber>, + private readonly onStop: ( + subscription: GenerationSubscription, + teardownError: Error | null + ) => void + ) { + this.queue = new LocalQueue(topic, GENERATION_SUBSCRIBER_QUEUE_CAPACITY); + this.sourceIteratorPromise = Promise.resolve().then(() => source.subscribe(topic)); + void this.pump(); + } + + [Symbol.asyncIterator](): AsyncIterableIterator { + return this; + } + + next(): Promise> { + return this.queue.next(); + } + + async return(value?: unknown): Promise> { + await this.stop(); + return { done: true, value: value as T }; + } + + async throw(error?: unknown): Promise> { + const failure = error instanceof Error ? error : new Error(String(error)); + await this.stop(failure); + throw failure; + } + + publish(value: T): void { + if (this.stopped) return; + const overflow = this.queue.push(value); + if (overflow) void this.stop(overflow).catch(() => {}); + } + + fail(error: Error): void { + if (this.stopped) return; + void this.stop(error).catch(() => {}); + } + + stop(error?: Error): Promise { + if (this.stopPromise) return this.stopPromise; + this.stopped = true; + if (error) this.queue.fail(error); + else this.queue.complete(); + this.stopPromise = this.returnSource().then( + () => this.onStop(this, null), + (reason) => { + const teardownError = toError(reason); + this.onStop(this, teardownError); + throw teardownError; + } + ); + return this.stopPromise; + } + + private async pump(): Promise { + try { + const iterator = await this.sourceIteratorPromise; + if (this.stopped) { + await this.returnSource(); + return; + } + for (;;) { + const result = await iterator.next(); + if (this.stopped) return; + if (result.done) { + this.fail(new RealtimeGenerationSourceEndedError(this.topic)); + return; + } + this.publish(result.value); + } + } catch (error) { + if (!this.stopped) { + this.fail(error instanceof Error ? error : new Error(String(error))); + } + } + } + + private returnSource(): Promise { + if (this.sourceReturnPromise) return this.sourceReturnPromise; + this.sourceReturnPromise = this.sourceIteratorPromise.then(async (iterator) => { + await iterator.return?.(); + }, () => { + // Source acquisition failure is already delivered to the output queue. + }); + return this.sourceReturnPromise; + } +} + +/** + * A Graphile-generation-local GrafastSubscriber. Database notifications are + * forwarded from the shared source, while cursor catch-up events published + * through publish() remain inside this exact generation. + */ +export class GenerationScopedRealtimeSubscriber< + TTopics extends RealtimeTopicMap = RealtimeTopicMap +> implements GrafastSubscriber, RealtimePublisher { + readonly allowedTopics: readonly (keyof TTopics & string)[]; + private readonly allowedTopicSet: ReadonlySet; + private readonly subscriptions = new Map< + string, + Set> + >(); + private readonly releaseSourceOnRelease: boolean; + private readonly source: GrafastSubscriber; + private readonly subscriptionTeardownErrors = new Set(); + private released = false; + private releasePromise: Promise | null = null; + + constructor(options: GenerationScopedRealtimeSubscriberOptions) { + if (!Array.isArray(options.allowedTopics) || options.allowedTopics.length === 0) { + throw new RealtimeGenerationTopicError(options.allowedTopics); + } + if (options.allowedTopics.some((topic) => typeof topic !== 'string')) { + throw new RealtimeGenerationTopicError(options.allowedTopics); + } + this.allowedTopics = Object.freeze([...new Set(options.allowedTopics)]); + this.allowedTopicSet = new Set(this.allowedTopics); + this.source = options.source; + this.releaseSourceOnRelease = options.releaseSourceOnRelease ?? true; + } + + subscribe( + topic: TTopic + ): AsyncIterableIterator { + if (this.released) throw new RealtimeGenerationReleasedError(); + if (typeof topic !== 'string' || !this.allowedTopicSet.has(topic)) { + throw new RealtimeGenerationTopicError(topic); + } + + let topicSubscriptions = this.subscriptions.get(topic); + if (!topicSubscriptions) { + topicSubscriptions = new Set(); + this.subscriptions.set(topic, topicSubscriptions); + } + const subscription = new GenerationSubscription( + topic, + this.source as GrafastSubscriber>, + (stopped, teardownError) => { + topicSubscriptions!.delete(stopped); + if (topicSubscriptions!.size === 0) this.subscriptions.delete(topic); + if (teardownError) this.subscriptionTeardownErrors.add(teardownError); + } + ); + topicSubscriptions.add(subscription); + return subscription as AsyncIterableIterator; + } + + assertTopics(topics: readonly string[]): void { + if (this.released) throw new RealtimeGenerationReleasedError(); + const invalid = topics.find((topic) => !this.allowedTopicSet.has(topic)); + if (invalid !== undefined) throw new RealtimeGenerationTopicError(invalid); + } + + publish(topic: string, payload: string): void { + this.assertTopics([topic]); + const subscriptions = this.subscriptions.get(topic); + if (!subscriptions) return; + for (const subscription of [...subscriptions]) subscription.publish(payload); + } + + release(): Promise { + if (this.releasePromise) return this.releasePromise; + this.released = true; + const active = [...this.subscriptions.values()].flatMap((entries) => [...entries]); + this.releasePromise = (async () => { + const results = await Promise.allSettled(active.map((subscription) => subscription.stop())); + const errors = new Set(this.subscriptionTeardownErrors); + for (const result of results) { + if (result.status === 'rejected') errors.add(toError(result.reason)); + } + if (this.releaseSourceOnRelease) { + try { + await this.source.release?.(); + } catch (reason) { + errors.add(toError(reason)); + } + } + const failures = [...errors]; + if (failures.length === 1) throw failures[0]; + if (failures.length > 1) { + throw new AggregateError( + failures, + 'Realtime generation subscriber release failed' + ); + } + })(); + return this.releasePromise; + } +} + +/** + * Stable subscriber identity installed into a PostGraphile pgService before + * schema construction. Activation installs the exact generation facade only + * after the build has reported all physical @realtime topics. + */ +export class ActivatableGenerationScopedRealtimeSubscriber< + TTopics extends RealtimeTopicMap = RealtimeTopicMap +> implements GrafastSubscriber, RealtimePublisher { + private delegate: GenerationScopedRealtimeSubscriber | null = null; + private released = false; + private releasePromise: Promise | null = null; + + async activate( + options: GenerationScopedRealtimeSubscriberOptions + ): Promise { + if (this.released) { + await this.releaseRejectedSource( + options.source, + new RealtimeGenerationReleasedError() + ); + } + if (this.delegate) { + await this.releaseRejectedSource( + options.source, + new RealtimeGenerationAlreadyActiveError() + ); + } + + try { + this.delegate = new GenerationScopedRealtimeSubscriber(options); + } catch (reason) { + await this.releaseRejectedSource(options.source, toError(reason)); + } + } + + subscribe( + topic: TTopic + ): AsyncIterableIterator { + if (this.released) throw new RealtimeGenerationReleasedError(); + if (!this.delegate) throw new RealtimeGenerationNotActiveError(); + return this.delegate.subscribe(topic); + } + + assertTopics(topics: readonly string[]): void { + if (this.released) throw new RealtimeGenerationReleasedError(); + if (!this.delegate) throw new RealtimeGenerationNotActiveError(); + this.delegate.assertTopics(topics); + } + + publish(topic: string, payload: string): void { + if (this.released) throw new RealtimeGenerationReleasedError(); + if (!this.delegate) throw new RealtimeGenerationNotActiveError(); + this.delegate.publish(topic, payload); + } + + release(): Promise { + if (this.releasePromise) return this.releasePromise; + this.released = true; + this.releasePromise = this.delegate?.release() ?? Promise.resolve(); + return this.releasePromise; + } + + private async releaseRejectedSource( + source: GrafastSubscriber, + error: Error + ): Promise { + try { + await source.release?.(); + } catch (reason) { + throw new AggregateError( + [error, toError(reason)], + 'Realtime generation activation and source release both failed' + ); + } + throw error; + } +} + +type LegacyEventEmitter = { + emit(topic: string, payload: string): boolean; +}; + +/** + * Transitional adapter for @dataplan/pg's current PgSubscriber. Private-field + * access is quarantined here; RealtimeManager and new integrations depend only + * on the explicit publisher capability. + */ +export const createPgSubscriberPublisher = ( + pgSubscriber: unknown +): RealtimePublisher | null => { + const candidate = pgSubscriber as { eventEmitter?: LegacyEventEmitter } | null; + const emitter = candidate && typeof candidate === 'object' + ? candidate.eventEmitter + : null; + if (!emitter || typeof emitter.emit !== 'function') return null; + const emit = emitter.emit.bind(emitter); + return Object.freeze({ + assertTopics(): void { + // The legacy PgSubscriber owns topic validation. New integrations use + // GenerationScopedRealtimeSubscriber's exact preflight instead. + }, + publish(topic: string, payload: string): void { + emit(topic, payload); + } + }); +}; diff --git a/graphile/graphile-realtime-subscriptions/src/index.ts b/graphile/graphile-realtime-subscriptions/src/index.ts index d0fdf741ca..456531e860 100644 --- a/graphile/graphile-realtime-subscriptions/src/index.ts +++ b/graphile/graphile-realtime-subscriptions/src/index.ts @@ -17,14 +17,52 @@ * ``` */ -export { CursorTracker } from './cursor-tracker'; +export { CursorTracker, CursorTrackerStartAbortedError } from './cursor-tracker'; +export type { GenerationScopedRealtimeSubscriberOptions } from './generation-subscriber'; +export { + ActivatableGenerationScopedRealtimeSubscriber, + createPgSubscriberPublisher, + GENERATION_SUBSCRIBER_QUEUE_CAPACITY, + GenerationScopedRealtimeSubscriber, + REALTIME_GENERATION_ALREADY_ACTIVE_ERROR_CODE, + REALTIME_GENERATION_NOT_ACTIVE_ERROR_CODE, + REALTIME_GENERATION_OVERFLOW_ERROR_CODE, + REALTIME_GENERATION_RELEASED_ERROR_CODE, + REALTIME_GENERATION_SOURCE_ENDED_ERROR_CODE, + REALTIME_GENERATION_TOPIC_ERROR_CODE, + RealtimeGenerationAlreadyActiveError, + RealtimeGenerationNotActiveError, + RealtimeGenerationOverflowError, + RealtimeGenerationReleasedError, + RealtimeGenerationSourceEndedError, + RealtimeGenerationTopicError +} from './generation-subscriber'; export { createRealtimeSubscriptionsPlugin, RealtimeSubscriptionsPlugin } from './plugin'; export { RealtimeSubscriptionsPreset } from './preset'; -export { RealtimeManager } from './realtime-manager'; -export type { RealtimeSubscriptionsPluginOptions } from './types'; +export { + RealtimeManager, + RealtimeManagerStartAbortedError, + RealtimeSourceSchemaConfigurationError, + RealtimeSourceSchemaViolationError, + RealtimeSubscriberUnavailableError +} from './realtime-manager'; +export { + REALTIME_TOPIC_DISCOVERY_CHANGED_ERROR_CODE, + REALTIME_TOPIC_DISCOVERY_EMPTY_ERROR_CODE, + REALTIME_TOPIC_DISCOVERY_FOREIGN_ERROR_CODE, + REALTIME_TOPIC_DISCOVERY_INVALID_ERROR_CODE, + REALTIME_TOPIC_DISCOVERY_MISSING_ERROR_CODE, + RealtimeTopicCollector, + RealtimeTopicDiscoveryError +} from './topic-collector'; +export type { + RealtimeSubscriptionsPluginOptions, + RealtimeTopicDescriptor +} from './types'; export type { ChangeLogEntry, CursorTrackerOptions, Queryable, RealtimeManagerOptions, + RealtimePublisher, } from './types'; diff --git a/graphile/graphile-realtime-subscriptions/src/plugin.ts b/graphile/graphile-realtime-subscriptions/src/plugin.ts index a013d778b4..6718f8ab9a 100644 --- a/graphile/graphile-realtime-subscriptions/src/plugin.ts +++ b/graphile/graphile-realtime-subscriptions/src/plugin.ts @@ -55,7 +55,10 @@ import { extendSchema } from 'graphile-utils'; import type { ParsedPayload } from './event-gate'; import { createGatedSubscriber } from './event-gate'; -import type { RealtimeSubscriptionsPluginOptions } from './types'; +import type { + RealtimeSubscriptionsPluginOptions, + RealtimeTopicDescriptor, +} from './types'; const log = new Logger('graphile-realtime-subscriptions'); @@ -223,6 +226,18 @@ export function createRealtimeSubscriptionsPlugin( return extendSchema( (build) => { const tables = discoverRealtimeTables(build); + const discoveredTopics: readonly RealtimeTopicDescriptor[] = Object.freeze( + tables + .map(({ notifyChannel, pgSchema, pgTable }) => Object.freeze({ + topic: notifyChannel, + schema: pgSchema, + table: pgTable, + })) + .sort((left, right) => ( + left.topic < right.topic ? -1 : left.topic > right.topic ? 1 : 0 + )), + ); + options.onTopicsDiscovered?.(discoveredTopics); if (tables.length === 0) { log.info('No tables with @realtime tag found — skipping subscription generation'); diff --git a/graphile/graphile-realtime-subscriptions/src/realtime-manager.ts b/graphile/graphile-realtime-subscriptions/src/realtime-manager.ts index 58bcf11926..6ff1f72cd9 100644 --- a/graphile/graphile-realtime-subscriptions/src/realtime-manager.ts +++ b/graphile/graphile-realtime-subscriptions/src/realtime-manager.ts @@ -1,17 +1,13 @@ /** * RealtimeManager — bridges CursorTracker (polling drain_changes) into - * PostGraphile's PgSubscriber so cursor-tracked events flow through the - * same subscription plans as NOTIFY events. + * a generation-local publisher so cursor-tracked events flow through the same + * subscription plans as NOTIFY events. * * Architecture: - * PgSubscriber uses an internal EventEmitter. NOTIFY payloads arrive via - * pg's `notification` event and are emitted as `eventEmitter.emit(channel, payload)`. - * The `listen()` step in grafast subscribes to the same EventEmitter. - * * RealtimeManager converts ChangeLogEntry objects from drain_changes() into - * the same NOTIFY payload format ("OP:rowId1,rowId2,...") and emits them on - * the PgSubscriber's EventEmitter, so existing subscription plans handle - * them identically to real NOTIFY events. + * the same NOTIFY payload format ("OP:rowId1,rowId2,...") and publishes them + * through an explicit capability. The generation-scoped subscriber keeps + * these cursor events local even when PostgreSQL LISTEN is shared. * * This provides at-least-once delivery: NOTIFY is instant but best-effort; * cursor polling catches up on anything missed (disconnects, restarts). @@ -19,20 +15,66 @@ * * Lifecycle: * 1. start() → registers listener node, begins polling + heartbeat - * 2. drain_changes() results are converted and emitted on PgSubscriber + * 2. drain_changes() results are converted and sent to the local publisher * 3. stop() → cleans up ephemeral subscriptions, removes listener node */ import { Logger } from '@pgpmjs/logger'; import { CursorTracker } from './cursor-tracker'; +import { createPgSubscriberPublisher } from './generation-subscriber'; import type { ChangeLogEntry, RealtimeManagerOptions, + RealtimePublisher, } from './types'; const log = new Logger('realtime-manager'); +type RealtimeManagerState = 'stopped' | 'starting' | 'running' | 'stopping'; + +export class RealtimeManagerStartAbortedError extends Error { + readonly code = 'REALTIME_MANAGER_START_ABORTED'; + + constructor() { + super('RealtimeManager was stopped before startup completed'); + this.name = 'RealtimeManagerStartAbortedError'; + } +} + +export class RealtimeSubscriberUnavailableError extends Error { + readonly code = 'REALTIME_SUBSCRIBER_UNAVAILABLE'; + + constructor() { + super('RealtimeManager requires a usable local publisher'); + this.name = 'RealtimeSubscriberUnavailableError'; + } +} + +export class RealtimeSourceSchemaViolationError extends Error { + readonly code = 'REALTIME_SOURCE_SCHEMA_VIOLATION'; + + constructor( + readonly sourceSchema: unknown, + readonly allowedSourceSchemas: readonly string[] + ) { + super( + `Realtime cursor returned source schema ${JSON.stringify(sourceSchema)} ` + + `outside the allowed Graphile schemas: ${allowedSourceSchemas.join(', ')}` + ); + this.name = 'RealtimeSourceSchemaViolationError'; + } +} + +export class RealtimeSourceSchemaConfigurationError extends Error { + readonly code = 'REALTIME_SOURCE_SCHEMAS_REQUIRED'; + + constructor() { + super('RealtimeManager requires at least one exact allowed source schema'); + this.name = 'RealtimeSourceSchemaConfigurationError'; + } +} + /** * Extract row IDs from a ChangeLogEntry. * @@ -69,12 +111,43 @@ function entryToChannel(entry: ChangeLogEntry): string { export class RealtimeManager { private readonly cursorTracker: CursorTracker; - private readonly subscriber: unknown; - private started = false; + private readonly publisher: RealtimePublisher | null; + private readonly allowedSourceSchemas: ReadonlySet; + private readonly allowedSourceSchemaList: readonly string[]; + private readonly requiresSourceSchemaAllowlist: boolean; + private readonly sourceSchemaConfigurationValid: boolean; + private readonly onFatalError?: (error: Error) => void; + private state: RealtimeManagerState = 'stopped'; + private generation = 0; + private dispatchEnabled = false; + private fatalError: Error | null = null; + private startPromise: Promise | null = null; + private stopPromise: Promise | null = null; constructor(options: RealtimeManagerOptions) { - const { pgSubscriber, pool, ...cursorOpts } = options; - this.subscriber = pgSubscriber; + const { + publisher, + pgSubscriber, + pool, + allowedSourceSchemas, + onFatalError, + ...cursorOpts + } = options; + this.publisher = publisher ?? createPgSubscriberPublisher(pgSubscriber); + this.onFatalError = onFatalError; + this.requiresSourceSchemaAllowlist = publisher !== undefined + || allowedSourceSchemas !== undefined; + this.sourceSchemaConfigurationValid = !this.requiresSourceSchemaAllowlist + || ( + Array.isArray(allowedSourceSchemas) + && allowedSourceSchemas.every( + (schema) => typeof schema === 'string' && schema.length > 0 + ) + ); + this.allowedSourceSchemaList = Object.freeze([ + ...new Set(allowedSourceSchemas ?? []) + ]); + this.allowedSourceSchemas = new Set(this.allowedSourceSchemaList); this.cursorTracker = new CursorTracker({ nodeId: cursorOpts.nodeId, @@ -84,9 +157,26 @@ export class RealtimeManager { batchLimit: cursorOpts.batchLimit, pool, onChanges: (entries) => this.dispatchEntries(entries), - onError: cursorOpts.onError ?? ((err) => { - log.error(`RealtimeManager error: ${err.message}`); - }), + onError: (error) => { + // Once readiness has completed, losing either cursor polling or the + // listener heartbeat means at-least-once delivery can no longer be + // claimed. Disable dispatch and begin shutdown before invoking the + // observational callback so a callback cannot leave a stale + // generation serving traffic by throwing or stopping it itself. + if (this.state === 'running') this.failDelivery(error); + + try { + if (cursorOpts.onError) { + cursorOpts.onError(error); + } else { + log.error(`RealtimeManager error: ${error.message}`); + } + } catch (callbackError) { + log.error( + `RealtimeManager error callback failed: ${String(callbackError)}` + ); + } + }, }); } @@ -95,62 +185,165 @@ export class RealtimeManager { } get isRunning(): boolean { - return this.started && this.cursorTracker.isRunning; + return this.state === 'running' && this.cursorTracker.isRunning; } - async start(): Promise { - if (this.started) return; - this.started = true; + start(): Promise { + if (this.state === 'running') return Promise.resolve(); + if (this.state === 'starting') return this.startPromise!; + if (this.state === 'stopping') { + return (this.stopPromise ?? Promise.resolve()).then(() => this.start()); + } + const generation = ++this.generation; + this.state = 'starting'; + this.dispatchEnabled = true; log.info(`Starting RealtimeManager: node=${this.nodeId}`); - await this.cursorTracker.start(); + const pending = this.startInternal(generation); + this.startPromise = pending; + void pending.then( + () => { + if (this.startPromise === pending) this.startPromise = null; + }, + () => { + if (this.startPromise === pending) this.startPromise = null; + } + ); + return pending; + } + + private async startInternal(generation: number): Promise { + try { + if ( + this.requiresSourceSchemaAllowlist + && ( + !this.sourceSchemaConfigurationValid + || this.allowedSourceSchemas.size === 0 + ) + ) { + throw new RealtimeSourceSchemaConfigurationError(); + } + if (!this.publisher || typeof this.publisher.publish !== 'function') { + throw new RealtimeSubscriberUnavailableError(); + } + await this.cursorTracker.start(); + if (this.state !== 'starting' || this.generation !== generation) { + throw new RealtimeManagerStartAbortedError(); + } + this.state = 'running'; + } catch (error) { + this.dispatchEnabled = false; + if (this.state === 'starting') this.state = 'stopped'; + throw error; + } } - async stop(): Promise { - if (!this.started) return; - this.started = false; + stop(): Promise { + if (this.state === 'stopped') return Promise.resolve(); + if (this.state === 'stopping') return this.stopPromise!; + const startInFlight = this.startPromise; + ++this.generation; + this.state = 'stopping'; + this.dispatchEnabled = false; log.info(`Stopping RealtimeManager: node=${this.nodeId}`); - await this.cursorTracker.stop(); + // Start the tracker shutdown synchronously so an in-flight drain is + // invalidated before it can dispatch after this method is called. + const trackerStop = this.cursorTracker.stop(); + const pending = this.stopInternal(startInFlight, trackerStop); + this.stopPromise = pending; + void pending.then( + () => { + if (this.stopPromise === pending) this.stopPromise = null; + }, + () => { + if (this.stopPromise === pending) this.stopPromise = null; + } + ); + return pending; + } + + private async stopInternal( + startInFlight: Promise | null, + trackerStop: Promise + ): Promise { + try { + if (startInFlight) await Promise.allSettled([startInFlight]); + await trackerStop; + } finally { + this.state = 'stopped'; + this.dispatchEnabled = false; + } } /** - * Convert ChangeLogEntry objects to NOTIFY-format payloads and emit - * them on the PgSubscriber's internal EventEmitter. + * Convert ChangeLogEntry objects to NOTIFY-format payloads and publish them + * through the exact generation's explicit local capability. */ private dispatchEntries(entries: ChangeLogEntry[]): void { - const emitter = this.getEventEmitter(); - if (!emitter) { - log.warn('PgSubscriber has no eventEmitter; cursor events cannot be dispatched'); - return; + if (!this.dispatchEnabled) return; + + const publisher = this.publisher; + if (!publisher) { + const error = new RealtimeSubscriberUnavailableError(); + this.failDelivery(error); + throw error; } - for (const entry of entries) { - const channel = entryToChannel(entry); - const payload = entryToNotifyPayload(entry); - emitter.emit(channel, payload); + // Validate the complete batch before emitting the first event. This keeps + // a mixed valid/foreign batch atomic from the tenant-isolation boundary's + // perspective: no event is delivered when routing is inconclusive. + const foreignEntry = this.requiresSourceSchemaAllowlist + ? entries.find( + (entry) => !this.allowedSourceSchemas.has(entry.source_schema) + ) + : undefined; + if (foreignEntry) { + const error = new RealtimeSourceSchemaViolationError( + foreignEntry.source_schema, + this.allowedSourceSchemaList + ); + this.failDelivery(error); + throw error; } - log.info(`Dispatched ${entries.length} cursor-tracked event(s) to PgSubscriber`); + const notifications = entries.map((entry) => ({ + channel: entryToChannel(entry), + payload: entryToNotifyPayload(entry) + })); + try { + publisher.assertTopics?.(notifications.map(({ channel }) => channel)); + for (const { channel, payload } of notifications) { + publisher.publish(channel, payload); + } + } catch (reason) { + const error = reason instanceof Error ? reason : new Error(String(reason)); + this.failDelivery(error); + throw error; + } + + log.info(`Dispatched ${entries.length} cursor-tracked event(s)`); } - /** - * Access PgSubscriber's internal EventEmitter. - * - * PgSubscriber from @dataplan/pg stores an EventEmitter3 instance as - * `this.eventEmitter`. It is private but stable across v1.x releases. - * This is the same emitter that NOTIFY events are dispatched through. - */ - private getEventEmitter(): { emit(event: string, payload: string): boolean } | null { - const sub = this.subscriber as Record; - if (sub && typeof sub === 'object' && 'eventEmitter' in sub) { - const ee = sub.eventEmitter as { emit(event: string, payload: string): boolean }; - if (typeof ee?.emit === 'function') { - return ee; + private failDelivery(error: Error): void { + this.dispatchEnabled = false; + const stopping = this.stop(); + if (!this.fatalError) { + this.fatalError = error; + try { + this.onFatalError?.(error); + } catch (callbackError) { + log.error( + `RealtimeManager fatal-error callback failed: ${String(callbackError)}` + ); } } - return null; + void stopping.catch((stopError) => { + log.error( + `RealtimeManager failed to stop after a delivery violation: ${String(stopError)}` + ); + }); } } -export { entryToChannel,entryToNotifyPayload, extractRowId }; +export { entryToChannel, entryToNotifyPayload, extractRowId }; diff --git a/graphile/graphile-realtime-subscriptions/src/topic-collector.ts b/graphile/graphile-realtime-subscriptions/src/topic-collector.ts new file mode 100644 index 0000000000..2704ccafc9 --- /dev/null +++ b/graphile/graphile-realtime-subscriptions/src/topic-collector.ts @@ -0,0 +1,173 @@ +import type { RealtimeTopicDescriptor } from './types'; + +export const REALTIME_TOPIC_DISCOVERY_MISSING_ERROR_CODE = + 'REALTIME_TOPIC_DISCOVERY_MISSING'; +export const REALTIME_TOPIC_DISCOVERY_EMPTY_ERROR_CODE = + 'REALTIME_TOPIC_DISCOVERY_EMPTY'; +export const REALTIME_TOPIC_DISCOVERY_INVALID_ERROR_CODE = + 'REALTIME_TOPIC_DISCOVERY_INVALID'; +export const REALTIME_TOPIC_DISCOVERY_FOREIGN_ERROR_CODE = + 'REALTIME_TOPIC_DISCOVERY_FOREIGN'; +export const REALTIME_TOPIC_DISCOVERY_CHANGED_ERROR_CODE = + 'REALTIME_TOPIC_DISCOVERY_CHANGED'; + +type RealtimeTopicDiscoveryCode = + | typeof REALTIME_TOPIC_DISCOVERY_MISSING_ERROR_CODE + | typeof REALTIME_TOPIC_DISCOVERY_EMPTY_ERROR_CODE + | typeof REALTIME_TOPIC_DISCOVERY_INVALID_ERROR_CODE + | typeof REALTIME_TOPIC_DISCOVERY_FOREIGN_ERROR_CODE + | typeof REALTIME_TOPIC_DISCOVERY_CHANGED_ERROR_CODE; + +export class RealtimeTopicDiscoveryError extends Error { + constructor( + readonly code: RealtimeTopicDiscoveryCode, + message: string + ) { + super(message); + this.name = 'RealtimeTopicDiscoveryError'; + } +} + +const containsUnpairedSurrogate = (value: string): boolean => { + for (let index = 0; index < value.length; index++) { + const code = value.charCodeAt(index); + if (code >= 0xd800 && code <= 0xdbff) { + const next = value.charCodeAt(index + 1); + if (!(next >= 0xdc00 && next <= 0xdfff)) return true; + index++; + } else if (code >= 0xdc00 && code <= 0xdfff) { + return true; + } + } + return false; +}; + +const assertIdentifierPart = ( + part: 'schema' | 'table', + value: unknown +): string => { + if (typeof value !== 'string' || value.length === 0) { + throw new RealtimeTopicDiscoveryError( + REALTIME_TOPIC_DISCOVERY_INVALID_ERROR_CODE, + `Realtime ${part} must be a non-empty string` + ); + } + if ( + value.includes('\0') + || value.includes('.') + || containsUnpairedSurrogate(value) + ) { + throw new RealtimeTopicDiscoveryError( + REALTIME_TOPIC_DISCOVERY_INVALID_ERROR_CODE, + `Realtime ${part} cannot be represented unambiguously in a notification topic` + ); + } + return value; +}; + +const normalizeDescriptor = ( + descriptor: RealtimeTopicDescriptor +): Readonly => { + const schema = assertIdentifierPart('schema', descriptor?.schema); + const table = assertIdentifierPart('table', descriptor?.table); + const expectedTopic = `realtime:${schema}.${table}`; + if ( + descriptor?.topic !== expectedTopic + || expectedTopic.includes('\0') + || containsUnpairedSurrogate(expectedTopic) + || Buffer.byteLength(expectedTopic, 'utf8') > 63 + ) { + throw new RealtimeTopicDiscoveryError( + REALTIME_TOPIC_DISCOVERY_INVALID_ERROR_CODE, + 'Realtime topic does not exactly match its physical schema/table or exceeds PostgreSQL limits' + ); + } + return Object.freeze({ topic: expectedTopic, schema, table }); +}; + +const descriptorKey = (descriptor: RealtimeTopicDescriptor): string => + `${descriptor.schema}\0${descriptor.table}\0${descriptor.topic}`; + +/** + * One schema-generation collector. It accepts repeated byte-equivalent build + * callbacks, but rejects topic drift so an already activated listener cannot + * silently become incomplete after a Graphile rebuild. + */ +export class RealtimeTopicCollector { + private descriptors: readonly Readonly[] | null = null; + + readonly collect = (input: readonly RealtimeTopicDescriptor[]): void => { + if (!Array.isArray(input)) { + throw new RealtimeTopicDiscoveryError( + REALTIME_TOPIC_DISCOVERY_INVALID_ERROR_CODE, + 'Realtime topic discovery did not provide an array' + ); + } + const byTopic = new Map>(); + for (const candidate of input) { + const descriptor = normalizeDescriptor(candidate); + const previous = byTopic.get(descriptor.topic); + if (previous && descriptorKey(previous) !== descriptorKey(descriptor)) { + throw new RealtimeTopicDiscoveryError( + REALTIME_TOPIC_DISCOVERY_INVALID_ERROR_CODE, + `Realtime notification topic ${JSON.stringify(descriptor.topic)} is ambiguous` + ); + } + byTopic.set(descriptor.topic, descriptor); + } + const next = Object.freeze( + [...byTopic.values()].sort((left, right) => ( + left.topic < right.topic ? -1 : left.topic > right.topic ? 1 : 0 + )) + ); + if (this.descriptors) { + const previousKeys = this.descriptors.map(descriptorKey); + const nextKeys = next.map(descriptorKey); + if ( + previousKeys.length !== nextKeys.length + || previousKeys.some((key, index) => key !== nextKeys[index]) + ) { + throw new RealtimeTopicDiscoveryError( + REALTIME_TOPIC_DISCOVERY_CHANGED_ERROR_CODE, + 'Realtime topics changed after the generation discovery boundary' + ); + } + return; + } + this.descriptors = next; + }; + + exactTopics(allowedSchemas: readonly string[]): readonly string[] { + if (!this.descriptors) { + throw new RealtimeTopicDiscoveryError( + REALTIME_TOPIC_DISCOVERY_MISSING_ERROR_CODE, + 'Realtime plugin did not report its compiled notification topics' + ); + } + if (this.descriptors.length === 0) { + throw new RealtimeTopicDiscoveryError( + REALTIME_TOPIC_DISCOVERY_EMPTY_ERROR_CODE, + 'Shared realtime requires at least one compiled @realtime topic' + ); + } + if ( + !Array.isArray(allowedSchemas) + || allowedSchemas.length === 0 + || allowedSchemas.some((schema) => typeof schema !== 'string' || schema.length === 0) + ) { + throw new RealtimeTopicDiscoveryError( + REALTIME_TOPIC_DISCOVERY_INVALID_ERROR_CODE, + 'Shared realtime requires at least one exact allowed physical schema' + ); + } + const allowed = new Set(allowedSchemas); + const foreign = this.descriptors.find(({ schema }) => !allowed.has(schema)); + if (foreign) { + throw new RealtimeTopicDiscoveryError( + REALTIME_TOPIC_DISCOVERY_FOREIGN_ERROR_CODE, + `Realtime topic ${JSON.stringify(foreign.topic)} is outside this Graphile generation` + ); + } + return Object.freeze(this.descriptors.map(({ topic }) => topic)); + } +} diff --git a/graphile/graphile-realtime-subscriptions/src/types.ts b/graphile/graphile-realtime-subscriptions/src/types.ts index bbf220ba4d..be0d5967f8 100644 --- a/graphile/graphile-realtime-subscriptions/src/types.ts +++ b/graphile/graphile-realtime-subscriptions/src/types.ts @@ -11,6 +11,25 @@ export interface RealtimeSubscriptionsPluginOptions { * Default: 50 */ overflowThreshold?: number; + + /** + * Receives the exact physical PostgreSQL notification topics compiled into + * this schema. The callback runs during schema construction, including with + * an empty list when no @realtime table was discovered. + * + * This is a build-time integration seam. It must not retain Graphile build + * objects or database resources; descriptors contain strings only. + */ + onTopicsDiscovered?: ( + topics: readonly RealtimeTopicDescriptor[] + ) => void; +} + +/** Credential-free description of one compiled @realtime channel. */ +export interface RealtimeTopicDescriptor { + readonly topic: string; + readonly schema: string; + readonly table: string; } /** @@ -28,6 +47,13 @@ export interface Queryable { ): Promise<{ rows: R[] }>; } +/** Explicit local delivery capability used by cursor catch-up. */ +export interface RealtimePublisher { + /** Optional batch preflight used to keep routing violations fail-closed. */ + assertTopics?(topics: readonly string[]): void; + publish(topic: string, payload: string): void; +} + /** * A single entry from drain_changes(), representing a change_log row * matched against subscriber tables. @@ -111,11 +137,34 @@ export interface CursorTrackerOptions { */ export interface RealtimeManagerOptions { /** - * The PgSubscriber instance from PostGraphile's context. - * RealtimeManager emits cursor-tracked events on its internal EventEmitter - * so they flow through existing subscription plans. + * Generation-local publisher used for cursor catch-up delivery. New callers + * should always provide this capability explicitly. */ - pgSubscriber: unknown; + publisher?: RealtimePublisher; + + /** + * Transitional compatibility input for the current @dataplan/pg + * PgSubscriber. Its private emitter is adapted outside RealtimeManager. + * @deprecated Provide publisher instead. + */ + pgSubscriber?: unknown; + + /** + * Exact physical schemas this Graphile instance exposes. Cursor rows naming + * any other source schema stop delivery and surface an error before any row + * in that batch is emitted. Required for the explicit publisher path. + * + * The field remains optional only for the deprecated pgSubscriber adapter, + * whose callers predate generation-scoped routing. + */ + allowedSourceSchemas?: readonly string[]; + + /** + * Called once when delivery can no longer be trusted, after new dispatch is + * disabled and manager shutdown has begun. Callers should synchronously + * remove the owning Graphile generation from service. + */ + onFatalError?: (error: Error) => void; /** * A query-capable object (typically a pg.Pool from pg-cache) used by @@ -160,8 +209,10 @@ export interface RealtimeManagerOptions { batchLimit?: number; /** - * Called when an error occurs during polling, heartbeat, or cleanup. - * If not provided, errors are logged via @pgpmjs/logger. + * Observes polling, heartbeat, or cleanup errors. A polling or heartbeat + * error after startup is independently treated as fatal and delivered to + * onFatalError because cursor recovery can no longer be guaranteed. + * If omitted, the error is logged via @pgpmjs/logger. */ onError?: (error: Error) => void; } diff --git a/packages/perf-harness/README.md b/packages/perf-harness/README.md new file mode 100644 index 0000000000..01fc2f1d47 --- /dev/null +++ b/packages/perf-harness/README.md @@ -0,0 +1,42 @@ +# Graphile performance harness + +Reusable infrastructure for measuring Graphile schema builds in fresh Node +processes. The core accepts any list of serializable benchmark cases; it does not +interpret case names or optimization-specific configuration. + +Each measurement receives a new PID, starts Node with `--expose-gc`, runs a +deterministic GC sequence, records build time and memory metrics, validates a +runtime query, and reports a schema hash. Cases can opt into schema equivalence +groups and provide their own lifecycle validation through the worker result. + +## Extending the harness + +Define a suite and provide a dedicated worker entry point: + +```ts +const suite = { + name: 'example', + cases: [ + { + name: 'baseline', + workerConfig: { schemas: ['cperf_example'] }, + expectedSchemaGroup: 'example-schema', + }, + ], +}; + +await runBenchmarkSuite(suite, options, workerPath); +``` + +`workerConfig` must be JSON-serializable. Logic is implemented in the worker +entry rather than serializing functions across process boundaries. + +The package includes `stock-worker.js` as a minimal upstream Graphile baseline. +The top-level commands require `--database-url`; the runner forwards it and the +opaque case configuration to each short-lived worker as CLI arguments. Database +credentials are redacted from worker failures and JSON reports. This harness is +intended for local development on a trusted machine because command arguments +may be visible to other local processes. + +The PostgreSQL fixture command only creates a previously absent schema whose +name starts with `cperf_`; it never drops or replaces schemas. diff --git a/packages/perf-harness/__tests__/fixture.test.ts b/packages/perf-harness/__tests__/fixture.test.ts new file mode 100644 index 0000000000..08e9109e1e --- /dev/null +++ b/packages/perf-harness/__tests__/fixture.test.ts @@ -0,0 +1,22 @@ +import { + validateFixtureSchema, + validateFixtureTableCount, +} from '../src/fixture'; + +describe('fixture safety', () => { + test('only accepts narrowly scoped benchmark schema names', () => { + expect(validateFixtureSchema('cperf_example_1')).toBe('cperf_example_1'); + expect(() => validateFixtureSchema('public')).toThrow( + 'must start with cperf_' + ); + expect(() => + validateFixtureSchema('cperf_example; drop schema public') + ).toThrow('must start with cperf_'); + }); + + test('bounds generated fixture size', () => { + expect(validateFixtureTableCount(64)).toBe(64); + expect(() => validateFixtureTableCount(0)).toThrow('between 1 and 500'); + expect(() => validateFixtureTableCount(501)).toThrow('between 1 and 500'); + }); +}); diff --git a/packages/perf-harness/__tests__/fixtures/fake-worker.js b/packages/perf-harness/__tests__/fixtures/fake-worker.js new file mode 100644 index 0000000000..5ba1e05b83 --- /dev/null +++ b/packages/perf-harness/__tests__/fixtures/fake-worker.js @@ -0,0 +1,40 @@ +'use strict'; + +const valueFor = (name) => { + const flag = `--${name}`; + const index = process.argv.indexOf(flag); + if (index < 0 || !process.argv[index + 1]) { + throw new Error(`${flag} is required`); + } + return process.argv[index + 1]; +}; + +valueFor('database-url'); +const envelope = JSON.parse( + Buffer.from(valueFor('worker-config'), 'base64url').toString('utf8') +); +const value = envelope.workerConfig.value; +const memory = { + rss: value, + heapTotal: value, + heapUsed: value, + external: value, + arrayBuffers: value, +}; +const result = { + status: 'ok', + pid: process.pid, + caseName: envelope.caseName, + buildMs: value, + schemaHash: envelope.workerConfig.schemaHash, + schemaTypeCount: 10, + runtimeVerified: true, + caseValidation: { passed: true, errors: [] }, + memory: { + baseline: memory, + afterBuild: memory, + delta: memory, + processPeakRss: value, + }, +}; +process.stdout.write(`CPERF_RESULT ${JSON.stringify(result)}\n`); diff --git a/packages/perf-harness/__tests__/process.test.ts b/packages/perf-harness/__tests__/process.test.ts new file mode 100644 index 0000000000..4016c8df51 --- /dev/null +++ b/packages/perf-harness/__tests__/process.test.ts @@ -0,0 +1,71 @@ +import { resolve } from 'node:path'; + +import { parseWorkerProcessArgs, runWorkerProcess } from '../src/process'; + +describe('worker CLI protocol', () => { + const encodedConfig = Buffer.from( + JSON.stringify({ caseName: 'baseline', workerConfig: { value: 1 } }) + ).toString('base64url'); + + test('parses the database URL and worker envelope from CLI arguments', () => { + expect( + parseWorkerProcessArgs([ + '--database-url', + 'postgres:///benchmark', + '--worker-config', + encodedConfig, + ]) + ).toEqual({ + databaseUrl: 'postgres:///benchmark', + envelope: { caseName: 'baseline', workerConfig: { value: 1 } }, + }); + }); + + test('rejects missing, duplicate, and unsupported worker arguments', () => { + expect(() => + parseWorkerProcessArgs(['--worker-config', encodedConfig]) + ).toThrow('--database-url is required'); + expect(() => + parseWorkerProcessArgs(['--database-url', 'postgres:///benchmark']) + ).toThrow('--worker-config is required'); + expect(() => + parseWorkerProcessArgs([ + '--database-url', + 'postgres:///one', + '--database-url', + 'postgres:///two', + '--worker-config', + encodedConfig, + ]) + ).toThrow('--database-url may only be specified once'); + expect(() => + parseWorkerProcessArgs([ + '--database-url', + 'postgres:///benchmark', + '--worker-config', + encodedConfig, + '--unexpected', + 'value', + ]) + ).toThrow("unsupported worker argument '--unexpected'"); + }); +}); + +describe('fresh worker process', () => { + test('uses distinct PIDs and does not expose the database URL', async () => { + const worker = resolve(__dirname, 'fixtures/fake-worker.js'); + const definition = { + name: 'baseline', + workerConfig: { value: 1, schemaHash: 'same' }, + }; + const databaseUrl = 'postgres://secret@example.test/database'; + const first = await runWorkerProcess(worker, databaseUrl, definition); + const second = await runWorkerProcess(worker, databaseUrl, definition); + expect(first.pid).not.toBe(process.pid); + expect(second.pid).not.toBe(process.pid); + expect(first.pid).not.toBe(second.pid); + expect(JSON.stringify([first.result, second.result])).not.toContain( + databaseUrl + ); + }); +}); diff --git a/packages/perf-harness/__tests__/report.test.ts b/packages/perf-harness/__tests__/report.test.ts new file mode 100644 index 0000000000..6d7e043d7a --- /dev/null +++ b/packages/perf-harness/__tests__/report.test.ts @@ -0,0 +1,78 @@ +import { + compareCases, + summarizeCase, + validateSchemaGroups, +} from '../src/report'; +import type { BenchmarkRun, SuccessfulWorkerResult } from '../src/types'; + +const result = (caseName: string, value: number): SuccessfulWorkerResult => ({ + status: 'ok', + pid: value, + caseName, + buildMs: value, + schemaHash: 'same', + schemaTypeCount: 10, + runtimeVerified: true, + caseValidation: { passed: true, errors: [] }, + memory: { + baseline: { + rss: 10, + heapTotal: 10, + heapUsed: 10, + external: 10, + arrayBuffers: 10, + }, + afterBuild: { + rss: value, + heapTotal: value, + heapUsed: value, + external: value, + arrayBuffers: value, + }, + delta: { + rss: value - 10, + heapTotal: value - 10, + heapUsed: value - 10, + external: value - 10, + arrayBuffers: value - 10, + }, + processPeakRss: value, + }, +}); + +describe('generic reports', () => { + test('summarizes arbitrary cases and compares medians', () => { + const runs: BenchmarkRun[] = [10, 30, 20].map((value, index) => ({ + repetition: index + 1, + position: 1, + caseName: 'base', + result: result('base', value), + })); + runs.push( + ...[5, 15, 10].map((value, index) => ({ + repetition: index + 1, + position: 2, + caseName: 'candidate', + result: result('candidate', value), + })) + ); + const base = summarizeCase(runs, 'base')!; + const candidate = summarizeCase(runs, 'candidate')!; + expect( + compareCases('base', 'candidate', base, candidate).buildMs.percentChange + ).toBe(-50); + expect( + validateSchemaGroups( + [ + { name: 'base', workerConfig: null, expectedSchemaGroup: 'schema' }, + { + name: 'candidate', + workerConfig: null, + expectedSchemaGroup: 'schema', + }, + ], + runs + ) + ).toEqual({ equivalent: true, hashes: { schema: 'same' }, errors: [] }); + }); +}); diff --git a/packages/perf-harness/__tests__/run.test.ts b/packages/perf-harness/__tests__/run.test.ts new file mode 100644 index 0000000000..4e8ef65d93 --- /dev/null +++ b/packages/perf-harness/__tests__/run.test.ts @@ -0,0 +1,43 @@ +import { resolve } from 'node:path'; + +import { cliMain, runBenchmarkSuite } from '../src/run'; + +describe('generic suite runner', () => { + test('requires the database URL as an explicit CLI argument', async () => { + await expect( + cliMain(['prepare', '--schema', 'cperf_explicit_cli']) + ).rejects.toThrow('--database-url is required'); + }); + + test('validates fresh processes and schema groups without fixed case names', async () => { + const report = await runBenchmarkSuite( + { + name: 'test-suite', + cases: ['alpha', 'beta', 'gamma'].map((name, index) => ({ + name, + workerConfig: { value: index + 1, schemaHash: 'same' }, + expectedSchemaGroup: 'schema', + })), + }, + { + databaseUrl: 'postgres:///not-used-by-fake-worker', + repetitions: 1, + seed: 1, + order: ['alpha', 'beta', 'gamma'], + }, + resolve(__dirname, 'fixtures/fake-worker.js') + ); + expect(report.validation).toEqual( + expect.objectContaining({ + allRunsSucceeded: true, + freshProcessPerRun: true, + caseValidationPassed: true, + schemaGroupsEquivalent: true, + schemaGroups: { schema: 'same' }, + errors: [], + }) + ); + expect(new Set(report.runs.map((run) => run.result.pid)).size).toBe(3); + expect(JSON.stringify(report)).not.toContain('postgres:///'); + }); +}); diff --git a/packages/perf-harness/__tests__/schedule.test.ts b/packages/perf-harness/__tests__/schedule.test.ts new file mode 100644 index 0000000000..89d5e672f8 --- /dev/null +++ b/packages/perf-harness/__tests__/schedule.test.ts @@ -0,0 +1,29 @@ +import { makeSchedule } from '../src/schedule'; + +const cases = [ + { name: 'a', workerConfig: null }, + { name: 'b', workerConfig: null }, + { name: 'c', workerConfig: null }, +]; + +describe('generic benchmark scheduling', () => { + test('is deterministic and supports any case list', () => { + const first = makeSchedule(cases, 4, 1234); + expect(makeSchedule(cases, 4, 1234)).toEqual(first); + expect(makeSchedule(cases, 4, 4321)).not.toEqual(first); + for (let repetition = 1; repetition <= 4; repetition += 1) { + expect( + first + .filter((item) => item.repetition === repetition) + .map((item) => item.caseName) + .sort() + ).toEqual(['a', 'b', 'c']); + } + }); + + test('accepts an exact order containing each case once', () => { + expect( + makeSchedule(cases, 2, 1, ['c', 'a', 'b']).map((item) => item.caseName) + ).toEqual(['c', 'a', 'b', 'c', 'a', 'b']); + }); +}); diff --git a/packages/perf-harness/jest.config.js b/packages/perf-harness/jest.config.js new file mode 100644 index 0000000000..f34711d4ed --- /dev/null +++ b/packages/perf-harness/jest.config.js @@ -0,0 +1,12 @@ +/** @type {import('ts-jest').JestConfigWithTsJest} */ +module.exports = { + preset: 'ts-jest', + testEnvironment: 'node', + transform: { + '^.+\\.tsx?$': ['ts-jest', { tsconfig: 'tsconfig.json' }], + }, + testRegex: '(/__tests__/.*|(\\.|/)(test|spec))\\.(jsx?|tsx?)$', + moduleFileExtensions: ['ts', 'tsx', 'js', 'jsx', 'json', 'node'], + modulePathIgnorePatterns: ['dist/*'], + testPathIgnorePatterns: ['/__tests__/fixtures/'], +}; diff --git a/packages/perf-harness/package.json b/packages/perf-harness/package.json new file mode 100644 index 0000000000..d975054937 --- /dev/null +++ b/packages/perf-harness/package.json @@ -0,0 +1,36 @@ +{ + "name": "@constructive-io/perf-harness", + "version": "0.1.0", + "private": true, + "description": "Reusable fresh-process Graphile performance harness", + "main": "index.js", + "module": "esm/index.js", + "types": "index.d.ts", + "bin": { + "cperf": "index.js" + }, + "scripts": { + "clean": "makage clean", + "build": "makage build", + "build:dev": "makage build --dev", + "lint": "eslint . --fix", + "test": "jest" + }, + "dependencies": { + "graphile-build": "5.1.1", + "graphile-build-pg": "5.1.3", + "graphile-config": "1.1.0", + "graphql": "16.13.0", + "pg": "^8.21.0", + "postgraphile": "5.1.4" + }, + "devDependencies": { + "@types/node": "^22.19.11", + "@types/pg": "^8.20.4", + "makage": "^0.8.0" + }, + "engines": { + "node": ">=22" + }, + "license": "MIT" +} diff --git a/packages/perf-harness/src/fixture.ts b/packages/perf-harness/src/fixture.ts new file mode 100644 index 0000000000..b7f35935b6 --- /dev/null +++ b/packages/perf-harness/src/fixture.ts @@ -0,0 +1,127 @@ +import { Pool } from 'pg'; + +export const FIXTURE_VERSION = 1; + +export interface PrepareFixtureOptions { + databaseUrl: string; + schema: string; + tables: number; +} + +export interface PreparedFixture { + fixtureVersion: number; + database: string; + serverVersion: string; + schema: string; + tableCount: number; + functionCount: number; +} + +export const validateFixtureSchema = (schema: string): string => { + if ( + !/^cperf_[a-z0-9_]*$/.test(schema) || + schema.length > 63 || + schema.includes('\0') + ) { + throw new Error( + 'fixture schema must start with cperf_, use only lowercase letters, digits, and underscores, and fit PostgreSQL identifiers' + ); + } + return schema; +}; + +export const validateFixtureTableCount = (tables: number): number => { + if (!Number.isSafeInteger(tables) || tables < 1 || tables > 500) { + throw new Error('fixture table count must be an integer between 1 and 500'); + } + return tables; +}; + +const quoteIdentifier = (identifier: string): string => + `"${identifier.replaceAll('"', '""')}"`; + +export const prepareFixture = async ( + options: PrepareFixtureOptions +): Promise => { + const schema = validateFixtureSchema(options.schema); + const tables = validateFixtureTableCount(options.tables); + const quotedSchema = quoteIdentifier(schema); + const pool = new Pool({ connectionString: options.databaseUrl, max: 1 }); + const client = await pool.connect(); + try { + await client.query('begin'); + const existing = await client.query<{ exists: boolean }>( + 'select exists(select 1 from pg_catalog.pg_namespace where nspname = $1) as exists', + [schema] + ); + if (existing.rows[0]?.exists) { + throw new Error( + `fixture schema '${schema}' already exists; this command never replaces schemas` + ); + } + await client.query(`create schema ${quotedSchema}`); + await client.query( + `comment on schema ${quotedSchema} is 'cperf fixture version ${FIXTURE_VERSION}'` + ); + await client.query( + `create type ${quotedSchema}."entity_status" as enum ('draft', 'active', 'archived')` + ); + await client.query(` + create table ${quotedSchema}."account" ( + id bigint generated always as identity primary key, + external_id uuid not null unique, + name text not null, + metadata jsonb not null default '{}'::jsonb, + created_at timestamptz not null default now() + ) + `); + for (let index = 1; index <= tables; index += 1) { + const table = quoteIdentifier(`entity_${index}`); + const functionName = quoteIdentifier(`entity_${index}_by_account`); + const indexName = quoteIdentifier(`entity_${index}_account_created_idx`); + await client.query(` + create table ${quotedSchema}.${table} ( + id bigint generated always as identity primary key, + account_id bigint not null references ${quotedSchema}."account"(id), + status ${quotedSchema}."entity_status" not null default 'draft', + title text not null, + tags text[] not null default array[]::text[], + metadata jsonb not null default '{}'::jsonb, + created_at timestamptz not null default now(), + unique (account_id, title) + ); + create index ${indexName} + on ${quotedSchema}.${table} (account_id, created_at desc); + create function ${quotedSchema}.${functionName}(requested_account_id bigint) + returns setof ${quotedSchema}.${table} + language sql stable + as 'select * from ${quotedSchema}.${table} where account_id = requested_account_id'; + `); + } + const identity = await client.query<{ + database: string; + server_version: string; + }>( + "select current_database() as database, current_setting('server_version') as server_version" + ); + await client.query('commit'); + return { + fixtureVersion: FIXTURE_VERSION, + database: identity.rows[0].database, + serverVersion: identity.rows[0].server_version, + schema, + tableCount: tables + 1, + functionCount: tables, + }; + } catch (error) { + try { + await client.query('rollback'); + } catch { + // Preserve the fixture preparation error; the client is discarded below. + } + throw error; + } finally { + client.release(); + await pool.end(); + } +}; diff --git a/packages/perf-harness/src/index.ts b/packages/perf-harness/src/index.ts new file mode 100644 index 0000000000..7829d7dc66 --- /dev/null +++ b/packages/perf-harness/src/index.ts @@ -0,0 +1,20 @@ +#!/usr/bin/env node + +export * from './fixture'; +export * from './metrics'; +export * from './process'; +export * from './report'; +export * from './run'; +export * from './schedule'; +export * from './types'; + +import { cliMain } from './run'; + +if (require.main === module) { + void cliMain().catch((error: unknown) => { + process.stderr.write( + `${error instanceof Error ? error.message : String(error)}\n` + ); + process.exitCode = 1; + }); +} diff --git a/packages/perf-harness/src/metrics.ts b/packages/perf-harness/src/metrics.ts new file mode 100644 index 0000000000..007e79fa13 --- /dev/null +++ b/packages/perf-harness/src/metrics.ts @@ -0,0 +1,79 @@ +import { performance } from 'node:perf_hooks'; + +import type { + CaseValidation, + JsonValue, + MemorySnapshot, + SuccessfulWorkerResult, +} from './types'; + +export interface MeasuredCaseResult { + schemaHash: string; + schemaTypeCount: number; + runtimeVerified: true; + caseValidation?: CaseValidation; + metadata?: Record; +} + +const collectGarbage = (): void => { + if (typeof global.gc !== 'function') { + throw new Error('benchmark worker requires Node --expose-gc'); + } + global.gc(); + global.gc(); + global.gc(); +}; + +const memorySnapshot = (): MemorySnapshot => { + const memory = process.memoryUsage(); + return { + rss: memory.rss, + heapTotal: memory.heapTotal, + heapUsed: memory.heapUsed, + external: memory.external, + arrayBuffers: memory.arrayBuffers, + }; +}; + +const memoryDelta = ( + baseline: MemorySnapshot, + afterBuild: MemorySnapshot +): MemorySnapshot => ({ + rss: afterBuild.rss - baseline.rss, + heapTotal: afterBuild.heapTotal - baseline.heapTotal, + heapUsed: afterBuild.heapUsed - baseline.heapUsed, + external: afterBuild.external - baseline.external, + arrayBuffers: afterBuild.arrayBuffers - baseline.arrayBuffers, +}); + +export const measureBenchmarkCase = async ( + caseName: string, + build: () => Promise, + validate: (built: Built) => Promise +): Promise => { + collectGarbage(); + const baseline = memorySnapshot(); + const startedAt = performance.now(); + const built = await build(); + const buildMs = performance.now() - startedAt; + const measured = await validate(built); + collectGarbage(); + const afterBuild = memorySnapshot(); + return { + status: 'ok', + pid: process.pid, + caseName, + buildMs, + schemaHash: measured.schemaHash, + schemaTypeCount: measured.schemaTypeCount, + runtimeVerified: measured.runtimeVerified, + caseValidation: measured.caseValidation ?? { passed: true, errors: [] }, + ...(measured.metadata ? { metadata: measured.metadata } : {}), + memory: { + baseline, + afterBuild, + delta: memoryDelta(baseline, afterBuild), + processPeakRss: process.resourceUsage().maxRSS * 1024, + }, + }; +}; diff --git a/packages/perf-harness/src/process.ts b/packages/perf-harness/src/process.ts new file mode 100644 index 0000000000..dee9097204 --- /dev/null +++ b/packages/perf-harness/src/process.ts @@ -0,0 +1,183 @@ +import { spawn } from 'node:child_process'; + +import type { + BenchmarkCaseDefinition, + WorkerConfigEnvelope, + WorkerResult, +} from './types'; + +export const WORKER_RESULT_PREFIX = 'CPERF_RESULT '; +export const DATABASE_URL_ARGUMENT = 'database-url'; +export const WORKER_CONFIG_ARGUMENT = 'worker-config'; + +export interface ParsedValueArgs { + values: Map; +} + +export interface WorkerProcessArgs { + databaseUrl: string; + envelope: WorkerConfigEnvelope; +} + +export interface SpawnedWorkerResult { + pid: number; + result: WorkerResult; +} + +const lastLines = (value: string, count = 20): string => + value.trim().split('\n').slice(-count).join('\n'); + +export const redactSecret = (value: string, secret: string): string => + secret ? value.replaceAll(secret, '') : value; + +export const parseValueArgs = (args: readonly string[]): ParsedValueArgs => { + const values = new Map(); + for (let index = 0; index < args.length; index += 2) { + const flag = args[index]; + const value = args[index + 1]; + if ( + !flag?.startsWith('--') || + value === undefined || + value.startsWith('--') + ) { + throw new Error(`expected --name value near '${flag ?? ''}'`); + } + const name = flag.slice(2); + if (values.has(name)) + throw new Error(`--${name} may only be specified once`); + values.set(name, value); + } + return { values }; +}; + +export const parseWorkerProcessArgs = ( + args: readonly string[] +): WorkerProcessArgs => { + const parsed = parseValueArgs(args); + for (const name of parsed.values.keys()) { + if (name !== DATABASE_URL_ARGUMENT && name !== WORKER_CONFIG_ARGUMENT) { + throw new Error(`unsupported worker argument '--${name}'`); + } + } + const databaseUrl = parsed.values.get(DATABASE_URL_ARGUMENT); + if (!databaseUrl) throw new Error('--database-url is required'); + return { + databaseUrl, + envelope: parseWorkerEnvelope(parsed.values.get(WORKER_CONFIG_ARGUMENT)), + }; +}; + +export const runWorkerProcess = ( + workerPath: string, + databaseUrl: string, + definition: BenchmarkCaseDefinition +): Promise => + new Promise((resolve, reject) => { + const config: WorkerConfigEnvelope = { + caseName: definition.name, + workerConfig: definition.workerConfig, + }; + const child = spawn( + process.execPath, + [ + '--expose-gc', + workerPath, + `--${DATABASE_URL_ARGUMENT}`, + databaseUrl, + `--${WORKER_CONFIG_ARGUMENT}`, + Buffer.from(JSON.stringify(config)).toString('base64url'), + ], + { + env: { + ...process.env, + NODE_ENV: 'production', + GRAPHILE_ENV: 'production', + }, + stdio: ['ignore', 'pipe', 'pipe'], + } + ); + const pid = child.pid; + let stdout = ''; + let stderr = ''; + child.stdout.setEncoding('utf8'); + child.stderr.setEncoding('utf8'); + child.stdout.on('data', (chunk: string) => { + stdout += chunk; + }); + child.stderr.on('data', (chunk: string) => { + stderr += chunk; + }); + child.once('error', reject); + child.once('close', (code, signal) => { + const resultLine = stdout + .split('\n') + .reverse() + .find((line) => line.startsWith(WORKER_RESULT_PREFIX)); + if (!resultLine) { + reject( + new Error( + redactSecret( + `benchmark worker ${pid ?? 'unknown'} exited without a result ` + + `(code=${String(code)}, signal=${String(signal)})` + + (stderr.trim() ? `\n${lastLines(stderr)}` : ''), + databaseUrl + ) + ) + ); + return; + } + try { + const result = JSON.parse( + resultLine.slice(WORKER_RESULT_PREFIX.length) + ) as WorkerResult; + if (typeof pid !== 'number' || result.pid !== pid) { + throw new Error( + `worker PID mismatch: spawned ${String(pid)}, reported ${String( + result.pid + )}` + ); + } + if (result.caseName !== definition.name) { + throw new Error( + `worker case mismatch: expected ${definition.name}, reported ${result.caseName}` + ); + } + if (result.status === 'ok' && code !== 0) { + throw new Error(`successful worker exited with code ${String(code)}`); + } + resolve({ pid, result }); + } catch (error) { + reject( + new Error( + redactSecret( + `invalid result from benchmark worker ${String(pid)}: ${String( + error instanceof Error ? error.message : error + )}`, + databaseUrl + ) + ) + ); + } + }); + }); + +export const parseWorkerEnvelope = ( + encoded: string | undefined +): WorkerConfigEnvelope => { + if (!encoded) throw new Error('--worker-config is required'); + const parsed = JSON.parse( + Buffer.from(encoded, 'base64url').toString('utf8') + ) as Partial; + if ( + typeof parsed.caseName !== 'string' || + parsed.caseName.length === 0 || + parsed.workerConfig === undefined + ) { + throw new Error('worker configuration envelope is invalid'); + } + return parsed as WorkerConfigEnvelope; +}; + +export const writeWorkerResult = (result: WorkerResult): void => { + process.stdout.write(`${WORKER_RESULT_PREFIX}${JSON.stringify(result)}\n`); +}; diff --git a/packages/perf-harness/src/report.ts b/packages/perf-harness/src/report.ts new file mode 100644 index 0000000000..680852a961 --- /dev/null +++ b/packages/perf-harness/src/report.ts @@ -0,0 +1,123 @@ +import type { + BenchmarkCaseDefinition, + BenchmarkRun, + CaseComparison, + CaseSummary, + MetricComparison, + MetricSummary, + SuccessfulWorkerResult, +} from './types'; + +const median = (values: readonly number[]): number => { + if (values.length === 0) throw new Error('cannot summarize zero values'); + const sorted = [...values].sort((left, right) => left - right); + const middle = Math.floor(sorted.length / 2); + return sorted.length % 2 === 0 + ? (sorted[middle - 1] + sorted[middle]) / 2 + : sorted[middle]; +}; + +const metricSummary = (values: number[]): MetricSummary => ({ + median: median(values), + min: Math.min(...values), + max: Math.max(...values), + samples: values, +}); + +const successfulResultsFor = ( + runs: readonly BenchmarkRun[], + caseName: string +): SuccessfulWorkerResult[] => + runs + .filter((run) => run.caseName === caseName && run.result.status === 'ok') + .map((run) => run.result as SuccessfulWorkerResult); + +export const summarizeCase = ( + runs: readonly BenchmarkRun[], + caseName: string +): CaseSummary | undefined => { + const results = successfulResultsFor(runs, caseName); + if (results.length === 0) return undefined; + return { + sampleCount: results.length, + buildMs: metricSummary(results.map((result) => result.buildMs)), + heapUsedAfterBuild: metricSummary( + results.map((result) => result.memory.afterBuild.heapUsed) + ), + heapUsedDelta: metricSummary( + results.map((result) => result.memory.delta.heapUsed) + ), + rssAfterBuild: metricSummary( + results.map((result) => result.memory.afterBuild.rss) + ), + rssDelta: metricSummary(results.map((result) => result.memory.delta.rss)), + processPeakRss: metricSummary( + results.map((result) => result.memory.processPeakRss) + ), + }; +}; + +const compareMetric = ( + baseline: MetricSummary, + candidate: MetricSummary +): MetricComparison => ({ + baseline: baseline.median, + candidate: candidate.median, + difference: candidate.median - baseline.median, + percentChange: + baseline.median === 0 + ? null + : ((candidate.median - baseline.median) / Math.abs(baseline.median)) * + 100, +}); + +export const compareCases = ( + baselineCase: string, + candidateCase: string, + baseline: CaseSummary, + candidate: CaseSummary +): CaseComparison => ({ + baselineCase, + candidateCase, + buildMs: compareMetric(baseline.buildMs, candidate.buildMs), + heapUsedAfterBuild: compareMetric( + baseline.heapUsedAfterBuild, + candidate.heapUsedAfterBuild + ), + heapUsedDelta: compareMetric(baseline.heapUsedDelta, candidate.heapUsedDelta), + rssAfterBuild: compareMetric(baseline.rssAfterBuild, candidate.rssAfterBuild), + rssDelta: compareMetric(baseline.rssDelta, candidate.rssDelta), + processPeakRss: compareMetric( + baseline.processPeakRss, + candidate.processPeakRss + ), +}); + +export const validateSchemaGroups = ( + definitions: readonly BenchmarkCaseDefinition[], + runs: readonly BenchmarkRun[] +): { + equivalent: boolean; + hashes: Record; + errors: string[]; +} => { + const groups = new Map>(); + for (const definition of definitions) { + if (!definition.expectedSchemaGroup) continue; + const hashes = + groups.get(definition.expectedSchemaGroup) ?? new Set(); + for (const result of successfulResultsFor(runs, definition.name)) { + hashes.add(result.schemaHash); + } + groups.set(definition.expectedSchemaGroup, hashes); + } + const output: Record = {}; + const errors: string[] = []; + for (const [group, hashes] of groups) { + output[group] = hashes.size === 1 ? [...hashes][0] : null; + if (hashes.size !== 1) { + errors.push(`schema group '${group}' did not produce one schema hash`); + } + } + return { equivalent: errors.length === 0, hashes: output, errors }; +}; diff --git a/packages/perf-harness/src/run.ts b/packages/perf-harness/src/run.ts new file mode 100644 index 0000000000..ea6cbaed89 --- /dev/null +++ b/packages/perf-harness/src/run.ts @@ -0,0 +1,245 @@ +import { mkdir, rename, writeFile } from 'node:fs/promises'; +import { dirname, resolve } from 'node:path'; + +import { prepareFixture } from './fixture'; +import { parseValueArgs, redactSecret, runWorkerProcess } from './process'; +import { summarizeCase, validateSchemaGroups } from './report'; +import { makeSchedule, validateCaseDefinitions } from './schedule'; +import type { + BenchmarkCaseDefinition, + BenchmarkReport, + BenchmarkRun, + BenchmarkSuiteDefinition, +} from './types'; + +export interface RunSuiteOptions { + databaseUrl: string; + repetitions: number; + seed: number; + order: string[] | null; + output?: string; +} + +export const runBenchmarkSuite = async ( + suite: BenchmarkSuiteDefinition, + options: RunSuiteOptions, + workerPath: string +): Promise => { + validateCaseDefinitions(suite.cases); + const byName = new Map( + suite.cases.map((definition) => [definition.name, definition]) + ); + const schedule = makeSchedule( + suite.cases, + options.repetitions, + options.seed, + options.order + ); + const runs: BenchmarkRun[] = []; + for (const coordinate of schedule) { + const definition = byName.get(coordinate.caseName)!; + process.stderr.write( + `[${runs.length + 1}/${schedule.length}] repetition ${ + coordinate.repetition + }, ${coordinate.caseName}\n` + ); + try { + const spawned = await runWorkerProcess( + workerPath, + options.databaseUrl, + definition + ); + runs.push({ ...coordinate, result: spawned.result }); + } catch (error) { + runs.push({ + ...coordinate, + result: { + status: 'error', + pid: -1, + caseName: coordinate.caseName, + error: redactSecret( + error instanceof Error ? error.message : String(error), + options.databaseUrl + ), + }, + }); + } + } + const successfulRuns = runs.filter((run) => run.result.status === 'ok'); + const allRunsSucceeded = successfulRuns.length === schedule.length; + const pids = successfulRuns.map((run) => run.result.pid); + const freshProcessPerRun = + allRunsSucceeded && + pids.every((pid) => pid > 0 && pid !== process.pid) && + new Set(pids).size === pids.length; + const caseValidationPassed = + allRunsSucceeded && + successfulRuns.every( + (run) => run.result.status === 'ok' && run.result.caseValidation.passed + ); + const schemaGroups = validateSchemaGroups(suite.cases, runs); + const errors: string[] = []; + for (const run of runs) { + if (run.result.status === 'error') { + errors.push( + `${run.caseName} repetition ${run.repetition}: ${run.result.error}` + ); + } else { + for (const error of run.result.caseValidation.errors) { + errors.push(`${run.caseName} repetition ${run.repetition}: ${error}`); + } + } + } + if (!freshProcessPerRun) { + errors.push('fresh-process validation did not pass for every run'); + } + errors.push(...schemaGroups.errors); + const summaries: Record< + string, + NonNullable> + > = {}; + for (const definition of suite.cases) { + const summary = summarizeCase(runs, definition.name); + if (summary) summaries[definition.name] = summary; + } + return { + format: 'constructive-performance-suite/v1', + generatedAt: new Date().toISOString(), + node: process.version, + platform: process.platform, + architecture: process.arch, + suite, + config: { + repetitions: options.repetitions, + seed: options.seed, + order: options.order, + }, + schedule, + runs, + validation: { + allRunsSucceeded, + freshProcessPerRun, + caseValidationPassed, + schemaGroupsEquivalent: schemaGroups.equivalent, + schemaGroups: schemaGroups.hashes, + errors, + }, + summaries, + }; +}; + +export const writeJsonAtomically = async ( + output: string, + value: unknown +): Promise => { + const absoluteOutput = resolve(output); + await mkdir(dirname(absoluteOutput), { recursive: true }); + const temporary = `${absoluteOutput}.tmp-${process.pid}`; + await writeFile(temporary, `${JSON.stringify(value, null, 2)}\n`, { + encoding: 'utf8', + mode: 0o600, + }); + await rename(temporary, absoluteOutput); + return absoluteOutput; +}; + +const positiveInteger = ( + value: string | undefined, + name: string, + defaultValue: number, + maximum: number +): number => { + if (value === undefined) return defaultValue; + if (!/^\d+$/.test(value)) throw new Error(`--${name} must be an integer`); + const parsed = Number(value); + if (!Number.isSafeInteger(parsed) || parsed < 1 || parsed > maximum) { + throw new Error(`--${name} must be between 1 and ${maximum}`); + } + return parsed; +}; + +const databaseUrl = (args: ReturnType): string => { + const value = args.values.get('database-url'); + if (!value) throw new Error('--database-url is required'); + return value; +}; + +const stringList = (value: string | undefined): string[] | null => { + if (value === undefined) return null; + const result = value.split(','); + if ( + result.some((item) => item.length === 0 || item.trim() !== item) || + new Set(result).size !== result.length + ) { + throw new Error('list values must be unique exact non-empty strings'); + } + return result; +}; + +const parseCases = (encoded: string): BenchmarkCaseDefinition[] => { + const parsed = JSON.parse(Buffer.from(encoded, 'base64url').toString('utf8')); + if (!Array.isArray(parsed)) + throw new Error('--cases must encode a JSON array'); + return parsed as BenchmarkCaseDefinition[]; +}; + +export const cliMain = async (args = process.argv.slice(2)): Promise => { + const [command, ...rest] = args; + const parsed = parseValueArgs(rest); + if (command === 'prepare') { + const schema = parsed.values.get('schema'); + if (!schema) throw new Error('--schema is required'); + const result = await prepareFixture({ + databaseUrl: databaseUrl(parsed), + schema, + tables: positiveInteger(parsed.values.get('tables'), 'tables', 64, 500), + }); + process.stdout.write(`${JSON.stringify(result)}\n`); + return; + } + if (command !== 'run') { + throw new Error('expected prepare or run command'); + } + const cases = parsed.values.get('cases'); + const worker = parsed.values.get('worker'); + if (!cases || !worker) throw new Error('--cases and --worker are required'); + const report = await runBenchmarkSuite( + { + name: parsed.values.get('suite') ?? 'benchmark-suite', + cases: parseCases(cases), + }, + { + databaseUrl: databaseUrl(parsed), + repetitions: positiveInteger( + parsed.values.get('repetitions'), + 'repetitions', + 3, + 50 + ), + seed: positiveInteger( + parsed.values.get('seed'), + 'seed', + 20260813, + 0xffffffff + ), + order: stringList(parsed.values.get('order')), + output: parsed.values.get('output'), + }, + resolve(worker) + ); + const output = await writeJsonAtomically( + parsed.values.get('output') ?? 'performance-report.json', + report + ); + process.stdout.write( + `${JSON.stringify({ output, validation: report.validation })}\n` + ); + if ( + !report.validation.allRunsSucceeded || + !report.validation.freshProcessPerRun || + !report.validation.caseValidationPassed || + !report.validation.schemaGroupsEquivalent + ) { + process.exitCode = 1; + } +}; diff --git a/packages/perf-harness/src/schedule.ts b/packages/perf-harness/src/schedule.ts new file mode 100644 index 0000000000..8eaa98fb86 --- /dev/null +++ b/packages/perf-harness/src/schedule.ts @@ -0,0 +1,77 @@ +import type { BenchmarkCaseDefinition, BenchmarkCoordinate } from './types'; + +const seededRandom = (seed: number): (() => number) => { + let state = seed >>> 0; + return () => { + state += 0x6d2b79f5; + let value = state; + value = Math.imul(value ^ (value >>> 15), value | 1); + value ^= value + Math.imul(value ^ (value >>> 7), value | 61); + return ((value ^ (value >>> 14)) >>> 0) / 4294967296; + }; +}; + +const shuffledCaseNames = ( + definitions: readonly BenchmarkCaseDefinition[], + seed: number, + repetition: number +): string[] => { + const result = definitions.map(({ name }) => name); + const random = seededRandom((seed ^ Math.imul(repetition, 0x9e3779b1)) >>> 0); + for (let index = result.length - 1; index > 0; index -= 1) { + const swapIndex = Math.floor(random() * (index + 1)); + [result[index], result[swapIndex]] = [result[swapIndex], result[index]]; + } + return result; +}; + +export const validateCaseDefinitions = ( + definitions: readonly BenchmarkCaseDefinition[] +): void => { + if (definitions.length === 0) { + throw new Error('benchmark suite must contain at least one case'); + } + const names = definitions.map(({ name }) => name); + if ( + names.some( + (name) => name.length === 0 || name.trim() !== name || name.includes('\0') + ) || + new Set(names).size !== names.length + ) { + throw new Error( + 'benchmark case names must be unique exact non-empty strings' + ); + } +}; + +export const makeSchedule = ( + definitions: readonly BenchmarkCaseDefinition[], + repetitions: number, + seed: number, + exactOrder: readonly string[] | null = null +): BenchmarkCoordinate[] => { + validateCaseDefinitions(definitions); + if (!Number.isSafeInteger(repetitions) || repetitions < 1) { + throw new Error('repetitions must be a positive safe integer'); + } + const expectedNames = definitions.map(({ name }) => name).sort(); + const schedule: BenchmarkCoordinate[] = []; + for (let repetition = 1; repetition <= repetitions; repetition += 1) { + const order = exactOrder + ? [...exactOrder] + : shuffledCaseNames(definitions, seed, repetition); + if ( + order.length !== definitions.length || + new Set(order).size !== definitions.length || + [...order].sort().some((name, index) => name !== expectedNames[index]) + ) { + throw new Error( + 'exact order must contain each benchmark case exactly once' + ); + } + order.forEach((caseName, index) => { + schedule.push({ repetition, position: index + 1, caseName }); + }); + } + return schedule; +}; diff --git a/packages/perf-harness/src/stock-worker.ts b/packages/perf-harness/src/stock-worker.ts new file mode 100644 index 0000000000..c68cc18f0d --- /dev/null +++ b/packages/perf-harness/src/stock-worker.ts @@ -0,0 +1,93 @@ +import { createHash } from 'node:crypto'; + +import { + defaultPreset as graphileBuildPreset, + makeSchema, +} from 'graphile-build'; +import { defaultPreset as graphileBuildPgPreset } from 'graphile-build-pg'; +import { execute, lexicographicSortSchema, parse, printSchema } from 'graphql'; +import { makePgService } from 'postgraphile/adaptors/pg'; + +import { measureBenchmarkCase } from './metrics'; +import { + parseWorkerProcessArgs, + redactSecret, + writeWorkerResult, +} from './process'; + +interface StockConfig { + schemas: string[]; +} + +const validateConfig = (value: unknown): StockConfig => { + const schemas = (value as Partial)?.schemas; + if ( + !Array.isArray(schemas) || + schemas.length === 0 || + schemas.some((schema) => typeof schema !== 'string' || schema.length === 0) + ) { + throw new Error('stock worker requires a non-empty schemas array'); + } + return { schemas }; +}; + +const main = async (): Promise => { + let databaseUrl = ''; + let caseName = 'unknown'; + try { + const workerArgs = parseWorkerProcessArgs(process.argv.slice(2)); + databaseUrl = workerArgs.databaseUrl; + const { envelope } = workerArgs; + caseName = envelope.caseName; + const config = validateConfig(envelope.workerConfig); + const service = makePgService({ + connectionString: databaseUrl, + schemas: config.schemas, + pubsub: false, + }); + try { + const result = await measureBenchmarkCase( + caseName, + async () => + makeSchema({ + extends: [graphileBuildPreset, graphileBuildPgPreset], + pgServices: [service], + }), + async ({ schema }) => { + const execution = await execute({ + schema, + document: parse('{ __typename }'), + }); + if ( + execution.errors?.length || + execution.data?.__typename !== 'Query' + ) { + throw new Error('runtime verification query failed'); + } + const schemaText = printSchema(lexicographicSortSchema(schema)); + return { + schemaHash: createHash('sha256').update(schemaText).digest('hex'), + schemaTypeCount: Object.keys(schema.getTypeMap()).length, + runtimeVerified: true as const, + }; + } + ); + writeWorkerResult(result); + } finally { + await service.release(); + } + } catch (error) { + writeWorkerResult({ + status: 'error', + pid: process.pid, + caseName, + error: redactSecret( + error instanceof Error ? error.message : String(error), + databaseUrl + ), + }); + process.exitCode = 1; + } +}; + +if (require.main === module) void main(); diff --git a/packages/perf-harness/src/types.ts b/packages/perf-harness/src/types.ts new file mode 100644 index 0000000000..3713a4edc4 --- /dev/null +++ b/packages/perf-harness/src/types.ts @@ -0,0 +1,129 @@ +export type JsonPrimitive = string | number | boolean | null; +export type JsonValue = + JsonPrimitive | JsonValue[] | { [key: string]: JsonValue }; + +export interface BenchmarkCaseDefinition { + name: string; + workerConfig: JsonValue; + expectedSchemaGroup?: string; +} + +export interface BenchmarkSuiteDefinition { + name: string; + cases: BenchmarkCaseDefinition[]; +} + +export interface BenchmarkCoordinate { + repetition: number; + position: number; + caseName: string; +} + +export interface WorkerConfigEnvelope { + caseName: string; + workerConfig: JsonValue; +} + +export interface MemorySnapshot { + rss: number; + heapTotal: number; + heapUsed: number; + external: number; + arrayBuffers: number; +} + +export interface CaseValidation { + passed: boolean; + errors: string[]; +} + +export interface SuccessfulWorkerResult { + status: 'ok'; + pid: number; + caseName: string; + buildMs: number; + schemaHash: string; + schemaTypeCount: number; + runtimeVerified: true; + caseValidation: CaseValidation; + metadata?: Record; + memory: { + baseline: MemorySnapshot; + afterBuild: MemorySnapshot; + delta: MemorySnapshot; + processPeakRss: number; + }; +} + +export interface FailedWorkerResult { + status: 'error'; + pid: number; + caseName: string; + error: string; +} + +export type WorkerResult = SuccessfulWorkerResult | FailedWorkerResult; + +export interface BenchmarkRun extends BenchmarkCoordinate { + result: WorkerResult; +} + +export interface MetricSummary { + median: number; + min: number; + max: number; + samples: number[]; +} + +export interface CaseSummary { + sampleCount: number; + buildMs: MetricSummary; + heapUsedAfterBuild: MetricSummary; + heapUsedDelta: MetricSummary; + rssAfterBuild: MetricSummary; + rssDelta: MetricSummary; + processPeakRss: MetricSummary; +} + +export interface MetricComparison { + baseline: number; + candidate: number; + difference: number; + percentChange: number | null; +} + +export interface CaseComparison { + baselineCase: string; + candidateCase: string; + buildMs: MetricComparison; + heapUsedAfterBuild: MetricComparison; + heapUsedDelta: MetricComparison; + rssAfterBuild: MetricComparison; + rssDelta: MetricComparison; + processPeakRss: MetricComparison; +} + +export interface BenchmarkReport { + format: 'constructive-performance-suite/v1'; + generatedAt: string; + node: string; + platform: string; + architecture: string; + suite: BenchmarkSuiteDefinition; + config: { + repetitions: number; + seed: number; + order: string[] | null; + }; + schedule: BenchmarkCoordinate[]; + runs: BenchmarkRun[]; + validation: { + allRunsSucceeded: boolean; + freshProcessPerRun: boolean; + caseValidationPassed: boolean; + schemaGroupsEquivalent: boolean; + schemaGroups: Record; + errors: string[]; + }; + summaries: Record; +} diff --git a/packages/perf-harness/tsconfig.esm.json b/packages/perf-harness/tsconfig.esm.json new file mode 100644 index 0000000000..1a3c9914f1 --- /dev/null +++ b/packages/perf-harness/tsconfig.esm.json @@ -0,0 +1,8 @@ +{ + "extends": "./tsconfig.json", + "compilerOptions": { + "outDir": "dist/esm", + "module": "esnext", + "moduleResolution": "bundler" + } +} diff --git a/packages/perf-harness/tsconfig.json b/packages/perf-harness/tsconfig.json new file mode 100644 index 0000000000..319daa4b0d --- /dev/null +++ b/packages/perf-harness/tsconfig.json @@ -0,0 +1,11 @@ +{ + "extends": "../../tsconfig.json", + "compilerOptions": { + "outDir": "dist", + "rootDir": "src", + "moduleResolution": "nodenext", + "module": "nodenext", + "isolatedModules": true + }, + "include": ["src/**/*"] +} diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 9f681e75ea..ca6ae59d08 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -2711,6 +2711,37 @@ importers: version: 0.8.0 publishDirectory: dist + packages/perf-harness: + dependencies: + graphile-build: + specifier: 5.1.1 + version: 5.1.1(grafast@1.1.2(graphql@16.13.0))(graphile-config@1.1.0)(graphql@16.13.0) + graphile-build-pg: + specifier: 5.1.3 + version: 5.1.3(@dataplan/pg@1.1.1(@dataplan/json@1.0.1(grafast@1.1.2(graphql@16.13.0)))(grafast@1.1.2(graphql@16.13.0))(graphile-config@1.1.0)(graphql@16.13.0)(pg-sql2@5.0.1)(pg@8.21.0))(grafast@1.1.2(graphql@16.13.0))(graphile-build@5.1.1(grafast@1.1.2(graphql@16.13.0))(graphile-config@1.1.0)(graphql@16.13.0))(graphile-config@1.1.0)(graphql@16.13.0)(pg-sql2@5.0.1)(pg@8.21.0)(tamedevil@0.1.1) + graphile-config: + specifier: 1.1.0 + version: 1.1.0 + graphql: + specifier: 16.13.0 + version: 16.13.0 + pg: + specifier: ^8.21.0 + version: 8.21.0 + postgraphile: + specifier: 5.1.4 + version: 5.1.4(f282a162d8bd20a217e08c60f5396af8) + devDependencies: + '@types/node': + specifier: ^22.19.11 + version: 22.19.19 + '@types/pg': + specifier: ^8.20.4 + version: 8.20.4 + makage: + specifier: ^0.8.0 + version: 0.8.0 + packages/postmaster: dependencies: 12factor-env: