Add "clientTtl" option to close and remove connections from the pool after a specified time. (#4175)

Co-authored-by: Carlos Fuentes <me@metcoder.dev>

Author: Dolan Halbrook

docs/docs/api/Pool.md

@@ -19,6 +19,7 @@ Extends: [`ClientOptions`](/docs/docs/api/Client.md#parameter-clientoptions)
 
 * **factory** `(origin: URL, opts: Object) => Dispatcher` - Default: `(origin, opts) => new Client(origin, opts)`
 * **connections** `number | null` (optional) - Default: `null` - The number of `Client` instances to create. When set to `null`, the `Pool` instance will create an unlimited amount of `Client` instances.
+* **clientTtl** `number | null` (optional) - Default: `null` - The amount of time before a `Client` instance is removed from the `Pool` and closed.   When set to `null`, `Client` instances will not be removed or closed based on age.
 
 ## Instance Properties
 

lib/dispatcher/pool.js

@@ -5,7 +5,8 @@ const {
   kClients,
   kNeedDrain,
   kAddClient,
-  kGetDispatcher
+  kGetDispatcher,
+  kRemoveClient
 } = require('./pool-base')
 const Client = require('./client')
 const {
@@ -35,6 +36,7 @@ class Pool extends PoolBase {
     autoSelectFamily,
     autoSelectFamilyAttemptTimeout,
     allowH2,
+    clientTtl,
     ...options
   } = {}) {
     if (connections != null && (!Number.isFinite(connections) || connections < 0)) {
@@ -65,12 +67,20 @@ class Pool extends PoolBase {
 
     this[kConnections] = connections || null
     this[kUrl] = util.parseOrigin(origin)
-    this[kOptions] = { ...util.deepClone(options), connect, allowH2 }
+    this[kOptions] = { ...util.deepClone(options), connect, allowH2, clientTtl }
     this[kOptions].interceptors = options.interceptors
       ? { ...options.interceptors }
       : undefined
     this[kFactory] = factory
 
+    this.on('connect', (origin, targets) => {
+      if (clientTtl != null && clientTtl > 0) {
+        for (const target of targets) {
+          Object.assign(target, { ttl: Date.now() })
+        }
+      }
+    })
+
     this.on('connectionError', (origin, targets, error) => {
       // If a connection error occurs, we remove the client from the pool,
       // and emit a connectionError event. They will not be re-used.
@@ -87,8 +97,12 @@ class Pool extends PoolBase {
   }
 
   [kGetDispatcher] () {
+    const clientTtlOption = this[kOptions].clientTtl
     for (const client of this[kClients]) {
-      if (!client[kNeedDrain]) {
+      // check ttl of client and if it's stale, remove it from the pool
+      if (clientTtlOption != null && clientTtlOption > 0 && client.ttl && ((Date.now() - client.ttl) > clientTtlOption)) {
+        this[kRemoveClient](client)
+      } else if (!client[kNeedDrain]) {
         return client
       }
     }

test/pool.js

@@ -578,6 +578,56 @@ test('pool connect', async (t) => {
   await t.completed
 })
 
+test('pool connect with clientTtl specified', async (t) => {
+  t = tspl(t, { plan: 1 })
+
+  const server = createServer({ joinDuplicateHeaders: true }, (c) => {
+    t.fail()
+  })
+  server.on('connect', (req, socket, firstBodyChunk) => {
+    socket.write('HTTP/1.1 200 Connection established\r\n\r\n')
+
+    let data = firstBodyChunk.toString()
+    socket.on('data', (buf) => {
+      data += buf.toString()
+    })
+
+    socket.on('end', () => {
+      socket.end(data)
+    })
+  })
+  after(() => server.close())
+
+  server.listen(0, async () => {
+    const client = new Pool(`http://localhost:${server.address().port}`, {
+      clientTtl: 10
+    })
+
+    const { socket } = await client.connect({
+      path: '/'
+    })
+
+    t.strictEqual(socket.closed, false)
+
+    let recvData = ''
+    socket.on('data', (d) => {
+      recvData += d
+    })
+
+    socket.on('end', () => {
+      t.strictEqual(recvData.toString(), 'Body')
+    })
+
+    socket.write('Body')
+    socket.end()
+
+    await new Promise(resolve => setTimeout(resolve, 10))
+    t.strictEqual(socket.closed, true)
+  })
+
+  await t.completed
+})
+
 test('pool dispatch', async (t) => {
   t = tspl(t, { plan: 2 })
 

types/pool.d.ts

@@ -33,6 +33,8 @@ declare namespace Pool {
     factory?(origin: URL, opts: object): Dispatcher;
     /** The max number of clients to create. `null` if no limit. Default `null`. */
     connections?: number | null;
+    /** The amount of time before a client is removed from the pool and closed. `null` if no time limit. Default `null` */
+    clientTtl?: number | null;
 
     interceptors?: { Pool?: readonly Dispatcher.DispatchInterceptor[] } & Client.Options['interceptors']
   }