Ensure callbacks are executed for all queued queries after connection-level errors

Separates socket errors from error messages, sends socket errors to all queries in the queue, marks clients as unusable after socket errors.

This is not very pleasant but should maintain backwards compatibility…?

Author: charmander

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) {
@@ -353,7 +380,13 @@ 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) {
+        this.activeQuery.handleError(queryError, this.connection)
+        this.readyForQuery = true
+        this._pulseQueryQueue()
+      }
     } else if (this.hasExecuted) {
       this.activeQuery = null
       this.emit('drain')
@@ -389,25 +422,36 @@ Client.prototype.query = function (config, values, callback) {
     query._result._getTypeParser = this._types.getTypeParser.bind(this._types)
   }
 
+  if (!this._queryable) {
+    query.handleError(new Error('Client has encountered a connection error and is not queryable'), this.connection)
+    return
+  }
+
+  if (this._ending) {
+    query.handleError(new Error('Client was closed and is not queryable'), this.connection)
+    return
+  }
+
   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

@@ -112,10 +112,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 () {