http: do not reuse expired free sockets

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])`
 

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;

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);