Force unlock timeout

README.md

@@ -35,6 +35,7 @@ var pool2 = new Pool({
   min: 4, // set min pool size to 4
   idleTimeoutMillis: 1000, // close idle clients after 1 second
   connectionTimeoutMillis: 1000, // return an error after 1 second if connection could not be established
+  forceUnlockTimeoutMillis: 1000 // force unlock a client taken from the pool after 1 second and return an error
 })
 
 //you can supply a custom client constructor
@@ -79,7 +80,7 @@ const pool = new Pool(config);
     ssl: true
   }
 */
-``` 
+```
 
 ### acquire clients with a promise
 
@@ -296,6 +297,31 @@ setTimeout(function () {
 
 ```
 
+#### forceUnlock
+
+Fired whenever the a client is forcefully unlocked by `forceUnlockTimeoutMillis`
+
+Example:
+
+This allows you to count the number of clients which have ever been acquired from the pool.
+
+```js
+var Pool = require('pg-pool')
+var pool = new Pool({ forceUnlockTimeoutMillis : 1000 })
+
+let forceUnlockCount = 0
+
+pool.on('forceUnlock', function (client) {
+  forceUnlockCount++;
+  console.trace('This should never happen');
+})
+
+setTimeout(function () {
+  console.log('force unlock count:', forceUnlockCount) // output: force unlock count: 1
+}, 1500)
+
+```
+
 ### environment variables
 
 pg-pool & node-postgres support some of the same environment variables as `psql` supports.  The most common are:

index.js

@@ -125,7 +125,8 @@ class Pool extends EventEmitter {
       const client = idleItem.client
       client.release = release.bind(this, client)
       this.emit('acquire', client)
-      return waiter.callback(undefined, client, client.release)
+
+      return this._callCallback(client, waiter.callback)
     }
     if (!this._isFull()) {
       return this.newClient(waiter)
@@ -148,6 +149,21 @@ class Pool extends EventEmitter {
     this.emit('remove', client)
   }
 
+  _callCallback (client, callback) {
+    if (!this.options.forceUnlockTimeoutMillis) {
+      return callback(undefined, client, client.release)
+    }
+
+    const tid = setTimeout(() => {
+      this.emit('forceUnlock', client)
+      client.release(new Error('Force unlock timeout hit'))
+    }, this.options.forceUnlockTimeoutMillis)
+    return callback(undefined, client, (err) => {
+      clearTimeout(tid)
+      client.release(err)
+    })
+  }
+
   connect (cb) {
     if (this.ending) {
       const err = new Error('Cannot use a pool after calling end on the pool')
@@ -248,9 +264,11 @@ class Pool extends EventEmitter {
         this.emit('acquire', client)
         if (!pendingItem.timedOut) {
           if (this.options.verify) {
-            this.options.verify(client, pendingItem.callback)
+            this._callCallback(client, (_, client) => {
+              this.options.verify(client, pendingItem.callback)
+            })
           } else {
-            pendingItem.callback(undefined, client, client.release)
+            this._callCallback(client, pendingItem.callback)
           }
         } else {
           if (this.options.verify) {

test/connection-timeout.js

@@ -217,4 +217,71 @@ describe('connection timeout', () => {
       })
     })
   })
+
+  it('force-unlocks new connections', (done) => {
+    const pool = new Pool({
+      connectionTimeoutMillis: 2000,
+      forceUnlockTimeoutMillis: 1000,
+      max: 1
+    })
+
+    // First connect with a slow query
+    pool.connect((err, client, release) => {
+      expect(err).to.be(undefined)
+
+      client.query('select pg_sleep(2)', (err) => {
+        expect(err).to.be.an(Error)
+      })
+    })
+
+    // Second connect waiting
+    pool.connect((err, client, release) => {
+      expect(err).to.be(undefined)
+
+      client.query('select $1::text as name', ['brianc'], (err, res) => {
+        expect(err).not.to.be.ok()
+        expect(res.rows).to.have.length(1)
+
+        release()
+        pool.end(done)
+      })
+    })
+  })
+
+  it('force-unlocks queued connections', (done) => {
+    const pool = new Pool({
+      connectionTimeoutMillis: 2000,
+      forceUnlockTimeoutMillis: 1000,
+      max: 1
+    })
+
+    // First connect with a slow query
+    pool.connect((err, client, release) => {
+      expect(err).to.be(undefined)
+
+      // Slow pending item next
+      pool.connect((err, client, release) => {
+        expect(err).to.be(undefined)
+
+        client.query('select pg_sleep(2)', (err) => {
+          expect(err).to.be.an(Error)
+        })
+      })
+
+      // Fast query queued afterwards
+      pool.connect((err, client, release) => {
+        expect(err).to.be(undefined)
+
+        client.query('select $1::text as name', ['brianc'], (err, res) => {
+          expect(err).not.to.be.ok()
+          expect(res.rows).to.have.length(1)
+
+          release()
+          pool.end(done)
+        })
+      })
+
+      release()
+    })
+  })
 })