POC: Time out pg.connect() call

Currently if you call pg.connect(), the call will block indefinitely until a
connection becomes available. In many cases, if a connection is not available
after some period of time, it's preferable to return an error (and call
control) to the client, instead of tying up resources forever.

Blocking on resource checkout also makes it easier for clients to deadlock -
recently at Shyp, we had a situation where a row got locked and the thread
that could unlock it was blocked waiting for a connection to become available,
leading to deadlock. In that situation, it would be better to abort the
checkout, which would have errored, but also broken the deadlock.

Add two new settings to defaults: `block`, which, if false, will immediately
return an error if the pool is full when you attempt to check out a new
connection. Also adds `acquireTimeout`, which will wait for `acquireTimeout`
milliseconds before giving up and returning an error.

This builds on two pull requests against `generic-pool`:

- Support options.block: coopernurse/node-pool#125
- Support options.timeout: coopernurse/node-pool#127

For the moment I'm pointing `generic-pool` at a branch that incorporates
both of these commits. I'm marking this as a proof-of-concept until those go
through, which hopefully they will soon. I'd also like feedback on the
API.

Adds semicolons in many places that omitted them and fixes several typos. I'm
happy to pull those out into a different commit.

Sets the TZ=GMT environment variable before running the tests; without this
value set, and with a Postgres server set to the America/Los_Angeles timezone,
a timezone test failed.

Fixes brianc#782 and brianc#805. Will help alleviate brianc#902. May help with brianc#397.

Author: Kevin Burke

Makefile

@@ -4,10 +4,10 @@ connectionString=postgres://
 
 params := $(connectionString)
 
-node-command := xargs -n 1 -I file node file $(params)
+node-command := TZ=GMT xargs -n 1 -I file node file $(params)
 
 .PHONY : test test-connection test-integration bench test-native \
-	 jshint publish test-missing-native update-npm
+	 jshint publish test-missing-native update-npm clean
 
 all:
 	npm install
@@ -67,3 +67,6 @@ prepare-test-db:
 jshint:
 	@echo "***Starting jshint***"
 	@./node_modules/.bin/jshint lib
+
+clean:
+	@rm -rf node_modules

lib/defaults.js

@@ -30,6 +30,24 @@ var defaults = module.exports = {
   //from the pool and destroyed
   poolIdleTimeout: 30000,
 
+  // block sets the behavior of node-postgres when the connection pool is
+  // full. Set to false to immediately return an error if no connections are
+  // available. By default, node-postgres will wait an infinite amount of time
+  // for a connection to become available.
+  block: true,
+
+  // acquireTimeout is the amount of time in ms to wait to check out a
+  // database connection. By default, node-postgres will wait infinitely for
+  // a connection to become available. If the acquireTimeout is negative, zero,
+  // or NaN, pg.connect will return an error immediately.
+  //
+  // If block is set to false, this setting is ignored - node-postgres will
+  // error immediately if block is false and the pool is full.
+  //
+  // The timeout message will be "Cannot acquire resource because the pool is
+  // full"
+  acquireTimeout: undefined,
+
   //frequency to check for idle clients within the client pool
   reapIntervalMillis: 1000,
 

lib/pool.js

@@ -70,6 +70,17 @@ var pools = {
         pool[key] = EventEmitter.prototype[key];
       }
     }
+    var block = true;
+    if (defaults.block === false || clientConfig.block === false) {
+      block = false;
+    }
+    var timeout;
+    if (typeof defaults.acquireTimeout === "number" && !isNaN(defaults.acquireTimeout)) {
+      timeout = defaults.acquireTimeout;
+    } else if (typeof clientConfig.acquireTimeout === "number" && !isNaN(clientConfig.acquireTimeout)) {
+      timeout = clientConfig.acquireTimeout;
+    }
+
     //monkey-patch with connect method
     pool.connect = function(cb) {
       var domain = process.domain;
@@ -86,7 +97,7 @@ var pools = {
             pool.release(client);
           }
         });
