Merge branch 'queued-query-errors' of https://github.com/charmander/node-postgres

Author: yocontra

lib/client.js

@@ -36,6 +36,7 @@ var Client = function (config) {
   this._connecting = false
   this._connected = false
   this._connectionError = false
+  this._queryable = true
 
   this.connection = c.connection || new Connection({
     stream: c.stream,
@@ -126,15 +127,39 @@ Client.prototype.connect = function (callback) {
   }
 
   const connectedErrorHandler = (err) => {
+    this._queryable = false
+
+    const enqueueError = (query) => {
+      process.nextTick(() => {
+        query.handleError(err, con)
+      })
+    }
+
     if (this.activeQuery) {
-      var activeQuery = self.activeQuery
+      enqueueError(this.activeQuery)
       this.activeQuery = null
-      return activeQuery.handleError(err, con)
     }
+
+    this.queryQueue.forEach(enqueueError)
+    this.queryQueue = []
+
     this.emit('error', err)
   }
 
+  const connectedErrorMessageHandler = (msg) => {
+    const activeQuery = this.activeQuery
+
+    if (!activeQuery) {
+      connectedErrorHandler(msg)
+      return
+    }
+
+    this.activeQuery = null
+    activeQuery.handleError(msg, con)
+  }
+
   con.on('error', connectingErrorHandler)
+  con.on('errorMessage', connectingErrorHandler)
 
   // hook up query handling events to connection
   // after the connection initially becomes ready for queries
@@ -143,7 +168,9 @@ Client.prototype.connect = function (callback) {
     self._connected = true
     self._attachListeners(con)
     con.removeListener('error', connectingErrorHandler)
+    con.removeListener('errorMessage', connectingErrorHandler)
     con.on('error', connectedErrorHandler)
+    con.on('errorMessage', connectedErrorMessageHandler)
 
     // process possible callback argument to Client#connect
     if (callback) {
@@ -181,10 +208,10 @@ Client.prototype.connect = function (callback) {
         if (callback) {
           callback(error)
         } else {
-          this.emit('error', error)
+          connectedErrorHandler(error)
         }
       } else if (!this._connectionError) {
-        this.emit('error', error)
+        connectedErrorHandler(error)
       }
     }
     this.emit('end')
@@ -353,7 +380,15 @@ Client.prototype._pulseQueryQueue = function () {
     if (this.activeQuery) {
       this.readyForQuery = false
       this.hasExecuted = true
-      this.activeQuery.submit(this.connection)
+
+      const queryError = this.activeQuery.submit(this.connection)
+      if (queryError) {
+        process.nextTick(() => {
+          this.activeQuery.handleError(queryError, this.connection)
+          this.readyForQuery = true
+          this._pulseQueryQueue()
+        })
+      }
     } else if (this.hasExecuted) {
       this.activeQuery = null
       this.emit('drain')
@@ -389,25 +424,40 @@ Client.prototype.query = function (config, values, callback) {
     query._result._getTypeParser = this._types.getTypeParser.bind(this._types)
   }
 
+  if (!this._queryable) {
+    process.nextTick(() => {
+      query.handleError(new Error('Client has encountered a connection error and is not queryable'), this.connection)
+    })
+    return result
+  }
+
+  if (this._ending) {
+    process.nextTick(() => {
+      query.handleError(new Error('Client was closed and is not queryable'), this.connection)
+    })
+    return result
+  }
+
   this.queryQueue.push(query)
   this._pulseQueryQueue()
   return result
 }
 
 Client.prototype.end = function (cb) {
   this._ending = true
+
   if (this.activeQuery) {
     // if we have an active query we need to force a disconnect
     // on the socket - otherwise a hung query could block end forever
-    this.connection.stream.destroy(new Error('Connection terminated by user'))
-    return cb ? cb() : Promise.resolve()
+    this.connection.stream.destroy()
+  } else {
+    this.connection.end()
   }
+
   if (cb) {
-    this.connection.end()
     this.connection.once('end', cb)
   } else {
-    return new global.Promise((resolve, reject) => {
-      this.connection.end()
+    return new Promise((resolve) => {
       this.connection.once('end', resolve)
     })
   }

lib/connection.js

@@ -111,10 +111,11 @@ Connection.prototype.attachListeners = function (stream) {
     var packet = self._reader.read()
     while (packet) {
       var msg = self.parseMessage(packet)
+      var eventName = msg.name === 'error' ? 'errorMessage' : msg.name
       if (self._emitMessage) {
         self.emit('message', msg)
       }
-      self.emit(msg.name, msg)
+      self.emit(eventName, msg)
       packet = self._reader.read()
     }
   })

lib/query.js

@@ -147,22 +147,17 @@ Query.prototype.handleError = function (err, connection) {
 
 Query.prototype.submit = function (connection) {
   if (typeof this.text !== 'string' && typeof this.name !== 'string') {
-    const err = new Error('A query must have either text or a name. Supplying neither is unsupported.')
-    connection.emit('error', err)
-    connection.emit('readyForQuery')
-    return
+    return new Error('A query must have either text or a name. Supplying neither is unsupported.')
   }
   if (this.values && !Array.isArray(this.values)) {
-    const err = new Error('Query values must be an array')
-    connection.emit('error', err)
-    connection.emit('readyForQuery')
-    return
+    return new Error('Query values must be an array')
   }
   if (this.requiresPreparation()) {
     this.prepare(connection)
   } else {
     connection.query(this.text)
   }
+  return null
 }
 
 Query.prototype.hasBeenParsed = function (connection) {

test/integration/client/error-handling-tests.js

@@ -50,6 +50,18 @@ suite.test('re-using connections results in promise rejection', (done) => {
   })
 })
 
+suite.test('using a client after closing it results in error', (done) => {
+  const client = new Client()
+  client.connect(() => {
+    client.end(assert.calls(() => {
+      client.query('SELECT 1', assert.calls((err) => {
+        assert.equal(err.message, 'Client was closed and is not queryable')
+        done()
+      }))
+    }))
+  })
+})
+
 suite.test('query receives error on client shutdown', function (done) {
   var client = new Client()
   client.connect(assert.success(function () {

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

@@ -5,15 +5,14 @@ const pg = helper.pg
 // first make pool hold 2 clients
 pg.defaults.poolSize = 2
 
-const pool = new pg.Pool()
-
 const suite = new helper.Suite()
 suite.test('connecting to invalid port', (cb) => {
   const pool = new pg.Pool({ port: 13801 })
   pool.connect().catch(e => cb())
 })
 
 suite.test('errors emitted on pool', (cb) => {
+  const pool = new pg.Pool()
   // get first client
   pool.connect(assert.success(function (client, done) {
     client.id = 1
@@ -48,3 +47,45 @@ suite.test('errors emitted on pool', (cb) => {
     })
   }))
 })
+
+suite.test('connection-level errors cause queued queries to fail', (cb) => {
+  const pool = new pg.Pool()
+  pool.connect(assert.success((client, done) => {
+    client.query('SELECT pg_terminate_backend(pg_backend_pid())', assert.calls((err) => {
+      assert.equal(err.code, '57P01')
+    }))
+
+    pool.once('error', assert.calls((err, brokenClient) => {
+      assert.equal(client, brokenClient)
+    }))
+
+    client.query('SELECT 1', assert.calls((err) => {
+      assert.equal(err.message, 'Connection terminated unexpectedly')
+
+      done()
+      pool.end()
+      cb()
+    }))
+  }))
+})
+
+suite.test('connection-level errors cause future queries to fail', (cb) => {
+  const pool = new pg.Pool()
+  pool.connect(assert.success((client, done) => {
+    client.query('SELECT pg_terminate_backend(pg_backend_pid())', assert.calls((err) => {
+      assert.equal(err.code, '57P01')
+    }))
+
+    pool.once('error', assert.calls((err, brokenClient) => {
+      assert.equal(client, brokenClient)
+
+      client.query('SELECT 1', assert.calls((err) => {
+        assert.equal(err.message, 'Client has encountered a connection error and is not queryable')
+
+        done()
+        pool.end()
+        cb()
+      }))
+    }))
+  }))
+})