Skip to content

Commit 026741b

Browse files
VaguelySeriousVaguelySerious
authored
[world-postgres] Fix process crash when the database closes an idle connection (#4817)
(AI) Extracted from #4661 to shrink it. No behavior change beyond the fix below. - **world-postgres**: the pool the World creates now has an `error` listener. pg-pool re-emits a terminated idle client's error (`57P01`, database restart, `pg_terminate_backend`) on the pool, and with no listener that became an uncaught exception. Graphile Worker only listens while the queue runs, and `pool.end()` resolves before idle sockets close, so this could kill the process during or after `close()`. Test teardown is already handled by `TestPool` (#4785); this covers the World's own pool in production. - **world-vercel** (test-only): tests that pin the backend URL via env now skip while `WORKFLOW_SERVER_URL_OVERRIDE` is set, since the override takes precedence over the env var
1 parent f6d93f9 commit 026741b

6 files changed

Lines changed: 134 additions & 95 deletions

File tree

‎.changeset/postgres-idle-connection-crash.md‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,5 @@
1+
---
2+
"@workflow/world-postgres": patch
3+
---
4+
5+
Fix a process crash when the database closes an idle connection

‎packages/world-postgres/src/index.ts‎

Lines changed: 28 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -59,12 +59,7 @@ export function createWorld(
5959
): World & { start(): Promise<void> } {
6060
const maxPoolSize =
6161
config.maxPoolSize ?? getPositiveIntEnv('WORKFLOW_POSTGRES_MAX_POOL_SIZE');
62-
const pool =
63-
config.pool ||
64-
new Pool({
65-
connectionString: config.connectionString || getDefaultConnectionString(),
66-
...(maxPoolSize !== undefined ? { max: maxPoolSize } : {}),
67-
});
62+
const pool = config.pool || createOwnedPool(config, maxPoolSize);
6863

6964
const drizzle = createClient(pool);
7065
const queue = createQueue(config, pool);
@@ -114,6 +109,33 @@ export function createWorld(
114109
};
115110
}
116111

112+
/**
113+
* The pool the World creates when the caller passes none.
114+
*
115+
* It listens for `error`: pg-pool re-emits an idle client's connection error
116+
* (a database restart, `pg_terminate_backend`) on the pool, and an
117+
* EventEmitter with no `error` listener throws it as an uncaught exception.
118+
* Graphile Worker listens only while the queue runs, and `pool.end()`
119+
* resolves before its idle clients have closed, so without this a backend
120+
* terminated during or after `close()` ends the process. The dropped client is
121+
* already out of the pool, and the next query opens a new one.
122+
*/
123+
function createOwnedPool(
124+
config: PostgresWorldConfig,
125+
maxPoolSize: number | undefined
126+
): Pool {
127+
const pool = new Pool({
128+
connectionString: config.connectionString || getDefaultConnectionString(),
129+
...(maxPoolSize !== undefined ? { max: maxPoolSize } : {}),
130+
});
131+
pool.on('error', (error) => {
132+
console.warn(
133+
`[world-postgres] idle database connection closed: ${error.message}`
134+
);
135+
});
136+
return pool;
137+
}
138+
117139
// Re-export schema for users who want to extend or inspect the database schema
118140
export type { PostgresWorldConfig } from './config.js';
119141
export * from './drizzle/schema.js';

‎packages/world-postgres/src/reenqueue.test.ts‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,7 @@ vi.mock('pg', () => ({
2525
return {
2626
query: vi.fn(async () => ({ rows: [{ exists: false }] })),
2727
end: vi.fn(),
28+
on: vi.fn(),
2829
};
2930
}),
3031
}));

‎packages/world-vercel/src/events-v4.test.ts‎

Lines changed: 21 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -2339,23 +2339,29 @@ describe('v4 transport wraps pre-response failures the allowlist misses', () =>
23392339
expect(rejection.message).toContain('transport failure');
23402340
});
23412341