-      });
+      }, 0, {block: block, timeout: timeout});
     };
     return pool;
   }

package.json

@@ -19,7 +19,7 @@
   "main": "./lib",
   "dependencies": {
     "buffer-writer": "1.0.1",
-    "generic-pool": "2.1.1",
+    "generic-pool": "git://github.com/Shyp/node-pool.git#timeout",
     "packet-reader": "0.2.0",
     "pg-connection-string": "0.1.3",
     "pg-types": "1.*",

script/test-connection.js

@@ -9,7 +9,7 @@ pg.connect(helper.config, function(err, client, done) {
     console.error(err);
     process.exit(255);
   }
-  console.log("Checking for existance of required test table 'person'")
+  console.log("Checking for existence of required test table 'person'")
   client.query("SELECT COUNT(name) FROM person", function(err, callback) {
     if(err != null) {
       console.error("Recieved error when executing query 'SELECT COUNT(name) FROM person'")

test/integration/connection-pool/error-tests.js

@@ -17,7 +17,7 @@ pg.connect(helper.config, assert.success(function(client, done) {
         var params = ['idle'];
         if(!isGreater) {
           killIdleQuery = 'SELECT procpid, (SELECT pg_terminate_backend(procpid)) AS killed FROM pg_stat_activity WHERE current_query LIKE $1';
-          params = ['%IDLE%']
+          params = ['%IDLE%'];
         }
 
         //subscribe to the pg error event
@@ -37,5 +37,5 @@ pg.connect(helper.config, assert.success(function(client, done) {
       }));
     }));
 
-  })
+  });
 }));

test/test-helper.js

@@ -3,7 +3,7 @@ assert = require('assert');
 
 var EventEmitter = require('events').EventEmitter;
 var sys = require('util');
-var BufferList = require(__dirname+'/buffer-list')
+var BufferList = require(__dirname+'/buffer-list');
 
 var Connection = require(__dirname + '/../lib/connection');
 
@@ -160,7 +160,7 @@ var expect = function(callback, timeout) {
       callback.apply(this, arguments)
     }
   } else {
-    throw new Error("Unsupported arrity " + callback.length);
+    throw new Error("Unsupported arity " + callback.length);
   }
 
 }
@@ -193,7 +193,7 @@ process.on('exit', function() {
 })
 
 process.on('uncaughtException', function(err) {
-  console.error("\n %s", err.stack || err.toString())
+  console.error("\n %s", err.stack || err.toString());
   //causes xargs to abort right away
   process.exit(255);
 });
@@ -248,5 +248,3 @@ module.exports = {
   setTimezoneOffset: setTimezoneOffset,
   resetTimezoneOffset: resetTimezoneOffset
 };
-
-

test/unit/client/query-queue-tests.js

@@ -46,7 +46,7 @@ test('drain', function() {
     test("emits drain", function() {
       process.nextTick(function() {
         assert.ok(raisedDrain);
-      })
+      });
     });
   });
 });

test/unit/pool/basic-tests.js

@@ -5,28 +5,30 @@ var libDir = __dirname + '/../../../lib';
 var defaults = require(libDir + '/defaults');
 var pools = require(libDir + '/pool');
 var poolId = 0;
+var pg = require(libDir);
+pg = pg;
 
 require(__dirname + '/../../test-helper');
 
 var FakeClient = function() {
   EventEmitter.call(this);
-}
+};
 
 util.inherits(FakeClient, EventEmitter);
 
 FakeClient.prototype.connect = function(cb) {
   process.nextTick(cb);
-}
+};
 
 FakeClient.prototype.end = function() {
   this.endCalled = true;
-}
+};
 
 //Hangs the event loop until 'end' is called on client
 var HangingClient = function(config) {
   EventEmitter.call(this);
   this.config = config;
-}
+};
 
 util.inherits(HangingClient, EventEmitter);
 
