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..cc901e9 --- /dev/null +++ b/test/poll-error.js @@ -0,0 +1,101 @@ +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; + 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 () {}); + upstream.on('error', function () {}); + }); + proxy.listen(0, '127.0.0.1', function () { + proxyPort = proxy.address().port; + done(); + }); + }); + + 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(); + }); + }); + + 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); + }); + }); +});