fix: replace stale pool clients under connection limit (#5145)

Author: trivikr

lib/dispatcher/pool-base.js

@@ -189,12 +189,12 @@ class PoolBase extends DispatcherBase {
   }
 
   [kRemoveClient] (client) {
-    client.close(() => {
-      const idx = this[kClients].indexOf(client)
-      if (idx !== -1) {
-        this[kClients].splice(idx, 1)
-      }
-    })
+    const idx = this[kClients].indexOf(client)
+    if (idx !== -1) {
+      this[kClients].splice(idx, 1)
+    }
+
+    client.close(() => {})
 
     this[kNeedDrain] = this[kClients].some(dispatcher => (
       !dispatcher[kNeedDrain] &&

lib/dispatcher/pool.js

@@ -97,10 +97,13 @@ class Pool extends PoolBase {
 
   [kGetDispatcher] () {
     const clientTtlOption = this[kOptions].clientTtl
-    for (const client of this[kClients]) {
+    for (let i = 0; i < this[kClients].length; i++) {
+      const client = this[kClients][i]
+
       // 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)
+        i--
       } else if (!client[kNeedDrain]) {
         return client
       }

lib/dispatcher/round-robin-pool.js

@@ -92,25 +92,24 @@ class RoundRobinPool extends PoolBase {
 
   [kGetDispatcher] () {
     const clientTtlOption = this[kOptions].clientTtl
-    const clientsLength = this[kClients].length
 
     // If we have no clients yet, create one
-    if (clientsLength === 0) {
+    if (this[kClients].length === 0) {
       const dispatcher = this[kFactory](this[kUrl], this[kOptions])
       this[kAddClient](dispatcher)
       return dispatcher
     }
 
     // Round-robin through existing clients
     let checked = 0
-    while (checked < clientsLength) {
-      this[kIndex] = (this[kIndex] + 1) % clientsLength
+    while (checked < this[kClients].length) {
+      this[kIndex] = (this[kIndex] + 1) % this[kClients].length
       const client = this[kClients][this[kIndex]]
 
       // Check if client is stale (TTL expired)
       if (clientTtlOption != null && clientTtlOption > 0 && client.ttl && ((Date.now() - client.ttl) > clientTtlOption)) {
         this[kRemoveClient](client)
-        checked++
+        this[kIndex]--
         continue
       }
 
@@ -123,7 +122,7 @@ class RoundRobinPool extends PoolBase {
     }
 
     // All clients are busy, create a new one if we haven't reached the limit
-    if (!this[kConnections] || clientsLength < this[kConnections]) {
+    if (!this[kConnections] || this[kClients].length < this[kConnections]) {
       const dispatcher = this[kFactory](this[kUrl], this[kOptions])
       this[kAddClient](dispatcher)
       return dispatcher

test/pool.js

@@ -652,6 +652,43 @@ test('pool connect with clientTtl specified', async (t) => {
   await t.completed
 })
 
+test('pool replaces stale client when connections limit is reached', async (t) => {
+  t = tspl(t, { plan: 2 })
+
+  const server = createServer((req, res) => {
+    res.end('ok')
+  })
+  after(() => server.close())
+
+  await new Promise(resolve => server.listen(0, resolve))
+
+  const client = new Pool(`http://localhost:${server.address().port}`, {
+    connections: 1,
+    clientTtl: 1
+  })
+  after(() => client.destroy())
+
+  async function request () {
+    const { statusCode, body } = await client.request({
+      path: '/',
+      method: 'GET'
+    })
+    await body.text()
+    return statusCode
+  }
+
+  t.strictEqual(await request(), 200)
+  await new Promise(resolve => setTimeout(resolve, 20))
+
+  const statusCode = await Promise.race([
+    request(),
+    new Promise((_resolve, reject) => {
+      setTimeout(() => reject(new Error('second request hung')), 1000)
+    })
+  ])
+  t.strictEqual(statusCode, 200)
+})
+
 test('pool dispatch', async (t) => {
   t = tspl(t, { plan: 2 })
 

test/round-robin-pool.js

@@ -96,6 +96,43 @@ test('basic get', async (t) => {
   await t.completed
 })
 
+test('replaces stale client when connections limit is reached', async (t) => {
+  t = tspl(t, { plan: 2 })
+
+  const server = createServer((req, res) => {
+    res.end('ok')
+  })
+  after(() => server.close())
+
+  await new Promise(resolve => server.listen(0, resolve))
+
+  const pool = new RoundRobinPool(`http://localhost:${server.address().port}`, {
+    connections: 1,
+    clientTtl: 1
+  })
+  after(() => pool.destroy())
+
+  async function request () {
+    const { statusCode, body } = await pool.request({
+      path: '/',
+      method: 'GET'
+    })
+    await body.text()
+    return statusCode
+  }
+
+  t.strictEqual(await request(), 200)
+  await new Promise(resolve => setTimeout(resolve, 20))
+
+  const statusCode = await Promise.race([
+    request(),
+    new Promise((_resolve, reject) => {
+      setTimeout(() => reject(new Error('second request hung')), 1000)
+    })
+  ])
+  t.strictEqual(statusCode, 200)
+})
+
 test('connect/disconnect event(s)', async (t) => {
   const clients = 2