2342-
it('rejects a credential-bearing backend URL without dispatch or retry', async () => {
2343-
vi.stubEnv('VERCEL_WORKFLOW_SERVER_URL', 'http://user:password@127.0.0.1');
2344-
const fetchSpy = vi.spyOn(globalThis, 'fetch');
2342+
it.skipIf(WORKFLOW_SERVER_URL_OVERRIDE !== '')(
2343+
'rejects a credential-bearing backend URL without dispatch or retry',
2344+
async () => {
2345+
vi.stubEnv(
2346+
'VERCEL_WORKFLOW_SERVER_URL',
2347+
'http://user:password@127.0.0.1'
2348+
);
2349+
const fetchSpy = vi.spyOn(globalThis, 'fetch');
23452350

2346-
const rejection = await getWorkflowRunEventsV4(
2347-
'wrun_1',
2348-
{},
2349-
{ token: 'test-token' }
2350-
).catch((error: unknown) => error);
2351+
const rejection = await getWorkflowRunEventsV4(
2352+
'wrun_1',
2353+
{},
2354+
{ token: 'test-token' }
2355+
).catch((error: unknown) => error);
23512356

2352-
expect(rejection).toMatchObject({
2353-
name: 'TypeError',
2354-
message: 'HTTP(S) URLs with embedded credentials are unsupported',
2355-
});
2356-
expect(StreamError.is(rejection)).toBe(false);
2357-
expect(fetchSpy).not.toHaveBeenCalled();
2358-
});
2357+
expect(rejection).toMatchObject({
2358+
name: 'TypeError',
2359+
message: 'HTTP(S) URLs with embedded credentials are unsupported',
2360+
});
2361+
expect(StreamError.is(rejection)).toBe(false);
2362+
expect(fetchSpy).not.toHaveBeenCalled();
2363+
}
2364+
);
23592365

