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; }