arbitrary UNSAFE_TRANSACTION errors when using sql.begin #823

Activity

1. JuHwon

   changed the title [-]Arbitrary UNSAFE_TRANSACTION errors[/-] [+]Arbitrary UNSAFE_TRANSACTION errors when using sql.begin[/+] on Mar 9, 2024
2. JuHwon

   changed the title [-]Arbitrary UNSAFE_TRANSACTION errors when using sql.begin[/-] [+]arbitrary UNSAFE_TRANSACTION errors when using sql.begin[/+] on Mar 9, 2024
3. JuHwon commented on Mar 15, 2024

   

   JuHwon

   on Mar 15, 2024

   Last edited by JuHwon

   Author

   More actions

   error object of notice callback:

   {
      "severity_local":"WARNING",
      "severity":"WARNING",
      "code":"25001",
      "message":"there is already a transaction in progress",
      "file":"xact.c",
      "line":"3799",
      "routine":"BeginTransactionBlock"
   }

   maybe this is similar to this one: #274

   according to the debug logs the client executes a BEGIN statement on connection id e.g. 4, and then a few ms laters it executes a BEGIN statement with the same connection id again, before the first transaction was handled. there were no other statements between with the affected connection id.
4. ngsgh commented on Sep 1, 2026

   

   ngsgh

   on Sep 1, 2026

   More actions

   We hit this in production and traced a mechanism that fits the "arbitrary" part. Sharing in case it helps — we could not reproduce it locally, so treat the root-cause part as analysis of the source, not a verified repro.

   Where the lock is skipped

   src/connection.js (v3.4.9, and master at time of writing):

   build(q)
   return write(toBuffer(q))
     && !q.describeFirst
     && !q.cursorFn
     && sent.length < max_pipeline
     && (!q.options.onexecute || q.options.onexecute(connection))

   onexecute is what marks the connection as reserved for the transaction. It sits at the end of an && chain, so it is skipped entirely whenever an earlier term is falsy:

   • write(toBuffer(q)) returns false under socket backpressure
   • sent.length < max_pipeline is false once the pipeline is full (default 100)

   If BEGIN goes out without onexecute running, c.reserved is never set, and when the BEGIN response arrives the guard at src/connection.js (errored(Errors.generic('UNSAFE_TRANSACTION', ...))) fires.

   Why it looks arbitrary

   The chain can only break when BEGIN is pipelined onto an already-busy connection. On an idle connection sent.length is 0 and BEGIN is ~50 bytes, well under the write threshold, so neither term can be false — it is structurally safe there. That matches the reporter's "large variable-size operations very often": the failure needs a connection that is already saturated at the moment BEGIN is queued.

   Blast radius is wider than one query

   errored() calls stream.destroy(), so other queries pipelined on the same connection are torn down too. A single UNSAFE_TRANSACTION in your logs may correspond to more than one failed request.

   Reproduction attempts (all negative)

   Against a local Postgres 17, six shapes: TCP proxy with read stalls and throttling, forced socket.write() returning false, max_pipeline: 1, and 1000 pool queries against 200 pipelined transactions. Zero occurrences. Loopback kernel buffers are large enough that real backpressure never materialises, which is consistent with this only showing up over real network paths.

   Workaround we are using

   Retrying once on error.code === 'UNSAFE_TRANSACTION'. This appears safe because the guard fires at the BEGIN response — nothing in the transaction body has executed — and errored() destroys the socket, so the server rolls back. We verified no partial commits locally (committed rows always equalled the number of successful transactions).

   Happy to test a patch if someone wants to move onexecute out of the chain; note its return value currently participates in the busy/full queue decision, so it is not a pure one-liner.
5. dominikwagner commented on Sep 1, 2026

   

   dominikwagner

   on Sep 1, 2026

   Last edited by dominikwagner

   More actions

   We hit this in production on postgres@3.4.7 and traced it to the same code path @ngsgh described.

   Deterministic reproduction against unmodified PostgreSQL on localhost — no proxy, no protocol tampering:

   • Pool: max: 3, max_pipeline: 2
   • Hold all 3 connections busy with pg_sleep(0.5)
   • Fire 6 regular queries + 5 sql.begin() concurrently
   • Run this scenario 20 times: 12–16 iterations produce UNSAFE_TRANSACTION
   • PostgreSQL itself reports 25001 "there is already a transaction in progress"

   The 25001 warning is the smoking gun — PostgreSQL is telling us the driver sent two BEGINs on one connection.

   Root cause: Under load, connection.reserved can be cleared or overwritten while BEGIN is in flight. Three code paths in src/connection.js / src/index.js:

   1. execute() short-circuits: when sent.length >= max_pipeline or write() returns false, the && chain skips onexecute, so connection.reserved is never set — this is the path @ngsgh described, and it's what our test triggers
   2. handler() dispatches queued BEGIN queries onto already-busy connections
   3. drain() calls onopen() on a reserved connection, returning it to the pool mid-transaction

   Fix (three parts, all needed):

   1. execute() fires onexecute unconditionally when write() succeeds, and returns false so no more queries pipeline onto the connection
   2. handler() always queues onexecute queries (BEGIN) instead of dispatching to busy connections
   3. drain() guards on !connection.reserved before calling onopen()

   @porsager: I opened a PR with a fix — #1207
