test to reproduce behavior of issue brianc/node-postgres#549

lib/client.js

@@ -176,13 +176,13 @@ Client.prototype.connect = function(callback) {
   con.once('end', function() {
     if ( callback ) {
       // haven't received a connection message yet !
-      var err = new Error("Stream unexpectedly ended before getting ready for query execution");
+      var err = new Error('Connection was ended during query');
       callback(err);
       callback = null;
       return;
     }
     if(self.activeQuery) {
-      var disconnectError = new Error('Stream unexpectedly ended during query execution');
+      var disconnectError = new Error('Connection was ended during query');
       self.activeQuery.handleError(disconnectError, con);
       self.activeQuery = null;
     }

lib/connection.js

@@ -285,14 +285,15 @@ Connection.prototype.sync = function() {
   this.writer.flush(0);
 
   this.writer.add(emptyBuffer);
+  this._ending = true;
   this._send(0x53);
 };
 
 Connection.prototype.end = function() {
   //0x58 = 'X'
   this.writer.add(emptyBuffer);
-  this._send(0x58);
   this._ending = true;
+  this._send(0x58);
 };
 
 Connection.prototype.describe = function(msg, more) {

test/integration/client/copy-tests.js

@@ -8,7 +8,8 @@ var prepareTable = function (client, callback) {
   client.query(
     'CREATE TEMP TABLE copy_test (id SERIAL, name CHARACTER VARYING(10), age INT)',
     assert.calls(function (err, result) {
-      assert.equal(err, null, "create table query should not fail");
+      assert.equal(err, null,
+        err && err.message ? "create table query should not fail: " + err.message : null);
       callback();
     })
   );
@@ -19,7 +20,7 @@ test('COPY FROM', function () {
     prepareTable(client, function () {
       var stream = client.copyFrom("COPY  copy_test (name, age)  FROM stdin WITH CSV");
       stream.on('error', function (error) {
-        assert.ok(false, "COPY FROM stream should not emit errors" + helper.sys.inspect(error)); 
+        assert.ok(false, "COPY FROM stream should not emit errors" + helper.sys.inspect(error));
       });
       for (var i = 0; i < ROWS_TO_INSERT; i++) {
         stream.write( String(Date.now() + Math.random()).slice(0,10) + ',' + i + '\n');
@@ -44,11 +45,11 @@ test('COPY TO', function () {
       var stream = client.copyTo("COPY  person (id, name, age)  TO stdin WITH CSV");
       var  buf = new Buffer(0);
       stream.on('error', function (error) {
-        assert.ok(false, "COPY TO stream should not emit errors" + helper.sys.inspect(error)); 
+        assert.ok(false, "COPY TO stream should not emit errors" + helper.sys.inspect(error));
       });
       assert.emits(stream, 'data', function (chunk) {
-        buf = Buffer.concat([buf, chunk]);  
-      }, "COPY IN stream should emit data event for each row"); 
+        buf = Buffer.concat([buf, chunk]);
+      }, "COPY IN stream should emit data event for each row");
       assert.emits(stream, 'end', function () {
         var lines = buf.toString().split('\n');
         assert.equal(lines.length >= 0, true, "copy in should return rows saved by copy from");
@@ -73,19 +74,19 @@ test('COPY TO, queue queries', function () {
       });
       var stream = client.copyTo("COPY  person (id, name, age)  TO stdin WITH CSV");
       //imitate long query, to make impossible,
-      //that copy query end callback runs after 
+      //that copy query end callback runs after
       //second query callback
       client.query("SELECT pg_sleep(1)", function () {
         query2Done = true;
         assert.ok(copyQueryDone && query2Done, "second query has to be executed after others");
       });
       var  buf = new Buffer(0);
       stream.on('error', function (error) {
-        assert.ok(false, "COPY TO stream should not emit errors" + helper.sys.inspect(error)); 
+        assert.ok(false, "COPY TO stream should not emit errors" + helper.sys.inspect(error));
       });
       assert.emits(stream, 'data', function (chunk) {
-        buf = Buffer.concat([buf, chunk]);  
-      }, "COPY IN stream should emit data event for each row"); 
+        buf = Buffer.concat([buf, chunk]);
+      }, "COPY IN stream should emit data event for each row");
       assert.emits(stream, 'end', function () {
         copyQueryDone = true;
         assert.ok(query1Done && ! query2Done, "copy query has to be executed before second query and after first");
@@ -100,14 +101,14 @@ test('COPY TO, queue queries', function () {
 
 test("COPY TO incorrect usage with large data", function () {
   if(helper.config.native) return false;
-  //when many data is loaded from database (and it takes a lot of time) 
+  //when many data is loaded from database (and it takes a lot of time)
   //there are chance, that query will be canceled before it ends
-  //but if there are not so much data, cancel message may be 
+  //but if there are not so much data, cancel message may be
   //send after copy query ends
   //so we need to test both situations
   pg.connect(helper.config, assert.calls(function (error, client, done) {
     assert.equal(error, null, "Failed to connect: " + helper.sys.inspect(error));
-    //intentionally incorrect usage of copy. 
+    //intentionally incorrect usage of copy.
     //this has to report error in standart way, instead of just throwing exception
     client.query(
       "COPY (SELECT GENERATE_SERIES(1, 10000000)) TO STDOUT WITH CSV",
@@ -127,7 +128,7 @@ test("COPY TO incorrect usage with small data", function () {
   if(helper.config.native) return false;
   pg.connect(helper.config, assert.calls(function (error, client, done) {
     assert.equal(error, null, "Failed to connect: " + helper.sys.inspect(error));
-    //intentionally incorrect usage of copy. 
+    //intentionally incorrect usage of copy.
     //this has to report error in standart way, instead of just throwing exception
     client.query(
       "COPY (SELECT GENERATE_SERIES(1, 1)) TO STDOUT WITH CSV",
@@ -147,7 +148,7 @@ test("COPY FROM incorrect usage", function () {
   pg.connect(helper.config, function (error, client, done) {
     assert.equal(error, null, "Failed to connect: " + helper.sys.inspect(error));
     prepareTable(client, function () {
-      //intentionally incorrect usage of copy. 
+      //intentionally incorrect usage of copy.
       //this has to report error in standart way, instead of just throwing exception
       client.query(
         "COPY copy_test from STDIN WITH CSV",
@@ -157,7 +158,7 @@ test("COPY FROM incorrect usage", function () {
             assert.isNull(error, "incorrect copy usage should not break connection: " + error);
             assert.ok(result, "incorrect copy usage should not break connection");
             done();
-            pg.end(helper.config);     
+            pg.end(helper.config);
           }));
         })
       );

test/integration/client/query-error-handling-prepared-statement-tests.js

@@ -0,0 +1,83 @@
+var helper = require(__dirname + '/test-helper');
+var util = require('util');
+
+function killIdleQuery(targetQuery) {
+  var client2 = new Client(helper.args);
+  var pidColName = 'procpid'
+  var queryColName = 'current_query';
+  client2.connect(assert.success(function() {
+    helper.versionGTE(client2, '9.2.0', assert.success(function(isGreater) {
+      if(isGreater) {
+        pidColName = 'pid';
+        queryColName = 'query';
+      }
+      var killIdleQuery = "SELECT " + pidColName + ", (SELECT pg_terminate_backend(" + pidColName + ")) AS killed FROM pg_stat_activity WHERE " + queryColName + " = $1";
+      client2.query(killIdleQuery, [targetQuery], assert.calls(function(err, res) {
+        assert.ifError(err);
+        assert.equal(res.rows.length, 1);
+        client2.end();
+        assert.emits(client2, 'end');
+      }));
+    }));
+  }));
+}
+
+test('query killed during query execution of prepared statement', function() {
+  if(helper.args.native) {
+    return false;
+  }
+  var client = new Client(helper.args);
+  client.connect(assert.success(function() {
+    var sleepQuery = 'select pg_sleep($1)';
+    var query1 = client.query({
+      name: 'sleep query',
+      text: sleepQuery,
+      values: [5] },
+      assert.calls(function(err, result) {
+        assert.equal(err.message, 'terminating connection due to administrator command');
+    }));
+
+    query1.on('error', function(err) {
+      assert.fail('Prepared statement should not emit error');
+    });
+
+    query1.on('row', function(row) {
+      assert.fail('Prepared statement should not emit row');
+    });
+
+    query1.on('end', function(err) {
+      assert.fail('Prepared statement when executed should not return before being killed');
+    });
+
+    killIdleQuery(sleepQuery);
+  }));
+});
+
+
+test('client end during query execution of prepared statement', function() {
+  var client = new Client(helper.args);
+  client.connect(assert.success(function() {
+    var sleepQuery = 'select pg_sleep($1)';
+    var query1 = client.query({
+      name: 'sleep query',
+      text: sleepQuery,
+      values: [5] },
+      assert.calls(function(err, result) {
+        assert.equal(err.message, 'Connection was ended during query');
+    }));
+
+    query1.on('error', function(err) {
+      assert.fail('Prepared statement should not emit error');
+    });
+
+    query1.on('row', function(row) {
+      assert.fail('Prepared statement should not emit row');
+    });
+
+    query1.on('end', function(err) {
+      assert.fail('Prepared statement when executed should not return before being killed');
+    });
+
+    client.end();
+  }));
+});