23602366
it('preserves unsupported headers from the backend configuration as non-retryable', async () => {
23612367
vi.stubEnv('VERCEL_WORKFLOW_SERVER_URL', 'http://127.0.0.1:12345');

‎packages/world-vercel/src/trace-propagation.test.ts‎

Lines changed: 39 additions & 36 deletions
Original file line numberDiff line numberDiff line change
@@ -367,46 +367,49 @@ describe('snapshot requests trace propagation', () => {
367367
expect(traceparent).toContain(traceId);
368368
});
369369

370-
it('injects traceparent on snapshots.save (undici request path)', async () => {
371-
const agent = new MockAgent();
372-
agent.disableNetConnect();
373-
let capturedHeaders: Record<string, unknown> | undefined;
374-
agent
375-
.get('https://vercel-workflow.com')
376-
.intercept({
377-
path: '/api/v2/runs/wrun_1/snapshot',
378-
method: 'PUT',
379-
})
380-
.reply(200, (opts) => {
381-
capturedHeaders = opts.headers as Record<string, unknown>;
382-
return 'ok';
383-
});
370+
it.skipIf(WORKFLOW_SERVER_URL_OVERRIDE !== '')(
371+
'injects traceparent on snapshots.save (undici request path)',
372+
async () => {
373+
const agent = new MockAgent();
374+
agent.disableNetConnect();
375+
let capturedHeaders: Record<string, unknown> | undefined;
376+
agent
377+
.get('https://vercel-workflow.com')
378+
.intercept({
379+
path: '/api/v2/runs/wrun_1/snapshot',
380+
method: 'PUT',
381+
})
382+
.reply(200, (opts) => {
383+
capturedHeaders = opts.headers as Record<string, unknown>;
384+
return 'ok';
385+
});
384386

385-
const { createSnapshotsStorage } = await import('./snapshots.js');
386-
const snapshots = createSnapshotsStorage({
387-
token: 'test-token',
388-
dispatcher: agent,
389-
});
387+
const { createSnapshotsStorage } = await import('./snapshots.js');
388+
const snapshots = createSnapshotsStorage({
389+
token: 'test-token',
390+
dispatcher: agent,
391+
});
390392

391-
const tracer = otelTrace.getTracer('test');
392-
let traceId = '';
393-
await tracer.startActiveSpan('flow-invocation', async (span) => {
394-
traceId = span.spanContext().traceId;
395-
await snapshots.save('wrun_1', new Uint8Array([1, 2, 3]), {
396-
eventsCursor: 'evnt_1',
397-
createdAt: new Date(),
393+
const tracer = otelTrace.getTracer('test');
394+
let traceId = '';
395+
await tracer.startActiveSpan('flow-invocation', async (span) => {
396+
traceId = span.spanContext().traceId;
397+
await snapshots.save('wrun_1', new Uint8Array([1, 2, 3]), {
398+
eventsCursor: 'evnt_1',
399+
createdAt: new Date(),
400+
});
401+
span.end();
398402
});
399-
span.end();
400-
});
401403

402-
const headers = new Headers(
403-
(capturedHeaders ?? {}) as Record<string, string>
404-
);
405-
const traceparent = headers.get('traceparent');
406-
expect(traceparent).toMatch(/^00-[0-9a-f]{32}-[0-9a-f]{16}-0[01]$/);
407-
expect(traceparent).toContain(traceId);
408-
agent.assertNoPendingInterceptors();
409-
});
404+
const headers = new Headers(
405+
(capturedHeaders ?? {}) as Record<string, string>
406+
);
407+
const traceparent = headers.get('traceparent');
408+
expect(traceparent).toMatch(/^00-[0-9a-f]{32}-[0-9a-f]{16}-0[01]$/);
409+
expect(traceparent).toContain(traceId);
410+
agent.assertNoPendingInterceptors();
411+
}
412+
);
410413
});
411414

412415
describe('ws stream transport upgrade trace propagation', () => {

‎packages/world-vercel/src/utils.test.ts‎

Lines changed: 40 additions & 38 deletions
Original file line numberDiff line numberDiff line change
@@ -625,48 +625,50 @@ describe('makeRequest URL validation', () => {
625625
vi.restoreAllMocks();
626626
});
627627

628-
it.each([
629-
'http',
630-
'https',
631-
])('preserves Fetch port-blocking errors for %s backend URLs', async (scheme) => {
632-
vi.stubEnv(NODE_HTTP_ENV_VAR, '0');
633-
vi.stubEnv('VERCEL_WORKFLOW_SERVER_URL', `${scheme}://127.0.0.1:21`);
634-
const fetchSpy = vi.spyOn(globalThis, 'fetch');
635-
const rejection = await makeRequest({
636-
endpoint: '/v3/runs/wrun_test/events',
637-
options: { method: 'GET' },
638-
schema: z.unknown(),
639-
config: { token: 'test-token' },
640-
}).catch((error: unknown) => error);
641-
642-
expect(fetchSpy).toHaveBeenCalledTimes(1);
643-
await expect(fetchSpy.mock.results[0].value).rejects.toBe(rejection);
644-
expect(rejection).toMatchObject({
645-
name: 'TypeError',
646-
message: 'fetch failed',
647-
cause: { message: 'bad port' },
648-
});
649-
expect(isRetryableEventPostError(rejection)).toBe(false);
650-
});
651-
652-
it.each([
653-
'0',
654-
'1',
655-
])('rejects unsupported backend protocols before dispatch (WORKFLOW_NODE_HTTP=%s)', async (mode) => {
656-
vi.stubEnv(NODE_HTTP_ENV_VAR, mode);
657-
vi.stubEnv('VERCEL_WORKFLOW_SERVER_URL', 'ftp://localhost');
658-
const fetchSpy = vi.spyOn(globalThis, 'fetch');
659-
660-
await expect(
661-
makeRequest({
628+
// These pin the backend URL through the environment, which the override
629+
// takes precedence over.
630+
it.skipIf(WORKFLOW_SERVER_URL_OVERRIDE !== '').each(['http', 'https'])(
631+
'preserves Fetch port-blocking errors for %s backend URLs',
632+
async (scheme) => {
633+
vi.stubEnv(NODE_HTTP_ENV_VAR, '0');
634+
vi.stubEnv('VERCEL_WORKFLOW_SERVER_URL', `${scheme}://127.0.0.1:21`);
635+
const fetchSpy = vi.spyOn(globalThis, 'fetch');
636+
const rejection = await makeRequest({
662637
endpoint: '/v3/runs/wrun_test/events',
663638
options: { method: 'GET' },
664639
schema: z.unknown(),
665640
config: { token: 'test-token' },
666-
})
667-
).rejects.toThrow(TypeError);
668-
expect(fetchSpy).not.toHaveBeenCalled();
669-
});
641+
}).catch((error: unknown) => error);
642+
643+
expect(fetchSpy).toHaveBeenCalledTimes(1);
644+
await expect(fetchSpy.mock.results[0].value).rejects.toBe(rejection);
645+
expect(rejection).toMatchObject({
646+
name: 'TypeError',
647+
message: 'fetch failed',
648+
cause: { message: 'bad port' },
649+
});
650+
expect(isRetryableEventPostError(rejection)).toBe(false);
651+
}
652+
);
653+
654+
it.skipIf(WORKFLOW_SERVER_URL_OVERRIDE !== '').each(['0', '1'])(
655+
'rejects unsupported backend protocols before dispatch (WORKFLOW_NODE_HTTP=%s)',
656+
async (mode) => {
657+
vi.stubEnv(NODE_HTTP_ENV_VAR, mode);
658+
vi.stubEnv('VERCEL_WORKFLOW_SERVER_URL', 'ftp://localhost');
659+
const fetchSpy = vi.spyOn(globalThis, 'fetch');
660+
661+
await expect(
662+
makeRequest({
663+
endpoint: '/v3/runs/wrun_test/events',
664+
options: { method: 'GET' },
665+
schema: z.unknown(),
666+
config: { token: 'test-token' },
667+
})
668+
).rejects.toThrow(TypeError);
669+
expect(fetchSpy).not.toHaveBeenCalled();
670+
}
671+
);
670672
});
671673

672674
describe('makeRequest transport errors', () => {

0 commit comments

Comments
 (0)