Fix connections not being destroyed when released to purged pool

This scenario happens when the pull for a given address is purged while a connection stills in use and then another pool for the same address is created. In this case, the connection was being wrongly added to the existing pool.

The correct behaviour is destroy this orphan connections.

Author: bigmontz

packages/bolt-connection/src/pool/pool.js

@@ -62,6 +62,7 @@ class Pool {
     this._pendingCreates = {}
     this._acquireRequests = {}
     this._activeResourceCounts = {}
+    this._poolState = {}
     this._release = this._release.bind(this)
     this._log = log
     this._closed = false
@@ -189,10 +190,13 @@ class Pool {
 
     const key = address.asKey()
     let pool = this._pools[key]
+    let poolState = this._poolState[key]
     if (!pool) {
       pool = []
+      poolState = new PoolState()
       this._pools[key] = pool
       this._pendingCreates[key] = 0
+      this._poolState[key] = poolState
     }
     while (pool.length) {
       const resource = pool.pop()
@@ -231,7 +235,7 @@ class Pool {
     let resource
     try {
       // Invoke callback that creates actual connection
-      resource = await this._create(address, this._release)
+      resource = await this._create(address, (address, resource) => this._release(poolState, address, resource))
 
       resourceAcquired(key, this._activeResourceCounts)
       if (this._log.isDebugEnabled()) {
@@ -243,11 +247,11 @@ class Pool {
     return resource
   }
 
-  async _release (address, resource) {
+  async _release (poolState, address, resource) {
     const key = address.asKey()
     const pool = this._pools[key]
 
-    if (pool) {
+    if (pool && poolState.isActive()) {
       // there exist idle connections for the given key
       if (!this._validate(resource)) {
         if (this._log.isDebugEnabled()) {
@@ -295,19 +299,23 @@ class Pool {
 
   async _purgeKey (key) {
     const pool = this._pools[key] || []
+    const poolState = this._poolState[key] || new PoolState()
     while (pool.length) {
       const resource = pool.pop()
       if (this._removeIdleObserver) {
         this._removeIdleObserver(resource)
       }
       await this._destroy(resource)
     }
+    poolState.close()
     delete this._pools[key]
+    delete this._poolState[key]
   }
 
   _processPendingAcquireRequests (address) {
     const key = address.asKey()
     const requests = this._acquireRequests[key]
+    const poolState = this._poolState[key]
     if (requests) {
       const pendingRequest = requests.shift() // pop a pending acquire request
 
@@ -326,7 +334,7 @@ class Pool {
               if (pendingRequest.isCompleted()) {
                 // request has been completed, most likely failed by a timeout
                 // return the acquired resource back to the pool
-                this._release(address, resource)
+                this._release(poolState, address, resource)
               } else {
                 // request is still pending and can be resolved with the newly acquired resource
                 pendingRequest.resolve(resource) // resolve the pending request with the acquired resource
@@ -404,4 +412,18 @@ class PendingRequest {
   }
 }
 
+class PoolState {
+  constructor() {
+    this._active = true;
+  }
+
+  isActive() {
+    return this._active;
+  }
+
+  close() {
+    this._active = false;
+  }
+}
+
 export default Pool

packages/bolt-connection/test/pool/pool.test.js

@@ -237,6 +237,44 @@ describe('#unit Pool', () => {
     expect(r0.destroyed).toBeTruthy()
   })
 
+  it('destroys resource when key was purged event when key is acquired again', async () => {
+    let counter = 0
+    const address = ServerAddress.fromUrl('bolt://localhost:7687')
+    const pool = new Pool({
+      create: (server, release) =>
+        Promise.resolve(new Resource(server, counter++, release)),
+      destroy: res => {
+        res.destroyed = true
+        return Promise.resolve()
+      }
+    })
+
+    // Acquire resource
+    const r0 = await pool.acquire(address)
+    expect(pool.has(address)).toBeTruthy()
+    expect(r0.id).toEqual(0)
+
+    // Purging the key
+    await pool.purge(address)
+    expect(pool.has(address)).toBeFalsy()
+    expect(r0.destroyed).toBeFalsy()
+
+    // Acquiring second resolve should recreate the pool
+    const r1 = await pool.acquire(address)
+    expect(pool.has(address)).toBeTruthy()
+    expect(r1.id).toEqual(1)
+
+    // Closing the first resource should destroy it
+    await r0.close()
+    expect(pool.has(address)).toBeTruthy()
+    expect(r0.destroyed).toBeTruthy()
+
+    // Closing the second resource should not destroy it
+    await r1.close()
+    expect(pool.has(address)).toBeTruthy()
+    expect(r1.destroyed).toBeFalsy()
+  })
+
   it('close purges all keys', async () => {
     let counter = 0