http: fix socket re-use races

doc/api/http.md

@@ -239,6 +239,9 @@ added: v0.11.4
 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'`.
+
 ### `agent.getName(options)`
 <!-- YAML
 added: v0.11.4

lib/_http_agent.js

@@ -120,6 +120,12 @@ function Agent(options) {
           socket[async_id_symbol] = -1;
           socket._httpMessage = null;
           this.removeSocket(socket, options);
+
+          const agentTimeout = this.options.timeout || 0;
+          if (socket.timeout !== agentTimeout) {
+            socket.setTimeout(agentTimeout);
+          }
+
           freeSockets.push(socket);
         } else {
           // Implementation doesn't want to keep socket alive
@@ -202,12 +208,21 @@ Agent.prototype.addRequest = function addRequest(req, options, port/* legacy */,
     this.sockets[name] = [];
   }
 
-  const freeLen = this.freeSockets[name] ? this.freeSockets[name].length : 0;
+  const freeSockets = this.freeSockets[name];
+  let socket;
+  if (freeSockets) {
+    while (freeSockets.length && freeSockets[0].destroyed) {
+      freeSockets.shift();
+    }
+    socket = freeSockets.shift();
+    if (!freeSockets.length)
+      delete this.freeSockets[name];
+  }
+
+  const freeLen = freeSockets ? freeSockets.length : 0;
   const sockLen = freeLen + this.sockets[name].length;
 
-  if (freeLen) {
-    // We have a free socket, so use that.
-    const socket = this.freeSockets[name].shift();
+  if (socket) {
     // Guard against an uninitialized or user supplied Socket.
     const handle = socket._handle;
     if (handle && typeof handle.asyncReset === 'function') {
@@ -216,10 +231,6 @@ Agent.prototype.addRequest = function addRequest(req, options, port/* legacy */,
       socket[async_id_symbol] = handle.getAsyncId();
     }
 
-    // don't leak
-    if (!this.freeSockets[name].length)
-      delete this.freeSockets[name];
-
     this.reuseSocket(socket, req);
     setRequestSocket(this, req, socket);
     this.sockets[name].push(socket);
@@ -319,6 +330,20 @@ function installListeners(agent, s, options) {
   }
   s.on('close', onClose);
 
+  function onTimeout() {
+    debug('CLIENT socket onTimeout');
+
+    // Destroy if in free list.
+    // TODO(ronag): Always destroy, even if not in free list.
+    const sockets = agent.freeSockets;
+    for (const name of ObjectKeys(sockets)) {
+      if (sockets[name].includes(s)) {
+        return s.destroy();
+      }
+    }
+  }
+  s.on('timeout', onTimeout);
+
   function onRemove() {
     // We need this function for cases like HTTP 'upgrade'
     // (defined by WebSockets) where we need to remove a socket from the
@@ -327,6 +352,7 @@ function installListeners(agent, s, options) {
     agent.removeSocket(s, options);
     s.removeListener('close', onClose);
     s.removeListener('free', onFree);
+    s.removeListener('timeout', onTimeout);
     s.removeListener('agentRemove', onRemove);
   }
   s.on('agentRemove', onRemove);
@@ -409,14 +435,6 @@ function setRequestSocket(agent, req, socket) {
     return;
   }
   socket.setTimeout(req.timeout);
-  // Reset timeout after response end
-  req.once('response', (res) => {
-    res.once('end', () => {
-      if (socket.timeout !== agentTimeout) {
-        socket.setTimeout(agentTimeout);
-      }
-    });
-  });
 }
 
 function emitErrorNT(emitter, err) {

test/parallel/test-http-agent-timeout-option.js

@@ -18,6 +18,6 @@ request.on('socket', mustCall((socket) => {
 
   const listeners = socket.listeners('timeout');
 
-  strictEqual(listeners.length, 1);
-  strictEqual(listeners[0], request.timeoutCb);
+  strictEqual(listeners.length, 2);
+  strictEqual(listeners[1], request.timeoutCb);
 }));

test/parallel/test-http-agent-timeout.js

@@ -0,0 +1,96 @@
+'use strict';
+
+const common = require('../common');
+const assert = require('assert');
+const http = require('http');
+
+{
+  // Ensure reuse of successful sockets.
+
+  const agent = new http.Agent({ keepAlive: true });
+
+  const server = http.createServer((req, res) => {
+    res.end();
+  });
+
+  server.listen(0, common.mustCall(() => {
+    let socket;
+    http.get({ port: server.address().port, agent })
+      .on('response', common.mustCall((res) => {
+        socket = res.socket;
+        assert(socket);
+        res.resume();
+        socket.on('free', common.mustCall(() => {
+          http.get({ port: server.address().port, agent })
+            .on('response', common.mustCall((res) => {
+              assert.strictEqual(socket, res.socket);
+              assert(socket);
+              agent.destroy();
+              server.close();
+            }));
+        }));
+      }));
+  }));
+}
+
+{
+  // Ensure that timeouted sockets are not reused.
+
+  const agent = new http.Agent({ keepAlive: true, timeout: 50 });
+
+  const server = http.createServer((req, res) => {
+    res.end();
+  });
+
+  server.listen(0, common.mustCall(() => {
+    let socket;
+    http.get({ port: server.address().port, agent })
+      .on('response', common.mustCall((res) => {
+        socket = res.socket;
+        assert(socket);
+        res.resume();
+        socket.on('free', common.mustCall(() => {
+          socket.on('timeout', common.mustCall(() => {
+            http.get({ port: server.address().port, agent })
+              .on('response', common.mustCall((res) => {
+                assert.notStrictEqual(socket, res.socket);
+                assert.strictEqual(socket.destroyed, true);
+                assert(socket);
+                agent.destroy();
+                server.close();
+              }));
+          }));
+        }));
+      }));
+  }));
+}
+
+{
+  // Ensure that destroyed sockets are not reused.
+
+  const agent = new http.Agent({ keepAlive: true });
+
+  const server = http.createServer((req, res) => {
+    res.end();
+  });
+
+  server.listen(0, common.mustCall(() => {
+    let socket;
+    http.get({ port: server.address().port, agent })
+      .on('response', common.mustCall((res) => {
+        socket = res.socket;
+        assert(socket);
+        res.resume();
+        socket.on('free', common.mustCall(() => {
+          socket.destroy();
+          http.get({ port: server.address().port, agent })
+            .on('response', common.mustCall((res) => {
+              assert.notStrictEqual(socket, res.socket);
+              assert(socket);
+              agent.destroy();
+              server.close();
+            }));
+        }));
+      }));
+  }));
+}

test/parallel/test-http-client-set-timeout-after-end.js

@@ -20,7 +20,7 @@ server.listen(0, () => {
   const req = get({ agent, port }, (res) => {
     res.on('end', () => {
       strictEqual(req.setTimeout(0), req);
-      strictEqual(socket.listenerCount('timeout'), 0);
+      strictEqual(socket.listenerCount('timeout'), 1);
       agent.destroy();
       server.close();
     });

test/parallel/test-http-client-set-timeout.js

@@ -42,7 +42,7 @@ server.listen(0, mustCall(() => {
   }));
 
   req.on('timeout', mustCall(() => {
-    strictEqual(req.socket.listenerCount('timeout'), 0);
+    strictEqual(req.socket.listenerCount('timeout'), 1);
     req.destroy();
   }));
 }));

test/parallel/test-http-client-timeout-option-listeners.js

@@ -24,9 +24,9 @@ const options = {
 server.listen(0, options.host, common.mustCall(() => {
   options.port = server.address().port;
   doRequest(common.mustCall((numListeners) => {
-    assert.strictEqual(numListeners, 1);
+    assert.strictEqual(numListeners, 2);
     doRequest(common.mustCall((numListeners) => {
-      assert.strictEqual(numListeners, 1);
+      assert.strictEqual(numListeners, 2);
       server.close();
       agent.destroy();
     }));

test/parallel/test-http-client-timeout-option-with-agent.js

@@ -18,6 +18,6 @@ request.on('socket', mustCall((socket) => {
 
   const listeners = socket.listeners('timeout');
 
-  strictEqual(listeners.length, 1);
-  strictEqual(listeners[0], request.timeoutCb);
+  strictEqual(listeners.length, 2);
+  strictEqual(listeners[1], request.timeoutCb);
 }));