fix(pg): settle query_timeout once via handleError and clear timer

packages/pg/lib/client.js

@@ -701,21 +701,15 @@ class Client extends EventEmitter {
 
     const readTimeout = config.query_timeout || this.connectionParameters.query_timeout
     if (readTimeout) {
-      const queryCallback = query.callback || (() => {})
-
       const readTimeoutTimer = setTimeout(() => {
         const error = new Error('Query read timeout')
-
         process.nextTick(() => {
           query.handleError(error, this.connection)
+          // we already returned an error,
+          // just do nothing if query completes
+          query.callback = () => {}
         })
 
-        queryCallback(error)
-
-        // we already returned an error,
-        // just do nothing if query completes
-        query.callback = () => {}
-
         // Remove from queue (only safe if not yet sent)
         const index = this._queryQueue.indexOf(query)
         if (index > -1) {
@@ -729,10 +723,15 @@ class Client extends EventEmitter {
 
         this._pulseQueryQueue()
       }, readTimeout)
-
-      query.callback = (err, res) => {
-        clearTimeout(readTimeoutTimer)
-        queryCallback(err, res)
+      //if there is onEndRead === null, use it, otherwise fallback to overriding callback
+      if (query.onEndRead === null) {
+        query.onEndRead = () => clearTimeout(readTimeoutTimer)
+      } else {
+        const queryCallback = query.callback || (() => {})
+        query.callback = (err, res) => {
+          clearTimeout(readTimeoutTimer)
+          queryCallback(err, res)
+        }
       }
     }
 

packages/pg/lib/native/client.js

@@ -190,21 +190,16 @@ Client.prototype.query = function (config, values, callback) {
   }
 
   if (readTimeout) {
-    queryCallback = query.callback || (() => {})
-
     readTimeoutTimer = setTimeout(() => {
       const error = new Error('Query read timeout')
 
       process.nextTick(() => {
         query.handleError(error, this.connection)
+        // we already returned an error,
+        // just do nothing if query completes
+        query.callback = () => {}
       })
 
-      queryCallback(error)
-
-      // we already returned an error,
-      // just do nothing if query completes
-      query.callback = () => {}
-
       // Remove from queue
       const index = this._queryQueue.indexOf(query)
       if (index > -1) {
@@ -214,9 +209,14 @@ Client.prototype.query = function (config, values, callback) {
       this._pulseQueryQueue()
     }, readTimeout)
 
-    query.callback = (err, res) => {
-      clearTimeout(readTimeoutTimer)
-      queryCallback(err, res)
+    if (query.onEndRead === null) {
+      query.onEndRead = () => clearTimeout(readTimeoutTimer)
+    } else {
+      queryCallback = query.callback || (() => {})
+      query.callback = (err, res) => {
+        clearTimeout(readTimeoutTimer)
+        queryCallback(err, res)
+      }
     }
   }
 

packages/pg/lib/native/query.js

@@ -12,6 +12,7 @@ const NativeQuery = (module.exports = function (config, values, callback) {
   this.name = config.name
   this.queryMode = config.queryMode
   this.callback = config.callback
+  this.onEndRead = null
   this.state = 'new'
   this._arrayMode = config.rowMode === 'array'
 
@@ -57,6 +58,7 @@ NativeQuery.prototype.handleError = function (err) {
       err[normalizedFieldName] = fields[key]
     }
   }
+  if (this.onEndRead) this.onEndRead()
   if (this.callback) {
     this.callback(err)
   } else {
@@ -116,6 +118,7 @@ NativeQuery.prototype.submit = function (client) {
       }
     }
 
+    if (self.onEndRead) self.onEndRead()
     // handle successful result
     self.state = 'end'
     self.emit('end', results)

packages/pg/lib/query.js

@@ -30,6 +30,7 @@ class Query extends EventEmitter {
     // potential for multiple results
     this._results = this._result
     this._canceledDueToError = false
+    this.onEndRead = null
   }
 
   requiresPreparation() {
@@ -125,6 +126,7 @@ class Query extends EventEmitter {
       err = this._canceledDueToError
       this._canceledDueToError = false
     }
+    if (this.onEndRead) this.onEndRead()
     // if callback supplied do not emit error event as uncaught error
     // events will bubble up to node process
     if (this.callback) {
@@ -146,6 +148,7 @@ class Query extends EventEmitter {
         })
       }
     }
+    if (this.onEndRead) this.onEndRead()
     this.emit('end', this._results)
   }