fix(db): stop the postgres driver from crashing the process on a write/close race (#12227)
## Thinking Path
> - Paperclip is the open source app people use to manage AI agents for
work
> - The server and its test suites talk to PostgreSQL through the
`postgres` (postgres.js) driver, and tests routinely tear their
databases down while connections still carry traffic
> - The driver flushes small buffered frames from a `setImmediate`, and
that deferred flush calls `socket.write()` without checking that the
socket still exists; a reserved connection whose backend died keeps
accepting queries, so the flush can fire with a null socket
> - The resulting `TypeError` escapes from a timer callback with no
try/catch above it, crashing the process — in CI this fails suites whose
tests all passed ("Vitest caught 1 unhandled error"), and the same crash
is reported against the driver in the wild after ECONNRESET
> - The latest driver release (3.4.9) still has the bug, so this pull
request adds a pnpm patch guarding the flush and normalizing timer state
on close, plus a deterministic regression test
> - The benefit is CI that no longer fails randomly on a teardown race,
and production processes that survive a database connection dying at the
wrong moment
## Linked Issues or Issue Description
No public issue exists; the underlying problem follows the bug-report
template.
**What happened?**
CI jobs fail with all tests passing: vitest reports `Vitest caught 1
unhandled error during the test run` with `TypeError: Cannot read
properties of null (reading 'write')` at `postgres/src/connection.js`
`Immediate.nextWrite`. The attribution points at whichever test file
happened to be running (e.g. `native-codex-runner.integration.test.ts`),
because the throw comes from a process-level timer callback, not from a
test. The identical crash is reported against the upstream driver by
other projects after `ECONNRESET` (e.g. immich-app/immich#25098).
**Expected behavior**
A connection dying between a write being scheduled and its deferred
flush must settle the affected queries through the driver's normal
connection-error path, never throw from a bare timer callback.
**Steps to reproduce**
Run the new `packages/db/src/postgres-driver-teardown.test.ts` with the
patch removed: reserve a connection (`sql.reserve()` — the same surface
`sql.begin()` uses), destroy the backend socket, wait for the client to
process the close, then issue one query on the reserved connection. The
deferred flush fires one tick later with `socket === null` and crashes
the process with exactly the CI signature.
**Paperclip version or commit**
master `198fc8b28`, `postgres@3.4.9` (latest release; bug still present
on the driver's master branch).
## What Changed
- `patches/postgres@3.4.9.patch` (new, wired via
`pnpm.patchedDependencies`): `nextWrite` returns without writing when
`socket === null`, dropping the buffered bytes — the close path has
already settled every in-flight query, so those bytes have nowhere to
go. The `closed()` and `terminate()` handlers additionally reset
`nextWriteTimer`/`chunk` after `clearImmediate`, so a stale cleared
handle cannot silently block a future reconnect's first flush. All three
shipped builds (`src`, `cjs`, `cf`) get the identical change.
- `packages/db/src/postgres-driver-teardown.test.ts` (new):
deterministic reproduction against a minimal in-process fake wire server
(startup auth + an empty result for the `fetch_types` bootstrap).
Asserts the late query settles with `CONNECTION_DESTROYED` through
`sql.end()` instead of crashing the process.
## Verification
- The regression test fails against unpatched `postgres@3.4.9` with the
exact CI signature (verified by running the same scenario against an
unpatched checkout) and passes with the patch.
- Full `packages/db` suite: 27 files / 101 tests pass.
- Spot-checked server suites that exercise the database through the
patched driver.
## Risks
- Low. The behavioral change activates only in a state that previously
crashed the process (write flush with no socket). Dropping the buffered
bytes matches what the connection's close path already promised callers:
every in-flight query has been settled with a connection error.
- The timer/chunk reset in `closed()`/`terminate()` prevents a
theoretical stale-handle hang after reconnect; on the normal path both
were already reset by `nextWrite`.
- The patch pins to `postgres@3.4.9`; a future driver upgrade will
surface the patch for re-evaluation (pnpm fails loudly on version
mismatch), and the guard can be dropped if the fix lands upstream.
## Model Used
Claude Fable 5 (`claude-fable-5`, Anthropic) via Claude Code — agentic
coding session with tool use (driver source analysis, wire-protocol fake
server, local test execution).
## Checklist
- [x] I have included a thinking path that traces from project context
to this change
- [x] I have specified the model used (with version and capability
details)
- [x] I have checked ROADMAP.md and confirmed this PR does not duplicate
planned core work
- [x] I have searched GitHub for duplicate or related PRs and linked
them above
- [x] I have either (a) linked existing issues with `Fixes: #` / `Closes
#` / `Refs #` OR (b) described the issue in-PR following the relevant
issue template
- [x] I have not referenced internal/instance-local Paperclip issues or
links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip`
URLs)
- [x] My branch name describes the change (e.g. `docs/...`, `fix/...`)
and contains no internal Paperclip ticket id or instance-derived details
- [x] I have run tests locally and they pass
- [x] I have added or updated tests where applicable
- [x] I have updated relevant documentation to reflect my changes
- [x] I have considered and documented any risks above
- [ ] All Paperclip CI gates are green
- [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups
- [x] I will address all Greptile and reviewer comments before
requesting merge
This commit is contained in:
Devin Foley
GitHub
parent
198fc8b281
commit
6c7c0fd1f2
|
|
@ -83,7 +83,8 @@
|
|||
"pnpm": {
|
||||
"patchedDependencies": {
|
||||
"embedded-postgres@18.1.0-beta.16": "patches/embedded-postgres@18.1.0-beta.16.patch",
|
||||
"acpx@0.12.0": "patches/acpx@0.12.0.patch"
|
||||
"acpx@0.12.0": "patches/acpx@0.12.0.patch",
|
||||
"postgres@3.4.9": "patches/postgres@3.4.9.patch"
|
||||
},
|
||||
"overrides": {
|
||||
"rollup": ">=4.59.0",
|
||||
|
|
|
|||
|
|
@ -0,0 +1,108 @@
|
|||
import net from "node:net";
|
||||
import { afterEach, describe, expect, it } from "vitest";
|
||||
import postgres from "postgres";
|
||||
|
||||
/**
|
||||
* Regression coverage for patches/postgres@3.4.9.patch.
|
||||
*
|
||||
* The unpatched driver crashes the whole process when a write races a
|
||||
* connection close: `write()` buffers small frames and flushes them from a
|
||||
* `setImmediate(nextWrite)`, and `nextWrite` called `socket.write()` without
|
||||
* checking that the socket still exists. A reserved connection (sql.reserve /
|
||||
* sql.begin) whose backend dies keeps accepting queries — its handler calls
|
||||
* `execute()` with no socket-state check — so the deferred flush fired with
|
||||
* `socket === null` and threw `TypeError: Cannot read properties of null
|
||||
* (reading 'write')` from a timer callback with no try/catch above it. In CI
|
||||
* that surfaced as vitest's "Vitest caught 1 unhandled error" failing suites
|
||||
* whose tests had all passed, whenever a test tore down its database while a
|
||||
* connection still had traffic.
|
||||
*
|
||||
* The fake server below speaks just enough wire protocol to hand the client
|
||||
* an open connection (startup auth, then an empty result set for the
|
||||
* fetch_types bootstrap query); the crash path itself never needs a real
|
||||
* query to succeed. Against the unpatched driver this test fails through
|
||||
* vitest's unhandled-error detection with exactly the CI signature; with the
|
||||
* patch, the late query settles through the normal CONNECTION_DESTROYED path.
|
||||
*/
|
||||
describe("postgres driver teardown race", () => {
|
||||
let server: net.Server | null = null;
|
||||
let sql: ReturnType<typeof postgres> | null = null;
|
||||
|
||||
afterEach(async () => {
|
||||
if (sql) await sql.end({ timeout: 1 }).catch(() => {});
|
||||
sql = null;
|
||||
if (server) await new Promise((resolve) => server!.close(resolve));
|
||||
server = null;
|
||||
});
|
||||
|
||||
it("does not crash the process when a reserved connection's backend dies before a write flushes", async () => {
|
||||
const backendSockets: net.Socket[] = [];
|
||||
|
||||
// AuthenticationOk (R, code 0) + ReadyForQuery (Z, idle).
|
||||
const authOk = Buffer.from([0x52, 0, 0, 0, 8, 0, 0, 0, 0]);
|
||||
const readyForQuery = Buffer.from([0x5a, 0, 0, 0, 5, 0x49]);
|
||||
const emptyQueryReply = Buffer.concat([
|
||||
Buffer.from([0x31, 0, 0, 0, 4]), // ParseComplete
|
||||
Buffer.from([0x32, 0, 0, 0, 4]), // BindComplete
|
||||
Buffer.from([0x54, 0, 0, 0, 6, 0, 0]), // RowDescription, zero fields
|
||||
// CommandComplete "SELECT 0"
|
||||
Buffer.from([0x43, 0, 0, 0, 0x0d, 0x53, 0x45, 0x4c, 0x45, 0x43, 0x54, 0x20, 0x30, 0]),
|
||||
]);
|
||||
|
||||
server = net.createServer((socket) => {
|
||||
backendSockets.push(socket);
|
||||
let greeted = false;
|
||||
socket.on("data", () => {
|
||||
if (!greeted) {
|
||||
// First frame is the StartupMessage.
|
||||
greeted = true;
|
||||
socket.write(Buffer.concat([authOk, readyForQuery]));
|
||||
return;
|
||||
}
|
||||
// Any later frame (the fetch_types bootstrap query) gets an empty
|
||||
// result set + ReadyForQuery, enough for the client to finish
|
||||
// opening and resolve the reserve.
|
||||
socket.write(Buffer.concat([emptyQueryReply, readyForQuery]));
|
||||
});
|
||||
socket.on("error", () => {});
|
||||
});
|
||||
await new Promise<void>((resolve) => server!.listen(0, "127.0.0.1", resolve));
|
||||
const port = (server.address() as net.AddressInfo).port;
|
||||
|
||||
const closed = new Promise<void>((resolve) => {
|
||||
sql = postgres({
|
||||
host: "127.0.0.1",
|
||||
port,
|
||||
user: "test",
|
||||
database: "test",
|
||||
max: 1,
|
||||
idle_timeout: 0,
|
||||
connect_timeout: 5,
|
||||
onclose: () => resolve(),
|
||||
});
|
||||
});
|
||||
|
||||
// A reserved connection keeps routing queries straight to this one
|
||||
// physical connection — the same surface sql.begin() transactions use.
|
||||
const reserved = await sql!.reserve();
|
||||
|
||||
// Kill the backend and wait until the client has fully processed the
|
||||
// close (socket nulled, pool notified).
|
||||
for (const socket of backendSockets) socket.destroy();
|
||||
await closed;
|
||||
|
||||
// Late query on the dead reserved connection: unpatched, this buffered
|
||||
// its frame and scheduled the deferred flush that crashed the process
|
||||
// one tick later. Patched, execute() refuses it up front, so it settles
|
||||
// immediately — no pool shutdown required.
|
||||
const settled = await reserved`select 1`.catch((error: unknown) => error);
|
||||
expect(settled).toBeInstanceOf(Error);
|
||||
expect(String((settled as Error).message)).toContain("CONNECTION_CLOSED");
|
||||
|
||||
// Let any stray deferred flush run before the test ends, so a regression
|
||||
// in the nextWrite guard still fails this test via the unhandled error.
|
||||
await new Promise((resolve) => setImmediate(() => setImmediate(resolve)));
|
||||
|
||||
reserved.release();
|
||||
});
|
||||
});
|
||||
|
|
@ -0,0 +1,156 @@
|
|||
diff --git a/cf/src/connection.js b/cf/src/connection.js
|
||||
index 8e79170aeb13efe8c3c77ccede0cd3a115b5e1b5..c17a54b62fefa1f13bc38cd82e25d52ed467c94e 100644
|
||||
--- a/cf/src/connection.js
|
||||
+++ b/cf/src/connection.js
|
||||
@@ -159,6 +159,14 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose
|
||||
if (terminated)
|
||||
return queryError(q, Errors.connection('CONNECTION_DESTROYED', options))
|
||||
|
||||
+ // The connection lost its socket (backend death, network close) and has
|
||||
+ // not reconnected yet. Reserved connections keep routing queries here,
|
||||
+ // and buffering their frames would schedule a flush with nowhere to
|
||||
+ // write. Refuse up front the same way `terminated` does, so the query
|
||||
+ // settles immediately instead of dangling until pool shutdown.
|
||||
+ if (socket === null)
|
||||
+ return queryError(q, Errors.connection('CONNECTION_CLOSED', options, socket))
|
||||
+
|
||||
if (stream)
|
||||
return queryError(q, Errors.generic('COPY_IN_PROGRESS', 'You cannot execute queries during copy'))
|
||||
|
||||
@@ -254,6 +262,16 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose
|
||||
}
|
||||
|
||||
function nextWrite(fn) {
|
||||
+ if (socket === null) {
|
||||
+ // The connection closed between this write being scheduled and the
|
||||
+ // immediate firing (or a caller raced the close handler). The close
|
||||
+ // path has already settled every in-flight query, so the buffered
|
||||
+ // bytes have nowhere to go: drop them instead of crashing the
|
||||
+ // process from a timer callback with no try/catch above it.
|
||||
+ nextWriteTimer !== null && clearImmediate(nextWriteTimer)
|
||||
+ chunk = nextWriteTimer = null
|
||||
+ return false
|
||||
+ }
|
||||
const x = socket.write(chunk, fn)
|
||||
nextWriteTimer !== null && clearImmediate(nextWriteTimer)
|
||||
chunk = nextWriteTimer = null
|
||||
@@ -427,6 +445,7 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose
|
||||
error(Errors.connection('CONNECTION_DESTROYED', options))
|
||||
|
||||
clearImmediate(nextWriteTimer)
|
||||
+ chunk = nextWriteTimer = null
|
||||
if (socket) {
|
||||
socket.removeListener('data', data)
|
||||
socket.removeListener('connect', connected)
|
||||
@@ -440,6 +459,7 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose
|
||||
remaining = 0
|
||||
incomings = null
|
||||
clearImmediate(nextWriteTimer)
|
||||
+ chunk = nextWriteTimer = null
|
||||
socket.removeListener('data', data)
|
||||
socket.removeListener('connect', connected)
|
||||
idleTimer.cancel()
|
||||
diff --git a/cjs/src/connection.js b/cjs/src/connection.js
|
||||
index 07f6716702ac888215c86ad6e0071aaf55ae519c..949aa427e1afd0f68bc4cdcdf062e571568fd24d 100644
|
||||
--- a/cjs/src/connection.js
|
||||
+++ b/cjs/src/connection.js
|
||||
@@ -157,6 +157,14 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose
|
||||
if (terminated)
|
||||
return queryError(q, Errors.connection('CONNECTION_DESTROYED', options))
|
||||
|
||||
+ // The connection lost its socket (backend death, network close) and has
|
||||
+ // not reconnected yet. Reserved connections keep routing queries here,
|
||||
+ // and buffering their frames would schedule a flush with nowhere to
|
||||
+ // write. Refuse up front the same way `terminated` does, so the query
|
||||
+ // settles immediately instead of dangling until pool shutdown.
|
||||
+ if (socket === null)
|
||||
+ return queryError(q, Errors.connection('CONNECTION_CLOSED', options, socket))
|
||||
+
|
||||
if (stream)
|
||||
return queryError(q, Errors.generic('COPY_IN_PROGRESS', 'You cannot execute queries during copy'))
|
||||
|
||||
@@ -252,6 +260,16 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose
|
||||
}
|
||||
|
||||
function nextWrite(fn) {
|
||||
+ if (socket === null) {
|
||||
+ // The connection closed between this write being scheduled and the
|
||||
+ // immediate firing (or a caller raced the close handler). The close
|
||||
+ // path has already settled every in-flight query, so the buffered
|
||||
+ // bytes have nowhere to go: drop them instead of crashing the
|
||||
+ // process from a timer callback with no try/catch above it.
|
||||
+ nextWriteTimer !== null && clearImmediate(nextWriteTimer)
|
||||
+ chunk = nextWriteTimer = null
|
||||
+ return false
|
||||
+ }
|
||||
const x = socket.write(chunk, fn)
|
||||
nextWriteTimer !== null && clearImmediate(nextWriteTimer)
|
||||
chunk = nextWriteTimer = null
|
||||
@@ -425,6 +443,7 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose
|
||||
error(Errors.connection('CONNECTION_DESTROYED', options))
|
||||
|
||||
clearImmediate(nextWriteTimer)
|
||||
+ chunk = nextWriteTimer = null
|
||||
if (socket) {
|
||||
socket.removeListener('data', data)
|
||||
socket.removeListener('connect', connected)
|
||||
@@ -438,6 +457,7 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose
|
||||
remaining = 0
|
||||
incomings = null
|
||||
clearImmediate(nextWriteTimer)
|
||||
+ chunk = nextWriteTimer = null
|
||||
socket.removeListener('data', data)
|
||||
socket.removeListener('connect', connected)
|
||||
idleTimer.cancel()
|
||||
diff --git a/src/connection.js b/src/connection.js
|
||||
index 1b1cccde43b4570d5d071d6ffaab2669b3c2065a..8bb205e5b4b92282711ec8d105960508ccfc4252 100644
|
||||
--- a/src/connection.js
|
||||
+++ b/src/connection.js
|
||||
@@ -157,6 +157,14 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose
|
||||
if (terminated)
|
||||
return queryError(q, Errors.connection('CONNECTION_DESTROYED', options))
|
||||
|
||||
+ // The connection lost its socket (backend death, network close) and has
|
||||
+ // not reconnected yet. Reserved connections keep routing queries here,
|
||||
+ // and buffering their frames would schedule a flush with nowhere to
|
||||
+ // write. Refuse up front the same way `terminated` does, so the query
|
||||
+ // settles immediately instead of dangling until pool shutdown.
|
||||
+ if (socket === null)
|
||||
+ return queryError(q, Errors.connection('CONNECTION_CLOSED', options, socket))
|
||||
+
|
||||
if (stream)
|
||||
return queryError(q, Errors.generic('COPY_IN_PROGRESS', 'You cannot execute queries during copy'))
|
||||
|
||||
@@ -252,6 +260,16 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose
|
||||
}
|
||||
|
||||
function nextWrite(fn) {
|
||||
+ if (socket === null) {
|
||||
+ // The connection closed between this write being scheduled and the
|
||||
+ // immediate firing (or a caller raced the close handler). The close
|
||||
+ // path has already settled every in-flight query, so the buffered
|
||||
+ // bytes have nowhere to go: drop them instead of crashing the
|
||||
+ // process from a timer callback with no try/catch above it.
|
||||
+ nextWriteTimer !== null && clearImmediate(nextWriteTimer)
|
||||
+ chunk = nextWriteTimer = null
|
||||
+ return false
|
||||
+ }
|
||||
const x = socket.write(chunk, fn)
|
||||
nextWriteTimer !== null && clearImmediate(nextWriteTimer)
|
||||
chunk = nextWriteTimer = null
|
||||
@@ -425,6 +443,7 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose
|
||||
error(Errors.connection('CONNECTION_DESTROYED', options))
|
||||
|
||||
clearImmediate(nextWriteTimer)
|
||||
+ chunk = nextWriteTimer = null
|
||||
if (socket) {
|
||||
socket.removeListener('data', data)
|
||||
socket.removeListener('connect', connected)
|
||||
@@ -438,6 +457,7 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose
|
||||
remaining = 0
|
||||
incomings = null
|
||||
clearImmediate(nextWriteTimer)
|
||||
+ chunk = nextWriteTimer = null
|
||||
socket.removeListener('data', data)
|
||||
socket.removeListener('connect', connected)
|
||||
idleTimer.cancel()
|
||||
Loading…
Reference in New Issue