Avoid reusing destroying connection pool streams

lib/connect.js

@@ -44,7 +44,8 @@ module.exports = function connect(dht, publicKey, opts = {}) {
   )
   publicKey = key
 
-  if (pool && pool.has(publicKey)) return pool.get(publicKey)
+  const existing = pool && pool.get(publicKey)
+  if (existing) return existing
 
   publicKey = unslab(publicKey)
 

lib/connection-pool.js

@@ -60,29 +60,24 @@ module.exports = class ConnectionPool extends EventEmitter {
       return
     }
 
-    const session = new ConnectionRef(this, stream)
-
     const keyString = b4a.toString(stream.remotePublicKey, 'hex')
+    const session = new ConnectionRef(this, keyString, stream)
 
     if (opened) {
       this._connections.set(keyString, session)
 
-      stream.on('close', () => {
-        this._connections.delete(keyString)
-      })
+      stream.on('close', () => session.gc())
 
       this.emit('connection', stream, session)
     } else {
       this._connecting.set(keyString, session)
 
       stream
         .on('error', noop)
-        .on('close', () => {
-          if (opened) this._connections.delete(keyString)
-          else this._connecting.delete(keyString)
-        })
+        .on('close', () => session.gc())
         .on('open', () => {
           opened = true
+          if (this._connecting.get(keyString) !== session) return
 
           this._connecting.delete(keyString)
           this._connections.set(keyString, session)
@@ -105,23 +100,27 @@ module.exports = class ConnectionPool extends EventEmitter {
   }
 
   has(publicKey) {
-    const keyString = b4a.toString(publicKey, 'hex')
-
-    return this._connections.has(keyString) || this._connecting.has(keyString)
+    return this.get(publicKey) !== null
   }
 
   get(publicKey) {
     const keyString = b4a.toString(publicKey, 'hex')
 
     const existing = this._connections.get(keyString) || this._connecting.get(keyString)
+    if (!existing) return null
+    if (existing.destroying) {
+      existing.gc()
+      return null
+    }
 
-    return existing?._stream || null
+    return existing._stream
   }
 }
 
 class ConnectionRef {
-  constructor(pool, stream) {
+  constructor(pool, keyString, stream) {
     this._pool = pool
+    this.keyString = keyString
     this._stream = stream
     this._refs = 0
   }
@@ -137,6 +136,20 @@ class ConnectionRef {
   release() {
     this._stream.destroy()
   }
+
+  get destroying() {
+    return this._stream.destroying || this._stream.destroyed
+  }
+
+  gc() {
+    if (this._pool._connections.get(this.keyString) === this) {
+      this._pool._connections.delete(this.keyString)
+    }
+
+    if (this._pool._connecting.get(this.keyString) === this) {
+      this._pool._connecting.delete(this.keyString)
+    }
+  }
 }
 
 function noop() {}

test/pool.js

@@ -1,4 +1,5 @@
 const test = require('brittle')
+const { once } = require('events')
 const { swarm } = require('./helpers')
 
 test('connection pool, client side', async function (t) {
@@ -113,3 +114,33 @@ test('connection pool, client and server side', async function (t) {
 
   await server.close()
 })
+
+test('connection pool ignores destroying streams', async function (t) {
+  const [a, b] = await swarm(t)
+
+  const server = a.createServer((socket) => {
+    socket.on('error', noop).resume()
+  })
+
+  await server.listen()
+
+  const pool = b.pool()
+  const first = b.connect(server.publicKey, { pool })
+  first.on('error', noop)
+
+  await once(first, 'open')
+
+  first.destroy()
+  t.ok(first.destroying, 'pooled stream is destroying')
+
+  const second = b.connect(server.publicKey, { pool })
+  second.on('error', noop)
+
+  t.is(second === first, false, 'destroying stream is not reused')
+
+  second.destroy()
+
+  await server.close()
+})
+
+function noop() {}