From c36c096dae4bb2abc16b5f831c37df19d69388c0 Mon Sep 17 00:00:00 2001 From: CTO Date: Fri, 31 Jul 2026 18:02:25 +0000 Subject: [PATCH] fix(db): patch postgres.js queued-write-after-close crash (BLO-19583) postgres@3.4.9 buffers sub-1024-byte writes and flushes them from a setImmediate(nextWrite) callback, while closed() nulls `socket` synchronously on peer disconnect. A write queued after that point dereferenced a null socket inside the immediate -- outside every try/catch on the stack -- so it escaped as an uncaughtException and terminated paperclip-0: TypeError: Cannot read properties of null (reading 'write') at Immediate.nextWrite (postgres/src/connection.js:255:22) The pool routes around dead connections, so ordinary queries never hit this. It needs a caller that dispatches straight to Connection.execute() -- sql.reserve() or sql.begin() -- and execute() only short-circuits on `terminated`, which an abrupt close never sets. We use transactions heavily via drizzle. Upstream PR porsager/postgres#1168 guards the dereference but silently drops the buffered write, leaving the query pending forever. This patch also routes the failure through the driver's own connection-closed path so the query rejects with CONNECTION_CLOSED and the pool reconnects. Patches all three builds (src/, cjs/, cf/), which carry the identical function. pnpm-lock.yaml is intentionally not committed: CI owns lockfile updates and refresh-lockfile.yml lands them after merge. Co-Authored-By: Claude --- package.json | 1 + .../postgres-connection-close-race.test.ts | 256 ++++++++++++++++++ patches/postgres@3.4.9.patch | 66 +++++ 3 files changed, 323 insertions(+) create mode 100644 packages/db/src/postgres-connection-close-race.test.ts create mode 100644 patches/postgres@3.4.9.patch diff --git a/package.json b/package.json index f77cf851f74..865675dfb0a 100644 --- a/package.json +++ b/package.json @@ -131,6 +131,7 @@ "pnpm": { "patchedDependencies": { "embedded-postgres@18.1.0-beta.16": "patches/embedded-postgres@18.1.0-beta.16.patch", + "postgres@3.4.9": "patches/postgres@3.4.9.patch", "brace-expansion@5.0.8": "patches/brace-expansion@5.0.8.patch" }, "overrides": { diff --git a/packages/db/src/postgres-connection-close-race.test.ts b/packages/db/src/postgres-connection-close-race.test.ts new file mode 100644 index 00000000000..a7ab2a78bcc --- /dev/null +++ b/packages/db/src/postgres-connection-close-race.test.ts @@ -0,0 +1,256 @@ +import { EventEmitter } from "node:events"; +import { describe, expect, it } from "vitest"; +import postgres from "postgres"; + +/** + * Regression coverage for the postgres.js queued-write-after-close race + * (BLO-19583, parent BLO-19578; upstream porsager/postgres#1154). + * + * `Connection.write()` buffers sub-1024-byte payloads and flushes them from a + * `setImmediate(nextWrite)` callback. `closed()` nulls `socket` synchronously + * when the peer disconnects. A write queued *after* that point therefore ran + * `socket.write(...)` against `null` inside the immediate — outside every + * `try/catch` on the stack — so it escaped as an `uncaughtException` and killed + * paperclip-0: + * + * TypeError: Cannot read properties of null (reading 'write') + * at Immediate.nextWrite (postgres/src/connection.js:255:22) + * at process.processImmediate (node:internal/timers:504:21) + * + * `patches/postgres@3.4.9.patch` guards the dereference and routes the affected + * query through the driver's normal connection-closed path. Reverting that patch + * makes the first test below fail with the exact TypeError above. + * + * The transport is injected through postgres.js's `socket` option, so these + * tests need no database and no timing luck. + */ + +const int32 = (value: number): Buffer => { + const buf = Buffer.alloc(4); + buf.writeInt32BE(value); + return buf; +}; + +const message = (type: string, payload: Buffer = Buffer.alloc(0)): Buffer => + Buffer.concat([Buffer.from(type, "latin1"), int32(4 + payload.length), payload]); + +const AUTHENTICATION_OK = message("R", int32(0)); +const READY_FOR_QUERY = message("Z", Buffer.from("I", "latin1")); +const HANDSHAKE = Buffer.concat([AUTHENTICATION_OK, READY_FOR_QUERY]); +/** ParseComplete, BindComplete, NoData, CommandComplete, ReadyForQuery. */ +const EMPTY_QUERY_RESULT = Buffer.concat([ + message("1"), + message("2"), + message("n"), + message("C", Buffer.from("SELECT 0\0", "latin1")), + READY_FOR_QUERY, +]); + +/** Minimal stand-in for a connected `net.Socket` speaking just enough protocol. */ +class FakeSocket extends EventEmitter { + readyState = "open"; + host = "127.0.0.1"; + port = 5432; + /** When false, the fake backend stops answering (simulates a dead peer). */ + responding = true; + private greeted = false; + + write(chunk: Buffer, callback?: () => void): boolean { + if (this.responding) { + const reply = this.greeted ? EMPTY_QUERY_RESULT : HANDSHAKE; + this.greeted = true; + setImmediate(() => this.emit("data", reply)); + } + callback?.(); + return true; + } + + end(): this { + this.readyState = "closed"; + return this; + } + destroy(): this { + this.readyState = "closed"; + return this; + } + pause(): this { + return this; + } + resume(): this { + return this; + } + setKeepAlive(): this { + return this; + } + + /** Abrupt peer disconnect, i.e. `close` with `hadError === false`. */ + remoteClose(): void { + this.readyState = "closed"; + this.emit("close", false); + } +} + +const nextTick = (): Promise => new Promise((resolve) => setImmediate(resolve)); + +async function drainImmediates(count = 8): Promise { + for (let index = 0; index < count; index += 1) { + await nextTick(); + } +} + +/** + * Swap vitest's `uncaughtException` handlers out for a recorder, so an escaping + * throw is asserted on directly instead of tearing down the worker. + */ +function captureUncaughtExceptions(): { errors: Error[]; restore: () => void } { + const previous = process.listeners("uncaughtException"); + const errors: Error[] = []; + const recorder = (error: Error): void => { + errors.push(error); + }; + process.removeAllListeners("uncaughtException"); + process.on("uncaughtException", recorder); + return { + errors, + restore: () => { + process.removeListener("uncaughtException", recorder); + for (const listener of previous) { + process.on("uncaughtException", listener as (error: Error) => void); + } + }, + }; +} + +interface Harness { + sql: ReturnType; + /** The socket backing the most recent connection attempt. */ + currentSocket: () => FakeSocket; + /** Resolves once the driver's `closed()` has run and nulled its socket. */ + socketClosed: Promise; + shutdown: () => Promise; +} + +function createHarness(): Harness { + let socket: FakeSocket | null = null; + let markClosed: () => void = () => {}; + const socketClosed = new Promise((resolve) => { + markClosed = resolve; + }); + + const sql = postgres({ + max: 1, + fetch_types: false, + prepare: false, + connect_timeout: 5, + idle_timeout: null, + // Keep the driver from parking a multi-minute lifetime timer that would + // otherwise hold the vitest worker's event loop open after the test. + max_lifetime: null, + onclose: () => markClosed(), + // `socket` is a real postgres.js option (src/index.js:495) that supplies the + // transport; it is simply absent from the package's shipped typings. + socket: () => { + socket = new FakeSocket(); + return socket; + }, + } as unknown as Parameters[0]); + + return { + sql, + currentSocket: () => { + if (!socket) throw new Error("no socket has been created yet"); + return socket; + }, + socketClosed, + shutdown: async () => { + await Promise.race([ + sql.end({ timeout: 0 }).catch(() => undefined), + new Promise((resolve) => setTimeout(resolve, 1_000)), + ]); + }, + }; +} + +/** Opens the pooled connection, then reserves it so queries reach `Connection.execute()` directly. */ +async function openAndReserve(harness: Harness) { + await harness.sql`select 1`; + return harness.sql.reserve(); +} + +describe("postgres.js queued-write-after-close race", () => { + it("rejects a write queued after the socket closed, without throwing in the immediate", async () => { + const harness = createHarness(); + const uncaught = captureUncaughtExceptions(); + try { + const reserved = await openAndReserve(harness); + const socket = harness.currentSocket(); + + // Kill the peer and wait until closed() has nulled the driver's socket. + socket.responding = false; + socket.remoteClose(); + await harness.socketClosed; + await drainImmediates(2); + + // A reserved handle dispatches straight to Connection.execute(), which + // only short-circuits on `terminated` — never set by an abrupt close. So + // this queues setImmediate(nextWrite) against a null socket. + // + // Deliberately not awaited: without the patch this query never settles, + // and awaiting it would surface the regression as an opaque test timeout + // instead of the assertions below. + let outcome = "pending"; + void reserved`select 1`.execute().then( + () => { + outcome = "resolved"; + }, + (error: { code?: string }) => { + outcome = error.code ?? "unknown"; + }, + ); + await drainImmediates(); + + expect(uncaught.errors.map((error) => error.message)).toEqual([]); + expect(outcome).toBe("CONNECTION_CLOSED"); + + // AC2: the pool must still recover on the next operation. + await expect(harness.sql`select 1`).resolves.toBeDefined(); + } finally { + uncaught.restore(); + await harness.shutdown(); + } + }); + + it("cancels a write already queued when the socket closes underneath it", async () => { + const harness = createHarness(); + const uncaught = captureUncaughtExceptions(); + try { + const reserved = await openAndReserve(harness); + const socket = harness.currentSocket(); + socket.responding = false; + + let outcome = "pending"; + void reserved`select 1`.execute().then( + () => { + outcome = "resolved"; + }, + (error: { code?: string }) => { + outcome = error.code ?? "unknown"; + }, + ); + + // Let the query dispatch through microtasks so write() has queued its + // immediate, then close within the same macrotask — before it fires. + for (let index = 0; index < 20; index += 1) { + await Promise.resolve(); + } + socket.remoteClose(); + await drainImmediates(); + + expect(uncaught.errors.map((error) => error.message)).toEqual([]); + expect(outcome).toBe("CONNECTION_CLOSED"); + } finally { + uncaught.restore(); + await harness.shutdown(); + } + }); +}); diff --git a/patches/postgres@3.4.9.patch b/patches/postgres@3.4.9.patch new file mode 100644 index 00000000000..a7c655928ae --- /dev/null +++ b/patches/postgres@3.4.9.patch @@ -0,0 +1,66 @@ +diff --git a/cf/src/connection.js b/cf/src/connection.js +index 8e79170..8270fd8 100644 +--- a/cf/src/connection.js ++++ b/cf/src/connection.js +@@ -254,6 +254,17 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose + } + + function nextWrite(fn) { ++ // BLO-19583 / porsager/postgres#1154: closed() nulls `socket`, but a write ++ // queued afterwards still schedules setImmediate(nextWrite). The immediate ++ // then dereferenced a null socket and threw *outside* any try/catch, taking ++ // the whole process down. Guard it, and fail the affected query through the ++ // driver's own connection-closed path so it rejects instead of hanging. ++ if (socket === null) { ++ nextWriteTimer !== null && clearImmediate(nextWriteTimer) ++ chunk = nextWriteTimer = null ++ error(Errors.connection('CONNECTION_CLOSED', options, socket)) ++ return false ++ } + const x = socket.write(chunk, fn) + nextWriteTimer !== null && clearImmediate(nextWriteTimer) + chunk = nextWriteTimer = null +diff --git a/cjs/src/connection.js b/cjs/src/connection.js +index 07f6716..0ac92ec 100644 +--- a/cjs/src/connection.js ++++ b/cjs/src/connection.js +@@ -252,6 +252,17 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose + } + + function nextWrite(fn) { ++ // BLO-19583 / porsager/postgres#1154: closed() nulls `socket`, but a write ++ // queued afterwards still schedules setImmediate(nextWrite). The immediate ++ // then dereferenced a null socket and threw *outside* any try/catch, taking ++ // the whole process down. Guard it, and fail the affected query through the ++ // driver's own connection-closed path so it rejects instead of hanging. ++ if (socket === null) { ++ nextWriteTimer !== null && clearImmediate(nextWriteTimer) ++ chunk = nextWriteTimer = null ++ error(Errors.connection('CONNECTION_CLOSED', options, socket)) ++ return false ++ } + const x = socket.write(chunk, fn) + nextWriteTimer !== null && clearImmediate(nextWriteTimer) + chunk = nextWriteTimer = null +diff --git a/src/connection.js b/src/connection.js +index 1b1cccd..657f98f 100644 +--- a/src/connection.js ++++ b/src/connection.js +@@ -252,6 +252,17 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose + } + + function nextWrite(fn) { ++ // BLO-19583 / porsager/postgres#1154: closed() nulls `socket`, but a write ++ // queued afterwards still schedules setImmediate(nextWrite). The immediate ++ // then dereferenced a null socket and threw *outside* any try/catch, taking ++ // the whole process down. Guard it, and fail the affected query through the ++ // driver's own connection-closed path so it rejects instead of hanging. ++ if (socket === null) { ++ nextWriteTimer !== null && clearImmediate(nextWriteTimer) ++ chunk = nextWriteTimer = null ++ error(Errors.connection('CONNECTION_CLOSED', options, socket)) ++ return false ++ } + const x = socket.write(chunk, fn) + nextWriteTimer !== null && clearImmediate(nextWriteTimer) + chunk = nextWriteTimer = null