From 940be1142cfca10fe79808dcb5cf7c360eda4da1 Mon Sep 17 00:00:00 2001 From: Nigro Simone Date: Sun, 9 Aug 2026 12:28:01 +0200 Subject: [PATCH 1/2] Wake JS on libuv poll errors instead of dropping the callback --- src/connection.cc | 18 +++++++++ test/poll-error.js | 93 ++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 111 insertions(+) create mode 100644 test/poll-error.js diff --git a/src/connection.cc b/src/connection.cc index 0285b3a..6145daa 100644 --- a/src/connection.cc +++ b/src/connection.cc @@ -824,6 +824,16 @@ void Connection::on_io_readable(uv_poll_t* handle, int status, int revents) { LOG("Connection::on_io_readable"); TRACEF("Connection::on_io_readable:status %d\n", status); TRACEF("Connection::on_io_readable:revents %d\n", revents); + if(status < 0) { + // libuv signals a poll error (POLLERR, as a reset peer produces) with status < 0 and + // revents == 0, having already stopped the handle, so no further event can arrive. Emit + // anyway: consumeInput() then surfaces the error, instead of the connection going silent. + LOG("Connection::on_io_readable poll error"); + Connection* self = (Connection*) handle->data; + self->is_reading = false; + self->Emit("readable"); + return; + } if(revents & UV_READABLE) { LOG("Connection::on_io_readable UV_READABLE"); Connection* self = (Connection*) handle->data; @@ -836,6 +846,14 @@ void Connection::on_io_writable(uv_poll_t* handle, int status, int revents) { LOG("Connection::on_io_writable"); TRACEF("Connection::on_io_writable:status %d\n", status); TRACEF("Connection::on_io_writable:revents %d\n", revents); + if(status < 0) { + // Same as the read side: the handle is already stopped, so release the writable() waiter + // now or it never runs. flush() reports the error. + LOG("Connection::on_io_writable poll error"); + Connection* self = (Connection*) handle->data; + self->Emit("writable"); + return; + } if(revents & UV_WRITABLE) { LOG("Connection::on_io_readable UV_WRITABLE"); Connection* self = (Connection*) handle->data; diff --git a/test/poll-error.js b/test/poll-error.js new file mode 100644 index 0000000..e3730d8 --- /dev/null +++ b/test/poll-error.js @@ -0,0 +1,93 @@ +var net = require('net'); +var assert = require('assert'); +var PQ = require('../'); + +// A reset peer reaches libuv as POLLERR, which it reports with status < 0 and revents == 0 after +// stopping the handle. Look only at revents and that wakeup is lost for good. +// +// The reset comes from a proxy, not from terminating the backend: a clean exit sends FIN, which is +// an ordinary readable event and misses this path. resetAndDestroy makes the RST deterministic. +describe('poll error', function () { + var proxy; + var proxyPort; + var clientSockets; + + beforeEach(function (done) { + clientSockets = []; + proxy = net.createServer(function (client) { + var upstream = net.connect( + Number(process.env.PGPORT || 5432), + process.env.PGHOST || 'localhost' + ); + clientSockets.push(client); + client.pipe(upstream); + upstream.pipe(client); + client.on('error', function () {}); + upstream.on('error', function () {}); + }); + proxy.listen(0, '127.0.0.1', function () { + proxyPort = proxy.address().port; + done(); + }); + }); + + afterEach(function (done) { + proxy.close(function () { + done(); + }); + }); + + it('wakes the reader when the peer resets the connection', function (done) { + if (typeof net.Socket.prototype.resetAndDestroy !== 'function') { + return this.skip(); + } + + var pq = new PQ(); + // connect asynchronously: the proxy runs in this process, so a blocking connect would never + // let it accept + pq.connect('host=127.0.0.1 port=' + proxyPort, function (err) { + assert.ifError(err); + assert(pq.setNonBlocking(true)); + + // nothing comes back for a long while, so the reset is the only thing left for the socket + // to report and the test cannot pass on a readable that carried real data + assert(pq.sendQuery('SELECT pg_sleep(10)'), pq.errorMessage()); + assert.strictEqual(pq.flush(), 0, 'should have flushed the query to the socket'); + + var finished = false; + // under mocha's own timeout, so the failure names the actual problem + var timer = setTimeout(function () { + finish(new Error('no readable event arrived after the peer reset the connection')); + }, 1500); + + var finish = function (err) { + if (finished) return; + finished = true; + clearTimeout(timer); + pq.removeListener('readable', onReadable); + pq.stopReader(); + pq.finish(); + done(err); + }; + + var onReadable = function () { + assert.strictEqual( + pq.consumeInput(), + false, + 'consumeInput should report the reset connection' + ); + assert(pq.errorMessage(), 'a reset connection should leave an error message'); + finish(); + }; + + pq.on('readable', onReadable); + pq.startReader(); + + setTimeout(function () { + clientSockets.forEach(function (socket) { + socket.resetAndDestroy(); + }); + }, 100); + }); + }); +}); From 5c96d5b7953b7cd5474243b49d450965c5f6fa7a Mon Sep 17 00:00:00 2001 From: Nigro Simone Date: Wed, 12 Aug 2026 06:01:22 +0200 Subject: [PATCH 2/2] Close the proxy upstream sockets so the test process can exit --- test/poll-error.js | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/test/poll-error.js b/test/poll-error.js index e3730d8..cc901e9 100644 --- a/test/poll-error.js +++ b/test/poll-error.js @@ -11,15 +11,18 @@ describe('poll error', function () { var proxy; var proxyPort; var clientSockets; + var sockets; beforeEach(function (done) { clientSockets = []; + sockets = []; proxy = net.createServer(function (client) { var upstream = net.connect( Number(process.env.PGPORT || 5432), process.env.PGHOST || 'localhost' ); clientSockets.push(client); + sockets.push(client, upstream); client.pipe(upstream); upstream.pipe(client); client.on('error', function () {}); @@ -32,6 +35,11 @@ describe('poll error', function () { }); afterEach(function (done) { + // the reset does not reach the upstream socket through the pipe, so close every socket by + // hand: one left open keeps the event loop alive and mocha never exits + sockets.forEach(function (socket) { + socket.destroy(); + }); proxy.close(function () { done(); });