From ccaff715dd6dde29977ad2d04e734c2bd2addb04 Mon Sep 17 00:00:00 2001 From: Pranay Prakash Date: Fri, 9 Oct 2026 12:58:23 -0700 Subject: [PATCH 1/2] [world-postgres] Drive LISTEN reconnect retries in the invoke test the way consumers do The "reconnects a terminated real LISTEN client" test made exactly one reconnect attempt: invocation notifications reconnect only from watch() and wait() once the 1s backoff has passed, and the test called wait() twice and then polled pg_stat_activity for 15s. If that one attempt failed (1s connect and LISTEN timeouts, so a loaded runner is enough), nothing retried it and the poll timed out. Production consumers in invocations.ts read and then wait(revision, INVOCATION_FALLBACK_MS) in a loop, so they retry on the next wait; the reconnect behaviour itself is correct. Poll the way consumers do, waiting between reads, and make the listener's first reconnect fail on purpose so the test always covers a retried attempt. The initial subscription gets the same loop: a failed first connect also satisfied the old revision > 0 check and stranded the pid poll. Readiness now means the backend has run LISTEN (pg_stat_activity reports a backend as idle before its LISTEN runs), and the restored transition is awaited before taking the notification baseline, so the final notification check cannot be satisfied by the reconnect's own wake. Co-Authored-By: Claude Opus 5.5 (1M context) --- .../world-postgres-listen-reconnect-test.md | 4 ++ packages/world-postgres/test/invoke.test.ts | 69 ++++++++++++++----- 2 files changed, 56 insertions(+), 17 deletions(-) create mode 100644 .changeset/world-postgres-listen-reconnect-test.md diff --git a/.changeset/world-postgres-listen-reconnect-test.md b/.changeset/world-postgres-listen-reconnect-test.md new file mode 100644 index 0000000000..84ab68539a --- /dev/null +++ b/.changeset/world-postgres-listen-reconnect-test.md @@ -0,0 +1,4 @@ +--- +--- + +Make the world-postgres LISTEN reconnect test retry its reconnect the way consumers do. diff --git a/packages/world-postgres/test/invoke.test.ts b/packages/world-postgres/test/invoke.test.ts index 07e2b75117..0e363ab799 100644 --- a/packages/world-postgres/test/invoke.test.ts +++ b/packages/world-postgres/test/invoke.test.ts @@ -15,7 +15,7 @@ import { encode } from 'cbor-x'; import { drizzle } from 'drizzle-orm/node-postgres'; import { migrate } from 'drizzle-orm/node-postgres/migrator'; import { makeWorkerUtils } from 'graphile-worker'; -import { Pool } from 'pg'; +import { Client, Pool } from 'pg'; import { ulid } from 'ulid'; import { afterAll, beforeAll, describe, expect, it, vi } from 'vitest'; import { registerStepFunction } from '../../core/dist/private.js'; @@ -34,6 +34,7 @@ import { import { createWorld } from '../src/index.js'; import { createInvocationNotifications, + INVOCATION_FALLBACK_MS, INVOCATION_INPUT_TOPIC, INVOCATION_RESULT_TOPIC, invocationNotificationKey, @@ -1147,44 +1148,76 @@ describe.skipIf(process.platform === 'win32')( } }); - it('reconnects a terminated real LISTEN client and receives later notifications', async () => { + it('reconnects a terminated real LISTEN client, retrying a failed attempt, and receives later notifications', async () => { const name = `notify-${randomUUID()}`; const listenerPool = new Pool({ connectionString: container.getConnectionUri(), application_name: name, }); + // The listener's first reconnect fails, as a connect timeout under load + // does. Nothing retries it but a later wait(). + const connect = Client.prototype.connect; + let listenerConnects = 0; + const connectSpy = vi + .spyOn(Client.prototype, 'connect') + .mockImplementation(function (this: Client, ...args: never[]) { + const { connectionParameters } = this as unknown as { + connectionParameters: { application_name?: string }; + }; + if ( + connectionParameters.application_name === `${name}:invocations` && + ++listenerConnects === 2 + ) { + return Promise.reject( + new Error('Connection terminated due to connection timeout') + ); + } + return Reflect.apply(connect, this, args); + } as typeof connect); + const warn = vi.spyOn(console, 'warn'); + // Logged right before the reconnected listener wakes its watches. + const restorations = () => + warn.mock.calls.filter(([message]) => + String(message).includes('Invocation notifications restored') + ).length; const observer = createInvocationNotifications(listenerPool); const key = invocationNotificationKey('reconnect'); const watch = observer.watch(INVOCATION_INPUT_TOPIC, key); const signal = new AbortController().signal; + // A backend shows as idle before its LISTEN has run. const listenerPid = async () => ( await pool.query<{ pid: number }>( - 'SELECT pid FROM pg_stat_activity WHERE application_name = $1 AND state = $2', - [`${name}:invocations`, 'idle'] + "SELECT pid FROM pg_stat_activity WHERE application_name = $1 AND state = 'idle' AND query LIKE 'LISTEN %'", + [`${name}:invocations`] ) ).rows[0]?.pid; + // Read, then wait, the way consumers do: each wait() past the backoff + // retries the connection, so a failed attempt cannot strand the watch. + const poll = (read: () => Promise, ready: (value: T) => boolean) => + until(async () => { + const value = await read(); + if (!ready(value)) + await watch.wait(watch.revision, INVOCATION_FALLBACK_MS, signal); + return value; + }, ready); try { - await until( - async () => watch.revision, - (revision) => revision > 0 - ); - const firstPid = await until(listenerPid, (pid) => pid !== undefined); + const firstPid = await poll(listenerPid, (pid) => pid !== undefined); const beforeDisconnect = watch.revision; + const beforeRestore = restorations(); await pool.query('SELECT pg_terminate_backend($1)', [firstPid]); + // Consumers reread on disconnect, then fall back to timed reads. await until( async () => watch.revision, (revision) => revision > beforeDisconnect ); - // Normal consumers reread on disconnect, then use the slow fallback - // while reconnect is backed off, then wait/read again. - await watch.wait(watch.revision, 1_000, signal); - const reconnected = watch.wait(watch.revision, 5_000, signal); - const secondPid = await until( - listenerPid, - (pid) => pid !== undefined && pid !== firstPid + await poll( + async () => restorations(), + (count) => count > beforeRestore ); - await reconnected; + expect(listenerConnects).toBeGreaterThanOrEqual(3); + const secondPid = await listenerPid(); + expect(secondPid).toBeDefined(); expect(secondPid).not.toBe(firstPid); const beforeNotify = watch.revision; await pool.query('SELECT pg_notify($1, $2)', [ @@ -1196,6 +1229,8 @@ describe.skipIf(process.platform === 'win32')( (revision) => revision > beforeNotify ); } finally { + connectSpy.mockRestore(); + warn.mockRestore(); watch.dispose(); await observer.close(); await listenerPool.end(); From 6cba93b2fcf9412da1c9357aafc5076c03c0441e Mon Sep 17 00:00:00 2001 From: "vercel[bot]" <35613825+vercel[bot]@users.noreply.github.com> Date: Mon, 12 Oct 2026 20:00:48 +0000 Subject: [PATCH 2/2] [world-postgres] Wait for the first LISTEN to settle before the reconnect test's baselines pg_stat_activity lists the LISTEN before the listener has read its reply. If the first connect failed (degraded), its "restored" log and wake can land after beforeRestore/beforeDisconnect are sampled, satisfying both waits before the terminated backend reconnects; secondPid is then undefined. A notification follows the LISTEN reply on the same socket, so round-trip one before sampling. Co-Authored-By: Claude Opus 5.5 Co-Authored-By: Pranay Prakash <1797812+pranaygp@users.noreply.github.com> --- packages/world-postgres/test/invoke.test.ts | 26 ++++++++++++++------- 1 file changed, 17 insertions(+), 9 deletions(-) diff --git a/packages/world-postgres/test/invoke.test.ts b/packages/world-postgres/test/invoke.test.ts index 06bcd02891..5ec9e37d5d 100644 --- a/packages/world-postgres/test/invoke.test.ts +++ b/packages/world-postgres/test/invoke.test.ts @@ -1202,8 +1202,24 @@ describe.skipIf(process.platform === 'win32')( await watch.wait(watch.revision, INVOCATION_FALLBACK_MS, signal); return value; }, ready); + const notified = async () => { + const before = watch.revision; + await pool.query('SELECT pg_notify($1, $2)', [ + INVOCATION_INPUT_TOPIC, + key, + ]); + await until( + async () => watch.revision, + (revision) => revision > before + ); + }; try { const firstPid = await poll(listenerPid, (pid) => pid !== undefined); + // The server lists the LISTEN before the client has read its reply, + // and a failed first connect leaves a restore log still to come. A + // notification follows that reply on the same socket, so once it is + // seen the first subscription has settled, wake and log included. + await notified(); const beforeDisconnect = watch.revision; const beforeRestore = restorations(); await pool.query('SELECT pg_terminate_backend($1)', [firstPid]); @@ -1220,15 +1236,7 @@ describe.skipIf(process.platform === 'win32')( const secondPid = await listenerPid(); expect(secondPid).toBeDefined(); expect(secondPid).not.toBe(firstPid); - const beforeNotify = watch.revision; - await pool.query('SELECT pg_notify($1, $2)', [ - INVOCATION_INPUT_TOPIC, - key, - ]); - await until( - async () => watch.revision, - (revision) => revision > beforeNotify - ); + await notified(); } finally { connectSpy.mockRestore(); warn.mockRestore();