[ecsbeats]

Fixes #1216.

If a backend disconnects while a sql.begin() callback is awaiting other work, begin() rejects but the callback keeps running. Once the pool reconnects, that callback can issue queries on the replacement connection. Returning or throwing from it can also commit or roll back a different transaction. Queries still queued inside the original transaction can remain pending indefinitely.

Keep the close error on the transaction, reject its queued queries, and reject subsequent queries through that transaction's handler, including automatic commit/rollback. Clear the connection's pending write and query/result/error state before reuse, too. Without that reset, the first query after reconnect can receive the old backend's fatal error.

The four regression tests cover queued queries, a query through the disconnected transaction, and late commit/rollback. The latter three share a fixture that pauses the original callback until a replacement transaction is open. No new dependencies or public API changes.

Related to #1204. This reproduction specifically covers a backend disconnect followed by connection reuse; I haven't verified that it explains the original report.

Reproduce

Save the following as repro.mjs in the repository root. It needs Node 22 and a PostgreSQL login that can terminate its own other sessions.

import assert from 'node:assert/strict'
import postgres from './src/index.js'

const pool = postgres({ max: 1 })
const admin = postgres({ max: 1 })
let resume
  , connected
const gate = new Promise(resolve => { resume = resolve })
const ready = new Promise(resolve => { connected = resolve })
const failed = pool.begin(async sql => {
  connected((await sql`select 1`).state.pid)
  await gate
}).catch(error => error)

try {
  await admin`select pg_terminate_backend(${ await ready }::int)`
  assert.equal((await failed).code, 'CONNECTION_CLOSED')
  await pool.begin(async sql => {
    const [{ id: before }] = await sql`select txid_current()::text as id`
    resume()
    await new Promise(resolve => setImmediate(resolve))
    const [{ id: after }] = await sql`select txid_current()::text as id`
    console.log({ before, after })
    assert.equal(after, before)
  })
} finally {
  resume()
  await Promise.all([pool.end({ timeout: 0 }), admin.end({ timeout: 0 })])
}

For a disposable local server:

docker run --rm -d --name postgres-disconnect-repro \
  -p 127.0.0.1:55432:5432 \
  -e POSTGRES_HOST_AUTH_METHOD=trust postgres:17
docker exec postgres-disconnect-repro pg_isready -U postgres
PGHOST=127.0.0.1 PGPORT=55432 PGUSER=postgres PGDATABASE=postgres node repro.mjs
docker stop postgres-disconnect-repro

Wait for pg_isready to report that the server is accepting connections before running the script. On upstream 411429e, the assertion fails because the transaction IDs differ: the old callback has committed the replacement transaction. On this branch they stay equal and the script exits successfully. I verified both outcomes.

Tests

The first commit, 0282301, adds only the regression tests; the second applies the fix. With the PostgreSQL setup from the existing test workflow in place:

git worktree add ../postgres-disconnect-before 0282301
(cd ../postgres-disconnect-before && PGUSER=postgres PGSOCKET=/var/run/postgresql npm run test:esm)
PGUSER=postgres PGSOCKET=/var/run/postgresql npm test
npm exec --yes --package=eslint@8.57.1 -- eslint src tests

The first run times out in Disconnect rejects queued transaction queries and allows reconnect. The fixed branch passes. The full suite needs both PostgreSQL servers from the workflow (ports 5432 and 5433), SSL, logical replication, prepared transactions, the PostgreSQL CLI tools, and Deno 1.x. Use disposable databases: the existing bootstrap changes server settings and recreates its test database and roles.

Validated against PostgreSQL 17.11:

• Node 12.22.12 and 22.22.0: all 268 tests pass in both ESM and CommonJS.
• Deno 1.46.3: all 268 tests pass.
• Bun 1.4.2: the standalone reproduction and focused probes for all four failure cases pass. The full suite cannot register tests because tests/test.js:20 assumes a Node stack format; the same failure occurs on unmodified upstream.
• ESLint passes.

[AlephNotation]

I've also recently gotten bitten by this one

[batalyse-gmbh]

Thanks for this PR. We tested it together with #1209: the two fix every crash case we could reproduce. We found one case that is still broken, and a one-line fix for it.

What we tested

• postgres 3.4.9, Node 26, PostgreSQL 18.4 and 18.6 (same results on both)
• max: 1, so the same connection is always reused

On 3.4.9 without the PRs