@@ -35,11 +37,11 @@ HangingClient.prototype.connect = function(cb) {
     console.log('hung client...');
   }, 1000);
   process.nextTick(cb);
-}
+};
 
 HangingClient.prototype.end = function() {
   clearInterval(this.intervalId);
-}
+};
 
 pools.Client = FakeClient;
 
@@ -153,7 +155,7 @@ test('pool with connection error on connection', function() {
   });
 });
 
-test('returnning an error to done()', function() {
+test('returning an error to done()', function() {
   var p = pools.getOrCreate(poolId++);
   pools.Client = FakeClient;
   p.connect(function(err, client, done) {
@@ -179,7 +181,6 @@ test('fetching pool by object', function() {
   assert.equal(p, p2);
 });
 
-
 test('pool#connect client.poolCount', function() {
   var p = pools.getOrCreate(poolId++);
   var tid;
@@ -188,7 +189,7 @@ test('pool#connect client.poolCount', function() {
     tid = setTimeout(function() {
       throw new Error("Connection callback was never called");
     }, 100);
-  }
+  };
 
   setConnectTimeout();
   p.connect(function(err, client, done) {
@@ -210,6 +211,66 @@ test('pool#connect client.poolCount', function() {
         assert.equal(client.poolCount, undefined,
           'after pool is destroyed, count should be undefined');
       });
-    })
+    });
+  });
+});
+
+pg.defaults.block = false;
+pg.defaults.poolSize = 1;
+
+test('pool#connect acquire errors if the pool is full', function() {
+  var p = pools.getOrCreate(poolId++);
+  p.connect(function(err, client, done1) {
+    p.connect(function(err, client, done2) {
+      assert.equal(err.message, "Cannot acquire resource because the pool is full");
+    });
+  });
+});
+
+pg.defaults.block = true;
+pg.defaults.poolSize = 1;
+
+test('pool#connect accepts block setting via the clientConfig', function() {
+  var p = pools.getOrCreate({id: poolId++, block: false});
+  p.connect(function(err, client, done1) {
+    p.connect(function(err, client, done2) {
+      assert.equal(err.message, "Cannot acquire resource because the pool is full");
+    });
+  });
+});
+
+pg.defaults.block = true;
+pg.defaults.poolSize = 1;
+pg.defaults.acquireTimeout = 10;
+pg.defaults.poolIdleTimeout = 10;
+
+test('pool#connect acquire errors if acquisition times out', function() {
+  var p = pools.getOrCreate(poolId++);
+  p.connect(function(err, client, done1) {
+    setTimeout(function() {
+      done1();
+    }, 40);
+
+    var start = Date.now();
+    p.connect(function(err, client, done2) {
+      assert.equal(err.message, "Cannot acquire resource because the pool is full");
+      assert.ok((Date.now() - start) < 15);
+    });
+  });
+});
+
+pg.defaults.acquireTimeout = 20;
+
+test('pool#connect acquire returns a resource if one becomes available before the timeout', function() {
+  var p = pools.getOrCreate(poolId++);
+  p.connect(function(err, client, done1) {
+    setTimeout(function() {
+      done1();
+    }, 10);
+
+    p.connect(function(err, client, done2) {
+      assert.equal(err, null);
+      done2();
+    });
   });
 });

test/unit/pool/timeout-tests.js

@@ -10,17 +10,17 @@ require(__dirname + '/../../test-helper');
 
 var FakeClient = function() {
   EventEmitter.call(this);
-}
+};
 
 util.inherits(FakeClient, EventEmitter);
 
 FakeClient.prototype.connect = function(cb) {
   process.nextTick(cb);
-}
+};
 
 FakeClient.prototype.end = function() {
   this.endCalled = true;
-}
+};
 
 defaults.poolIdleTimeout = 10;
 defaults.reapIntervalMillis = 10;