diff --git a/doc/api/http.md b/doc/api/http.md index af939001c9e5..8859b25583e7 100644 --- a/doc/api/http.md +++ b/doc/api/http.md @@ -353,7 +353,10 @@ An object which contains arrays of sockets currently awaiting use by the agent when `keepAlive` is enabled. Do not modify. Sockets in the `freeSockets` list will be automatically destroyed and -removed from the array on `'timeout'`. +removed from the array on `'timeout'`. When a request would reuse a free +socket that has been in the list for at least its timeout, for example because +the event loop was blocked and `'timeout'` could not be emitted in time, the +agent destroys that socket instead. ### `agent.getName([options])` diff --git a/lib/internal/http/agent.js b/lib/internal/http/agent.js index 3871c5b56a2f..d01bb865621e 100644 --- a/lib/internal/http/agent.js +++ b/lib/internal/http/agent.js @@ -60,6 +60,9 @@ const { kReadBytesOrError, streamBaseState, } = internalBinding('stream_wrap'); +// Call getLibuvNow() on the binding: its fast API call reads the binding +// from the receiver. +const timersBinding = internalBinding('timers'); const { validateNumber, validateOneOf, @@ -72,6 +75,7 @@ const kOnKeylog = Symbol('onkeylog'); const kRequestOptions = Symbol('requestOptions'); const kRequestAsyncResource = Symbol('requestAsyncResource'); const kFreeSocketDataGuard = Symbol('freeSocketDataGuard'); +const kFreeSince = Symbol('freeSince'); // New Agent code. @@ -263,6 +267,7 @@ function Agent(options) { socket.once('error', freeSocketErrorListener); installFreeSocketDataGuard(socket); + socket[kFreeSince] = timersBinding.getLibuvNow(); freeSockets.push(socket); }); @@ -390,9 +395,18 @@ Agent.prototype.addRequest = function addRequest(req, options, port/* legacy */, while (freeSockets.length && freeSockets[0].destroyed) { freeSockets.shift(); } - socket = this.scheduling === 'fifo' ? - freeSockets.shift() : - freeSockets.pop(); + const now = timersBinding.getLibuvNow(); + while (freeSockets.length) { + socket = this.scheduling === 'fifo' ? + freeSockets.shift() : + freeSockets.pop(); + if (!socket.destroyed && !isFreeSocketExpired(socket, now)) + break; + debug('free socket destroyed or expired'); + socket.destroy(); + socket.emit('agentRemove'); + socket = undefined; + } if (!freeSockets.length) delete this.freeSockets[name]; } @@ -690,6 +704,17 @@ Agent.prototype.destroy = function destroy() { } }; +// A free socket is destroyed by its idle timeout timer, but timers cannot run +// while the event loop is blocked, so the socket can still be in the pool after +// its timeout has passed, when the server may have closed the connection. +// kFreeSince is read from the timers' clock when the socket enters the pool, +// after keepSocketAlive(). The socket's idle timer was last started then or +// earlier (at its last read or write), so this does not expire a socket before +// that timer is due. +function isFreeSocketExpired(socket, now) { + return socket.timeout > 0 && now - socket[kFreeSince] >= socket.timeout; +} + function setRequestSocket(agent, req, socket) { req.onSocket(socket); const agentTimeout = agent.options.timeout || 0; diff --git a/test/parallel/test-http-agent-keepalive-expired-free-socket.js b/test/parallel/test-http-agent-keepalive-expired-free-socket.js new file mode 100644 index 000000000000..aedc2dbdea15 --- /dev/null +++ b/test/parallel/test-http-agent-keepalive-expired-free-socket.js @@ -0,0 +1,189 @@ +// Flags: --expose-internals --no-warnings --allow-natives-syntax +'use strict'; + +const common = require('../common'); + +// A free socket expires by its idle timeout timer. When the event loop is +// blocked longer than that timeout, the timer cannot run, and the next +// request must not reuse the socket, because the server may already have +// closed it. +// Refs: https://github.com/nodejs/node/issues/54293 + +const assert = require('node:assert'); +const http = require('node:http'); +const { internalBinding } = require('internal/test/binding'); + +function count(lists) { + return Object.values(lists).reduce((n, list) => n + list.length, 0); +} + +// Makes `sockets` concurrent requests, then calls `callback` once all their +// sockets are in the agent's free pool. +function fillPool(agent, port, sockets, callback) { + let free = 0; + agent.on('free', function onFree() { + if (++free === sockets) { + agent.removeListener('free', onFree); + setImmediate(callback); + } + }); + for (let i = 0; i < sockets; i++) { + http.get({ port, agent }, common.mustCall((res) => res.resume())); + } +} + +function testReuse(agentOptions, blockMs, expectReuse, next) { + const server = http.createServer(common.mustCall((req, res) => { + res.end('ok'); + }, 2)); + // No Keep-Alive hint from the server, so the agent's `timeout` alone sets + // the socket's idle timeout. + server.keepAliveTimeout = 0; + + server.listen(0, common.mustCall(() => { + const agent = new http.Agent({ keepAlive: true, ...agentOptions }); + const { port } = server.address(); + + fillPool(agent, port, 1, common.mustCall(() => { + const [socket] = Object.values(agent.freeSockets)[0]; + assert.strictEqual(socket.timeout, agentOptions.timeout ?? 0); + common.sleepSync(blockMs); + const req = http.get({ port, agent }, common.mustCall((res) => { + res.resume(); + res.on('end', common.mustCall(() => { + agent.destroy(); + server.close(next); + })); + })); + assert.strictEqual(req.reusedSocket, expectReuse); + assert.strictEqual(socket.destroyed, !expectReuse); + })); + })); +} + +// The expired sockets release their slots at once, so a burst of requests +// after the block opens no more than maxSockets connections. +function testMaxSockets(next) { + const timeout = common.platformTimeout(500); + const server = http.createServer(common.mustCall((req, res) => { + res.end('ok'); + }, 6)); + + server.listen(0, common.mustCall(() => { + const agent = new http.Agent({ keepAlive: true, timeout, maxSockets: 2 }); + const { port } = server.address(); + + fillPool(agent, port, 2, common.mustCall(() => { + const pooled = [...Object.values(agent.freeSockets)[0]]; + common.sleepSync(timeout * 2); + let done = 0; + for (let i = 0; i < 4; i++) { + http.get({ port, agent }, common.mustCall((res) => { + res.resume(); + res.on('end', () => { + if (++done === 4) { + agent.destroy(); + server.close(next); + } + }); + })); + } + assert(pooled.every((socket) => socket.destroyed)); + assert.strictEqual(agent.totalSocketCount, 2); + assert.strictEqual(count(agent.sockets), 2); + assert.strictEqual(count(agent.requests), 2); + })); + })); +} + +// A destroyed socket can stay in the free pool until its 'close' event. A +// request made before that event must not get it. +function testDestroyedSocket(next) { + const server = http.createServer(common.mustCall((req, res) => { + res.end('ok'); + }, 3)); + // No Keep-Alive hint from the server, so the pooled sockets have no idle + // timeout and cannot expire. + server.keepAliveTimeout = 0; + + server.listen(0, common.mustCall(() => { + const agent = new http.Agent({ keepAlive: true }); + const { port } = server.address(); + + fillPool(agent, port, 2, common.mustCall(() => { + const [first, last] = Object.values(agent.freeSockets)[0]; + last.destroy(); + const req = http.get({ port, agent }, common.mustCall((res) => { + assert.strictEqual(req.socket, first); + res.resume(); + res.on('end', common.mustCall(() => { + agent.destroy(); + server.close(next); + })); + })); + assert.strictEqual(req.reusedSocket, true); + assert.strictEqual(agent.totalSocketCount, 1); + assert.strictEqual(count(agent.freeSockets), 0); + assert.strictEqual(count(agent.sockets), 1); + })); + })); +} + +// addRequest() reads the timers' clock through a fast API call when it takes a +// socket from the pool; the optimized function must be able to make it. +function testOptimizedAddRequest(next) { + const server = http.createServer(common.mustCall((req, res) => { + res.end('ok'); + }, 3)); + + server.listen(0, common.mustCall(() => { + const agent = new http.Agent({ keepAlive: true, timeout: 60_000 }); + const { port } = server.address(); + + function get(callback) { + http.get({ port, agent }, common.mustCall((res) => { + res.resume(); + res.on('end', common.mustCall(() => setImmediate(callback))); + })); + } + + eval('%PrepareFunctionForOptimization(http.Agent.prototype.addRequest)'); + get(common.mustCall(() => get(common.mustCall(() => { + eval('%OptimizeFunctionOnNextCall(http.Agent.prototype.addRequest)'); + const fastCalls = getFastCallCount(); + get(common.mustCall(() => { + agent.destroy(); + server.close(next); + })); + if (common.isDebug) { + assert.strictEqual(getFastCallCount() - fastCalls, 1); + } + })))); + })); +} + +// The fast API call counter exists in debug builds only. +function getFastCallCount() { + if (!common.isDebug) return 0; + const { getV8FastApiCallCount } = internalBinding('debug'); + return getV8FastApiCallCount('timers.getLibuvNow'); +} + +// The cases run one after another, so that one case's block does not delay +// the timers of another. +const cases = [ + // The idle timeout passed during the block. + (next) => testReuse({ timeout: 100 }, 300, false, next), + (next) => testReuse({ timeout: 100, scheduling: 'fifo' }, 300, false, next), + // The idle timeout did not pass during the block. + (next) => testReuse({ timeout: 60_000 }, 300, true, next), + // A socket without an idle timeout is not expired by the clock. + (next) => testReuse({}, 300, true, next), + testMaxSockets, + testDestroyedSocket, + testOptimizedAddRequest, +]; + +(function run(i) { + if (i < cases.length) cases[i](common.mustCall(() => run(i + 1))); +})(0);