Scenario	What happens
transaction_timeout or pg_terminate_backend during a query in sql.begin()	Crash: the nextWrite TypeError from #1133. If you catch it, the connection never comes back: every reconnect ends in CONNECT_TIMEOUT.
idle_in_transaction_session_timeout, then another begin() reuses the connection	The dead transaction's INSERT and COMMIT run inside the other transaction. That transaction throws, but its inserts stay committed (#1216).
Any kill during a query	The next query fails with the old backend's 57P01 / 25P04 error.

A plain null check in nextWrite (as in #1168) is not enough. It turns the crash into the same endless CONNECT_TIMEOUT.

With #1209 + #1215

All three are fixed. ✅

Still broken: a connection reset inside begin()

Symptom: after an ECONNRESET (for example a failover), one query works. After that the connection hangs forever.

Cause:

1. On a reset, the socket emits 'error' first and 'close' a moment later. Promise callbacks run in between.
2. In that gap, begin() sends its rollback. The socket is still set and closedError is not set yet, so the rollback is added to sent.
3. closed(true) only clears sent when !hadError, so the rollback stays there.
4. After reconnecting, the connection treats that old rollback as in flight and waits for a reply that never comes.

Repro:

import net from "node:net";
import postgres from "postgres";

// A TCP proxy in front of Postgres, so we can reset the socket like a failover would.
const pairs = [];
const proxy = net.createServer((client) => {
  const upstream = net.connect(5432, "127.0.0.1");
  client.pipe(upstream).pipe(client);
  client.on("error", () => {});
  upstream.on("error", () => {});
  pairs.push([client, upstream]);
}).listen(0, "127.0.0.1");
await new Promise((r) => proxy.once("listening", r));

const sql = postgres({ host: "127.0.0.1", port: proxy.address().port, max: 1 /* , user, password */ });

await sql.begin(async (tx) => {
  setTimeout(() => pairs.forEach(([c, u]) => (u.destroy(), c.resetAndDestroy())), 500);
  await tx`select pg_sleep(3)`;
}).catch((e) => console.log("begin rejected:", e.code));

const ask = (tag) => sql`select ${tag}::text as tag`.then(([r]) => console.log(tag, "->", r.tag));
await ask("Q1");
await Promise.race([ask("Q2"), new Promise((r) => setTimeout(() => r(console.log("Q2 still pending after 3s")), 3000))]);
process.exit(0);

Output on 3.4.9, and on 3.4.9 + #1209 + #1215:

begin rejected: CONNECTION_CLOSED
Q1 -> Q1
Q2 still pending after 3s

Expected: Q2 -> Q2.

Fix

Always clear sent when the socket closes:

-    !hadError && (query || sent.length) && error(Errors.connection('CONNECTION_CLOSED', options, socket))
+    if (query || sent.length)
+      error(Errors.connection('CONNECTION_CLOSED', options, socket))

It has to be an if. The code has no semicolons, so a line starting with ( would join onto the return reconnect() above it.

We tested it on top of #1209 + #1215:

• the repro above passes;
• a plain query outside a transaction still gets ECONNRESET;
• all the scenarios in the table still pass.

Written by Claude Opus

[cat-a-guerra]

We independently found and patched the same two issues in production (3.4.9): the stale chunk/nextWriteTimer that leaves reconnects ending in CONNECT_TIMEOUT, and the stale query/errorResponse that replays the old backend's 57P01 onto the next query. Our patch makes the same two resets, so +1 to this approach. We also saw sql.end() never resolve while the connection was in that stale state.

One path this PR doesn't reach: closed() returns early via if (initial) return reconnect() before the new reset lines run. So a connection killed during startup (the fetch_types query) still replays its 57P01 onto the first query, plus an unhandled rejection. Repro and a one-line fix in #1223. We confirmed this PR alone doesn't change the output there.

[realies]

A reproduction for the stale chunk part of this PR, which shows up as a reconnect that stalls and then sends old query bytes before the StartupMessage.

If a query is written in the same event-loop turn as the socket closes, its bytes stay in chunk. closed() cancels the pending nextWrite but leaves chunk and nextWriteTimer set, so after the reconnect connected() appends the StartupMessage to the old bytes, and no flush is scheduled until chunk reaches 1024 bytes.

Run against docker run --rm -p 5432:5432 -e POSTGRES_HOST_AUTH_METHOD=trust postgres:17:

import postgres from 'postgres'
const o = { host: '127.0.0.1', user: 'postgres', database: 'postgres', ssl: false, max: 1, connect_timeout: 2 }
let closed = false
const sql = postgres({ ...o, onclose: () => closed = true }), admin = postgres(o)
const [{ pid }] = await sql`select pg_backend_pid() as pid`
const tick = () => { // one query per loop turn, from a setImmediate queued behind the client's flush
  if (closed) return
  sql.unsafe(`select 1 /*${'x'.repeat(900)}*/`).catch(() => {})
  queueMicrotask(() => setImmediate(tick))
}
setImmediate(tick)
await admin`select pg_terminate_backend(${pid})`
while (!closed) await new Promise(r => setTimeout(r, 10))
for (let i = 1; i <= 3; i++)
  await sql`select ${i}::int as i`.then(() => console.log(i, 'ok'), e => console.log(i, e.code, e.message))
await Promise.all([sql.end(), admin.end()])

On master 411429e (Node 24, PostgreSQL 17): 1 CONNECT_TIMEOUT, 2 57P01, 3 ok, and the server logs invalid length of startup packet (3 of 3 runs; 9 of 10 in a looped version).

With this PR's src/ changes applied: 1 ok every time, and no invalid startup packet (0 of 3, and 0 of 10 looped). So the reset in closed() fixes it.

One thing it leaves: with this PR applied, in 3 of 6 runs queries 2 and 3 never settle (no reply within 4 s each). On master the same hang appears with a different ordering of the loop (3 of 6 runs), so it looks pre-existing rather than caused by this PR. It seems related to queries left in sent/query after a socket error, as in #1226.

[evk-soft]

Independent verification at 18cffe016927e01481b9e746587da789f4260a4c, compared with upstream 411429e7bd7a3d61155ca9a70a97c111823702ea (this PR alone, without #1209):

Windows x64, Node 24.21.0 / Bun 1.4.2, ESM / regenerated CommonJS, PostgreSQL 17.11 in a disposable container.

Five focused cases failed on upstream and passed on this PR in all four runtime/entry combinations (20 before/after pairs):

• transaction-local queued queries all reject after backend termination;
• a late query rejects after the connection closes, without a null-socket exception;
• the old transaction cannot read transaction-local settings on the replacement connection;
• the old callback's late COMMIT cannot finish the replacement transaction;
• the old callback's late ROLLBACK cannot finish the replacement transaction.

I also exercised actual TCP FIN/RST through a local relay during an observed pg_sleep inside begin():

• FIN: upstream failed in all four combinations; this PR passed, including three successive queries after reconnect.
• RST: still fails on this PR in all four combinations, including Bun 1.4.2 on Windows. The interrupted transaction rejects, reconnect query 41 succeeds, and query 42 remains pending until the six-second process deadline. This corroborates the remaining gap already discussed here and in A query accepted between a socket's error and close never settles, and its connection wedges after the next reconnect (ECONNRESET on Node) #1246; it is not a regression introduced by this PR.

So these results support the transaction-local close latch and queue rejection, while also confirming that this PR alone is not a complete fix for recovery after TCP reset. I did not run the full upstream suite, Deno, or Cloudflare.

Standalone reproduction (save as repro-1215.mjs at the repository root)

Set REVIEW_DATABASE_URL to a disposable PostgreSQL database with permission to terminate its own sessions. Run node repro-1215.mjs <case> esm for queued, late-query, reused-query, late-commit, late-rollback, fin, and rst. Repeat with cjs after node transpile.cjs, and with bun instead of node. Each process asserts successful recovery; rst currently exits nonzero on this PR too.

import assert from 'node:assert/strict'
import net from 'node:net'
import { createRequire } from 'node:module'

const [scenario, entry = 'esm'] = process.argv.slice(2)
const imported = entry === 'cjs'
  ? createRequire(import.meta.url)('./cjs/src/index.js')
  : (await import('./src/index.js')).default
const postgres = imported.default ?? imported
const url = process.env.REVIEW_DATABASE_URL
assert.ok(url, 'Set REVIEW_DATABASE_URL to a disposable PostgreSQL server')
const delay = ms => new Promise(resolve => setTimeout(resolve, ms))
const deferred = () => {
  let resolve, reject
  const promise = new Promise((a, b) => { resolve = a; reject = b })
  return { promise, resolve, reject }
}
const timer = setTimeout(() => {
  console.error('FAIL: operation did not settle within 6 seconds')
  process.exit(2)
}, 6000)
const options = { max: 1, fetch_types: false, prepare: false, connect_timeout: 2, onnotice: () => {} }
const clients = []
const client = extra => {
  const sql = postgres(url, { ...options, ...extra })
  clients.push(sql)
  return sql
}
const observer = client()

async function disconnected() {
  const sql = client()
  const ready = deferred(), resume = deferred(), attempted = deferred()
  let oldTx
  const failed = sql.begin(async tx => {
    oldTx = tx
    ready.resolve((await tx`select pg_backend_pid() as pid`)[0].pid)
    await resume.promise
    if (scenario === 'late-query') {
      const error = await tx`select 42 as value`.then(() => null, error => error)
      attempted.resolve(error)
    }
    if (scenario === 'late-rollback') throw new Error('original callback failed')
  }).then(() => null, error => error)
  try {
    const pid = await ready.promise
    await observer`select pg_terminate_backend(${pid}::int)`
    const closed = await failed
    assert.equal(closed?.code, 'CONNECTION_CLOSED')
    if (scenario === 'late-query') {
      resume.resolve()
      assert.equal((await attempted.promise)?.code, 'CONNECTION_CLOSED')
      await delay(30)
    } else {
      await sql.begin(async tx => {
        const before = (await tx`select txid_current()::text as id`)[0].id
        if (scenario === 'reused-query') {
          await tx`select set_config('review.owner', 'replacement', true)`
          const error = await oldTx`select current_setting('review.owner')`.then(() => null, error => error)
          assert.equal(error?.code, 'CONNECTION_CLOSED', 'Old handle must not read the replacement transaction')
        }
        resume.resolve()
        await delay(50)
        const after = (await tx`select txid_current()::text as id`)[0].id
        assert.equal(after, before, 'Old callback must not finish the replacement transaction')
      })
    }
    assert.equal((await sql`select 43 as value`)[0].value, 43)
  } finally {
    resume.resolve()
  }
}

async function queued() {
  const sql = client({ max_pipeline: 1 })
  let outcomes
  const closed = await sql.begin(tx => {
    outcomes = Promise.allSettled([
      tx`select pg_terminate_backend(pg_backend_pid())`.execute(),
      tx`select 1`.execute(),
      tx`select 2`.execute()
    ])
    return outcomes
  }).then(() => null, error => error)
  assert.equal(closed?.code, 'CONNECTION_CLOSED')
  const results = await outcomes
  assert.deepEqual(results.map(result => result.status), ['rejected', 'rejected', 'rejected'])
  assert.equal((await sql`select 43 as value`)[0].value, 43)
}

async function transportLoss() {
  const target = new URL(url)
  const sockets = new Set()
  let pair
  const relay = net.createServer(downstream => {
    const upstream = net.connect(Number(target.port), target.hostname)
    pair = { downstream, upstream }
    for (const socket of [downstream, upstream]) {
      sockets.add(socket)
      socket.on('error', () => {})
      socket.on('close', () => sockets.delete(socket))
    }
    downstream.pipe(upstream).pipe(downstream)
  })
  await new Promise(resolve => relay.listen(0, '127.0.0.1', resolve))
  const via = new URL(url)
  via.port = String(relay.address().port)
  const sql = postgres(via.toString(), options)
  clients.push(sql)
  try {
    const [{ pid }] = await sql`select pg_backend_pid() as pid`
    const failed = sql.begin(async tx => {
      await tx`select pg_sleep(5)`
    }).then(() => null, error => error)
    for (;;) {
      const [state] = await observer`select wait_event from pg_stat_activity where pid = ${pid}`
      if (state?.wait_event === 'PgSleep') break
      await delay(5)
    }
    pair.upstream.unpipe(pair.downstream)
    pair.downstream.unpipe(pair.upstream)
    if (scenario === 'rst') pair.downstream.resetAndDestroy()
    else pair.downstream.end()
    pair.upstream.destroy()
    const error = await failed
    assert.ok(error, 'The interrupted transaction must reject')
    console.log(`Transaction rejected: ${error.code}`)
    for (const value of [41, 42, 43]) {
      assert.equal((await sql`select ${value}::int as value`)[0].value, value)
      console.log(`Reconnect query ${value}: passed`)
    }
  } finally {
    await sql.end({ timeout: 0 })
    for (const socket of sockets) socket.destroy()
    await new Promise(resolve => relay.close(resolve))
  }
}
try {
  if (scenario === 'queued') await queued()
  else if (scenario === 'fin' || scenario === 'rst') await transportLoss()
  else await disconnected()
  console.log(`PASS ${scenario} ${entry}`)
} catch (error) {
  console.error(`FAIL ${scenario} ${entry}:`, error)
  process.exitCode = 1
} finally {
  await Promise.allSettled(clients.map(sql => sql.end({ timeout: 0 })))
  clearTimeout(timer)
}

Prepared with Codex assistance; the results above come from executed checks.

[GrahamCampbell]

A reproduction of the transaction case and of the state a closed connection keeps, with a patch combining this PR's change with #1240's and #1246's, is in #1240 (comment).