Add connection & query timeout if all clients are checked out (brianc#70)

* Add connection & query timeout if all clients are checked out

This addresses [pg#1390](brianc#1390).

Ensure connection timeout applies both for new connections and on an exhuasted pool.  I also made the library return an error when passing a function as the first param to `pool.query` - previosuly this threw a sync type error.

* Add pg-cursor to dev deps

Author: brianc

index.js

@@ -128,14 +128,34 @@ class Pool extends EventEmitter {
       const err = new Error('Cannot use a pool after calling end on the pool')
       return cb ? cb(err) : this.Promise.reject(err)
     }
+
+    // if we don't have to connect a new client, don't do so
     if (this._clients.length >= this.options.max || this._idle.length) {
       const response = promisify(this.Promise, cb)
       const result = response.result
-      this._pendingQueue.push(response.callback)
+
       // if we have idle clients schedule a pulse immediately
       if (this._idle.length) {
         process.nextTick(() => this._pulseQueue())
       }
+
+      if (!this.options.connectionTimeoutMillis) {
+        this._pendingQueue.push(response.callback)
+        return result
+      }
+
+      // set connection timeout on checking out an existing client
+      const tid = setTimeout(() => {
+        // remove the callback from pending waiters because
+        // we're going to call it with a timeout error
+        this._pendingQueue = this._pendingQueue.filter(cb => cb === response.callback)
+        response.callback(new Error('timeout exceeded when trying to connect'))
+      }, this.options.connectionTimeoutMillis)
+
+      this._pendingQueue.push(function (err, res, done) {
+        clearTimeout(tid)
+        response.callback(err, res, done)
+      })
       return result
     }
 
@@ -199,6 +219,16 @@ class Pool extends EventEmitter {
   }
 
   query (text, values, cb) {
+    // guard clause against passing a function as the first parameter
+    if (typeof text === 'function') {
+      const response = promisify(this.Promise, text)
+      setImmediate(function () {
+        return response.callback(new Error('Passing a function as the first parameter to pool.query is not supported'))
+      })
+      return response.result
+    }
+
+    // allow plain text query without values
     if (typeof values === 'function') {
       cb = values
       values = undefined

package.json

@@ -7,7 +7,7 @@
     "test": "test"
   },
   "scripts": {
-    "test": "node_modules/.bin/standard && node_modules/.bin/mocha"
+    "test": " node_modules/.bin/mocha && node_modules/.bin/standard"
   },
   "repository": {
     "type": "git",
@@ -32,6 +32,7 @@
     "lodash": "4.13.1",
     "mocha": "^2.3.3",
     "pg": "*",
+    "pg-cursor": "^1.3.0",
     "standard": "7.1.2",
     "standard-format": "2.2.1"
   },

test/connection-timeout.js

@@ -58,5 +58,50 @@ describe('connection timeout', () => {
     }
     expect(errors).to.have.length(15)
   }.bind(this)))
-})
 
+  it('should timeout on checkout of used connection', (done) => {
+    const pool = new Pool({ connectionTimeoutMillis: 100, max: 1 })
+    pool.connect((err, client, release) => {
+      expect(err).to.be(undefined)
+      expect(client).to.not.be(undefined)
+      pool.connect((err, client) => {
+        expect(err).to.be.an(Error)
+        expect(client).to.be(undefined)
+        release()
+        pool.end(done)
+      })
+    })
+  })
+
+  it('should timeout on query if all clients are busy', (done) => {
+    const pool = new Pool({ connectionTimeoutMillis: 100, max: 1 })
+    pool.connect((err, client, release) => {
+      expect(err).to.be(undefined)
+      expect(client).to.not.be(undefined)
+      pool.query('select now()', (err, result) => {
+        expect(err).to.be.an(Error)
+        expect(result).to.be(undefined)
+        release()
+        pool.end(done)
+      })
+    })
+  })
+
+  it('should recover from timeout errors', (done) => {
+    const pool = new Pool({ connectionTimeoutMillis: 100, max: 1 })
+    pool.connect((err, client, release) => {
+      expect(err).to.be(undefined)
+      expect(client).to.not.be(undefined)
+      pool.query('select now()', (err, result) => {
+        expect(err).to.be.an(Error)
+        expect(result).to.be(undefined)
+        release()
+        pool.query('select $1::text as name', ['brianc'], (err, res) => {
+          expect(err).to.be(undefined)
+          expect(res.rows).to.have.length(1)
+          pool.end(done)
+        })
+      })
+    })
+  })
+})

test/error-handling.js

@@ -112,6 +112,17 @@ describe('pool error handling', function () {
     }))
   })
 
+  describe('passing a function to pool.query', () => {
+    it('calls back with error', (done) => {
+      const pool = new Pool()
+      console.log('passing fn to query')
+      pool.query((err) => {
+        expect(err).to.be.an(Error)
+        pool.end(done)
+      })
+    })
+  })
+
   describe('pool with lots of errors', () => {
     it('continues to work and provide new clients', co.wrap(function * () {
       const pool = new Pool({ max: 1 })

test/submittable.js

@@ -0,0 +1,19 @@
+'use strict'
+const Cursor = require('pg-cursor')
+const expect = require('expect.js')
+const describe = require('mocha').describe
+const it = require('mocha').it
+
+const Pool = require('../')
+
+describe('submittle', () => {
+  it('is returned from the query method', false, (done) => {
+    const pool = new Pool()
+    const cursor = pool.query(new Cursor('SELECT * from generate_series(0, 1000)'))
+    cursor.read((err, rows) => {
+      expect(err).to.be(undefined)
+      expect(!!rows).to.be.ok()
+      cursor.close(done)
+    })
+  })
+})