From 40ca3bceefe829d93f6bff7abadfd25d04f01310 Mon Sep 17 00:00:00 2001 From: Burak Keskin Date: Tue, 20 Oct 2026 19:32:05 +0300 Subject: [PATCH] fix(pg): keep the connection usable after an error in a rows-limited query With the rows option, Query drives the extended protocol with Flush instead of Sync (_getRows) and only handleCommandComplete / handleEmptyQuery send the terminating Sync. When the server answers with an ErrorResponse instead, it discards messages until a Sync arrives and withholds ReadyForQuery - handleError sent nothing, so readyForQuery never returned and every later query on the connection queued forever. Restore the error-path Sync by tracking the one Sync a rows-limited portal owes, sent by whichever of handleCommandComplete, handleEmptyQuery or handleError observes the end of the sequence. The flag also keeps exactly one Sync from going out when an ErrorResponse follows a CommandComplete (a timed-out query on some backends) or when a row-parse error is surfaced on ReadyForQuery, where a second Sync would produce a second ReadyForQuery and let a later query settle mid-flight. Fixes #3707 --- packages/pg/lib/query.js | 38 +++-- ...error-handling-prepared-statement-tests.js | 130 ++++++++++++++++++ 2 files changed, 160 insertions(+), 8 deletions(-) diff --git a/packages/pg/lib/query.js b/packages/pg/lib/query.js index 6b9214199..0f35033e9 100644 --- a/packages/pg/lib/query.js +++ b/packages/pg/lib/query.js @@ -30,6 +30,9 @@ class Query extends EventEmitter { // potential for multiple results this._results = this._result this._canceledDueToError = false + // set when _getRows pipelines Flush instead of Sync, cleared by whichever + // of handleCommandComplete / handleEmptyQuery / handleError sends that Sync + this._syncOwed = false } requiresPreparation() { @@ -104,9 +107,7 @@ class Query extends EventEmitter { this._result.addCommandComplete(msg) // need to sync after each command complete of a prepared statement // if we were using a row count which results in multiple calls to _getRows - if (this.rows) { - connection.sync() - } + this._sendOwedSync(connection) } // if a named prepared statement is created with empty query text @@ -114,13 +115,17 @@ class Query extends EventEmitter { // since we pipeline sync immediately after execute we don't need to do anything here // unless we have rows specified, in which case we did not pipeline the initial sync call handleEmptyQuery(connection) { - if (this.rows) { - connection.sync() - } + this._sendOwedSync(connection) } handleError(err, connection) { - // need to sync after error during a prepared statement + // after an ErrorResponse the backend discards messages until it sees a Sync + // and withholds ReadyForQuery. In rows-mode no Sync was pipelined (_getRows + // flushed instead), so send the one we owe or the client never leaves the + // 'awaiting ReadyForQuery' state and every later query on the connection + // queues forever. + // http://developer.postgresql.org/pgdocs/postgres/protocol-flow.html#PROTOCOL-FLOW-EXT-QUERY + this._sendOwedSync(connection) if (this._canceledDueToError) { err = this._canceledDueToError this._canceledDueToError = false @@ -200,11 +205,28 @@ class Query extends EventEmitter { if (!rows) { connection.sync() } else { - // otherwise flush the call out to read more rows + // otherwise flush the call out to read more rows; the backend stays in + // the middle of an extended-protocol sequence until CommandComplete (or + // an ErrorResponse / EmptyQuery — see _sendOwedSync) sends the Sync that + // terminates it + this._syncOwed = true connection.flush() } } + // send the terminating Sync for a rows-limited portal, exactly once. In + // rows-mode the Sync is not pipelined with Execute, so whoever observes the + // end of the extended-protocol sequence owes the backend one Sync before it + // will answer ReadyForQuery. An ErrorResponse can follow a CommandComplete + // (e.g. a timed-out query on some backends), which is why the flag guards + // against a second Sync — and a second ReadyForQuery — being sent. + _sendOwedSync(connection) { + if (this._syncOwed) { + this._syncOwed = false + connection.sync() + } + } + // http://developer.postgresql.org/pgdocs/postgres/protocol-flow.html#PROTOCOL-FLOW-EXT-QUERY prepare(connection) { // TODO refactor this poor encapsulation diff --git a/packages/pg/test/integration/client/query-error-handling-prepared-statement-tests.js b/packages/pg/test/integration/client/query-error-handling-prepared-statement-tests.js index 13ecf4b53..b8682238b 100644 --- a/packages/pg/test/integration/client/query-error-handling-prepared-statement-tests.js +++ b/packages/pg/test/integration/client/query-error-handling-prepared-statement-tests.js @@ -124,3 +124,133 @@ suite.test('query killed during query execution of prepared statement', function }) ) }) + +// When a query runs with the `rows` option, Execute is pipelined with Flush +// instead of Sync, so an ErrorResponse leaves the backend in ignore-till-sync: +// it discards every message until a Sync arrives and withholds ReadyForQuery. +// The client used to never send that Sync, so the query rejected correctly but +// the connection was wedged — every later query queued forever. +// https://github.com/brianc/node-postgres/issues/3707 +suite.test('rows-limited query that errors mid-portal stays usable', function (done) { + if (helper.args.native) { + return done() + } + const client = new Client(helper.args) + client.connect( + assert.success(function () { + // one ReadyForQuery per completed extended-protocol sequence — the + // errored query and the follow-up, no more (a stray second Sync would + // produce a third and desync the queue) + let readyForQueryCount = 0 + client.connection.on('readyForQuery', function () { + readyForQueryCount++ + }) + // divides by zero on row 5, fetched in pages of 2: the error arrives + // between PortalSuspended round trips, with the portal still open + client.query( + { text: 'select 10 / (5 - i) as v from generate_series(1, 7) g(i)', rows: 2 }, + assert.calls(function (err) { + assert.equal(err.message, 'division by zero') + client.query( + 'select 41 + 1 as v', + assert.calls(function (err, res) { + assert.ifError(err) + assert.equal(res.rows[0].v, 42) + client.end( + assert.success(function () { + assert.equal(readyForQueryCount, 2) + done() + }) + ) + }) + ) + }) + ) + }) + ) +}) + +suite.test('rows-limited query that errors before any row stays usable', function (done) { + if (helper.args.native) { + return done() + } + const client = new Client(helper.args) + client.connect( + assert.success(function () { + let readyForQueryCount = 0 + client.connection.on('readyForQuery', function () { + readyForQueryCount++ + }) + // rejected at Parse, before the portal is ever created + client.query( + { text: 'select 1; select 2', rows: 2 }, + assert.calls(function (err) { + assert.equal(err.message, 'cannot insert multiple commands into a prepared statement') + client.query( + 'select 41 + 1 as v', + assert.calls(function (err, res) { + assert.ifError(err) + assert.equal(res.rows[0].v, 42) + client.end( + assert.success(function () { + assert.equal(readyForQueryCount, 2) + done() + }) + ) + }) + ) + }) + ) + }) + ) +}) + +// A row that throws while being parsed cancels the query client-side: the +// portal still ends with CommandComplete, which sends the owed Sync, and the +// error is only surfaced on ReadyForQuery. handleError must not send a second +// Sync for the same sequence — two ReadyForQuery frames would let the next +// query settle on the first one, mid-flight. +suite.test('rows-limited query that fails while parsing a row syncs once', function (done) { + if (helper.args.native) { + return done() + } + const client = new Client(helper.args) + client.connect( + assert.success(function () { + let readyForQueryCount = 0 + client.connection.on('readyForQuery', function () { + readyForQueryCount++ + }) + const query = { + text: 'select 1 as v', + rows: 1, + types: { + getTypeParser: function () { + return function () { + throw new Error('boom') + } + }, + }, + } + client.query( + query, + assert.calls(function (err) { + assert.equal(err.message, 'boom') + client.query( + 'select 41 + 1 as v', + assert.calls(function (err, res) { + assert.ifError(err) + assert.equal(res.rows[0].v, 42) + client.end( + assert.success(function () { + assert.equal(readyForQueryCount, 2) + done() + }) + ) + }) + ) + }) + ) + }) + ) +})