6. matheusbrramos commented on Sep 25, 2026

   

   matheusbrramos

   on Sep 25, 2026

   More actions

   Adding data to this thread, because one conclusion above is unsafe to act on: retrying on UNSAFE_TRANSACTION is not safe in 3.4.9, and the damage lands on other callers.

   errored() does not destroy the socket

   function errored(err) {
     stream && (stream.destroy(err), stream = null)

   stream here is the COPY stream (assigned only in the Copy*Response handlers), not the socket. When the guard fires, the BEGIN has already run on the server, the socket stays open, and the connection goes back to the pool inside the open transaction.

   What that does to unrelated queries

   Default options, max: 2. BEGIN is made the 100th in-flight query on a busy connection, then 20 unrelated autocommit INSERTs run through the same pool:

   sql.begin	rejects with UNSAFE_TRANSACTION
   unrelated inserts that resolved successfully	20
   visible to another connection right after	1
   sessions idle in transaction	1
   rows persisted after sql.end() (same as a deploy or idle_timeout)	1

   19 writes resolved and were silently rolled back. If a later sql.begin lands on that connection, its COMMIT commits them instead, as part of someone else's transaction. PostgreSQL's 25001 warning reported by @JuHwon above (a BEGIN arriving on a connection already in a transaction) is consistent with that second case.

   All four paths, three builds

   Standalone script, no application code: only the library under test and a disposable database (below). Run against PostgreSQL 16.12 on localhost, Node 24.13.

   Scenario	3.4.9	#1207	execute() as suggested in #1207 review + drain() guard, no handler() change
   1. BEGIN at the pipeline boundary, default max_pipeline	FAIL	PASS	PASS
   2. Unrelated writes after UNSAFE_TRANSACTION	FAIL (19/20 lost)	PASS	PASS
   3. BEGIN while socket.write() returns false	FAIL	FAIL	PASS
   4. 'drain' while a transaction sits between statements	FAIL (other caller's write lost)	PASS	PASS

   Scenarios 3 and 4 use the documented socket option. Bytes go to the server for real; only the return value of write() and the 'drain' event are what Node itself produces under backpressure.

   repro.mjs

   // Standalone reproduction for porsager/postgres #823 / #1189 / PR #1207.
   // No application code: only the library under test and a disposable database.
   //
   //   DATABASE_URL=postgres://... node repro.mjs <path-to-postgres/src/index.js>
   //
   // Four scenarios. Each prints PASS/FAIL and the evidence behind it.
   import net from 'node:net'
   import { pathToFileURL } from 'node:url'
   
   const libPath = process.argv[2]
   if (!libPath) throw new Error('usage: node repro.mjs <path to postgres/src/index.js>')
   const { default: postgres } = await import(pathToFileURL(libPath).href)
   const url = process.env.DATABASE_URL
   if (!url) throw new Error('Set DATABASE_URL to a disposable database')
   const endpoint = new URL(url)
   
   const probe = postgres(url, { max: 1, onnotice: () => {} })
   const table = 'pgjs_repro_' + Date.now()
   await probe`create table ${ probe(table) } (id text primary key)`
   const count = async (like) => (await probe`select count(*)::int as n from ${ probe(table) } where id like ${ like }`)[0].n
   const idleInTx = async (app) => (await probe`
     select count(*)::int as n from pg_stat_activity
     where state = 'idle in transaction' and application_name = ${ app }`)[0].n
   const sleep = ms => new Promise(r => setTimeout(r, ms))
   
   async function openSocket({ writeReturnsFalse = false } = {}) {
     const socket = net.createConnection({ host: endpoint.hostname, port: Number(endpoint.port || 5432) })
     await new Promise((ok, fail) => { socket.once('connect', ok); socket.once('error', fail) })
     if (writeReturnsFalse) {
       // What Node does under backpressure: the bytes are ACCEPTED and will be
       // sent; only the return value says "stop writing for now".
       const native = socket.write.bind(socket)
       socket.write = (...args) => { native(...args); return false }
     }
     return socket
   }
   
   const results = []
   function report(name, pass, evidence) {
     results.push({ name, pass, evidence })
     console.log(`${ pass ? 'PASS' : 'FAIL' }  ${ name }\n      ${ JSON.stringify(evidence) }`)
   }
   
   // 1. Pipeline boundary, DEFAULT options (max_pipeline 100). BEGIN becomes the
   //    100th in-flight query on a busy connection.
   {
     const app = 'repro-s1'
     let fails = 0
     for (const K of [98, 99, 100]) {
       const sql = postgres(url, { max: 2, onnotice: () => {}, connection: { application_name: app + K } })
       await Promise.all([sql`select 1`, sql`select 1`])
       const pending = [sql`select pg_sleep(0.3)`.execute(), sql`select pg_sleep(0.3)`.execute()]
       for (let i = 0; i < 2 * K; i++) pending.push(sql`select 1`.execute())
       let err = null
       const b = sql.begin(tx => tx`select 1`).catch(e => { err = e.code })
       await Promise.all([...pending, b])
       if (err) fails++
       await sql.end({ timeout: 5 })
     }
     report('1. BEGIN at the pipeline boundary (default max_pipeline)', fails === 0, { beginsFailedOutOf3: fails })
   }
   
   // 2. Collateral damage after UNSAFE_TRANSACTION: is the connection destroyed
   //    (so the server rolls back and retrying is safe), or does it go back to
   //    the pool inside the open transaction?
   {
     const app = 'repro-s2'
     const sql = postgres(url, { max: 2, idle_timeout: 20, onnotice: () => {}, connection: { application_name: app } })
     await Promise.all([sql`select 1`, sql`select 1`])
     const pending = [sql`select pg_sleep(0.3)`.execute(), sql`select pg_sleep(0.3)`.execute()]
     for (let i = 0; i < 2 * 99; i++) pending.push(sql`select 1`.execute())
     let err = null
     const b = sql.begin(tx => tx`select 1`).catch(e => { err = e.code })
     await Promise.all([...pending, b])
     // 20 unrelated autocommit inserts from "other callers", all resolve OK.
     const ids = Array.from({ length: 20 }, (_, i) => 's2-' + i)
     const settled = await Promise.allSettled(ids.map(id => sql`insert into ${ sql(table) } (id) values (${ id })`))
     const resolvedOk = settled.filter(s => s.status === 'fulfilled').length
     const visible = await count('s2-%')
     const idle = await idleInTx(app)
     await sql.end({ timeout: 5 }) // like a deploy / idle_timeout closing the connection
     const persisted = await count('s2-%')
     report('2. After UNSAFE_TRANSACTION, unrelated writes survive', err === null || persisted === resolvedOk,
       { beginError: err, resolvedOk, visibleRightAfter: visible, idleInTransaction: idle, persistedAfterClose: persisted })
   }
   
   // 3. Backpressure: socket.write() returns false for BEGIN (bytes accepted).
   {
     const sql = postgres(url, { max: 2, onnotice: () => {}, socket: () => openSocket({ writeReturnsFalse: true }) })
     let err = null
     // write() only reaches socket.write with >= 1024 buffered bytes; begin
     // options (letters/spaces, public API) push this BEGIN past it.
     await sql.begin(' '.repeat(1100), tx => tx`insert into ${ tx(table) } (id) values ('s3')`).catch(e => { err = e.code })
     await sql.end({ timeout: 5 })
     report('3. BEGIN while socket.write() returns false (backpressure)', err === null && await count('s3') === 1,
       { beginError: err, rowPersisted: await count('s3') })
   }
   
   // 4. 'drain' fires while a transaction sits between two statements.
   {
     const sockets = []
     const sql = postgres(url, { max: 1, onnotice: () => {}, socket: async () => { const s = await openSocket(); sockets.push(s); return s } })
     let release, started
     const idle = new Promise(r => release = r)
     const began = new Promise(r => started = r)
     const tx = sql.begin(async t => {
       await t`insert into ${ t(table) } (id) values ('s4-tx')`
       started()
       await idle                      // between statements, nothing in flight
       throw new Error('rollback on purpose')
     }).catch(() => 'rolled back')
     await began
     let otherResolved = false
     const other = sql`insert into ${ sql(table) } (id) values ('s4-other')`.execute().then(() => { otherResolved = true })
     await sleep(50)
     sockets[0].emit('drain')          // what Node emits when the write buffer empties
     await sleep(200)
     release()
     await tx
     await other
     await sql.end({ timeout: 5 })
     const persisted = await count('s4-other')
     report("4. 'drain' mid-transaction keeps other callers out", !(otherResolved && persisted === 0),
       { otherCallerResolved: otherResolved, otherCallerRowPersisted: persisted })
   }
   
   await probe`drop table ${ probe(table) }`
   await probe.end()
   console.log('\nSUMMARY ' + results.map(r => (r.pass ? 'P' : 'F')).join(''))

   DATABASE_URL=postgres://... node repro.mjs ./node_modules/postgres/src/index.js
   

   Happy to test further changes against this.
7. okomarov

   added a commit that references this issue on Oct 8, 2026

   fix: bind reserve() and begin() handles to a connection lease

   ...

   3d3f7b4