Always check if activeQuery is null before using it (#3586)

* check that activeQuery is not null every time before use and emit an error if is.

* docs: code style consistency

* Add more tests

---------

Co-authored-by: Charmander <~@charmander.me>
Co-authored-by: Brian Carlson <brian.m.carlson@gmail.com>

packages/pg/lib/client.js

@@ -407,23 +407,47 @@ class Client extends EventEmitter {
   }
 
   _handleRowDescription(msg) {
+    const activeQuery = this._getActiveQuery()
+    if (activeQuery == null) {
+      const error = new Error('Received unexpected rowDescription message from backend.')
+      this._handleErrorEvent(error)
+      return
+    }
     // delegate rowDescription to active query
-    this._getActiveQuery().handleRowDescription(msg)
+    activeQuery.handleRowDescription(msg)
   }
 
   _handleDataRow(msg) {
+    const activeQuery = this._getActiveQuery()
+    if (activeQuery == null) {
+      const error = new Error('Received unexpected dataRow message from backend.')
+      this._handleErrorEvent(error)
+      return
+    }
     // delegate dataRow to active query
-    this._getActiveQuery().handleDataRow(msg)
+    activeQuery.handleDataRow(msg)
   }
 
   _handlePortalSuspended(msg) {
+    const activeQuery = this._getActiveQuery()
+    if (activeQuery == null) {
+      const error = new Error('Received unexpected portalSuspended message from backend.')
+      this._handleErrorEvent(error)
+      return
+    }
     // delegate portalSuspended to active query
-    this._getActiveQuery().handlePortalSuspended(this.connection)
+    activeQuery.handlePortalSuspended(this.connection)
   }
 
   _handleEmptyQuery(msg) {
+    const activeQuery = this._getActiveQuery()
+    if (activeQuery == null) {
+      const error = new Error('Received unexpected emptyQuery message from backend.')
+      this._handleErrorEvent(error)
+      return
+    }
     // delegate emptyQuery to active query
-    this._getActiveQuery().handleEmptyQuery(this.connection)
+    activeQuery.handleEmptyQuery(this.connection)
   }
 
   _handleCommandComplete(msg) {
@@ -453,11 +477,23 @@ class Client extends EventEmitter {
   }
 
   _handleCopyInResponse(msg) {
-    this._getActiveQuery().handleCopyInResponse(this.connection)
+    const activeQuery = this._getActiveQuery()
+    if (activeQuery == null) {
+      const error = new Error('Received unexpected copyInResponse message from backend.')
+      this._handleErrorEvent(error)
+      return
+    }
+    activeQuery.handleCopyInResponse(this.connection)
   }
 
   _handleCopyData(msg) {
-    this._getActiveQuery().handleCopyData(msg, this.connection)
+    const activeQuery = this._getActiveQuery()
+    if (activeQuery == null) {
+      const error = new Error('Received unexpected copyData message from backend.')
+      this._handleErrorEvent(error)
+      return
+    }
+    activeQuery.handleCopyData(msg, this.connection)
   }
 
   _handleNotification(msg) {

packages/pg/test/integration/gh-issues/3174-tests.js

@@ -165,4 +165,10 @@ const testErrorBuffer = (bufferName, errorBuffer) => {
 if (!helper.args.native) {
   testErrorBuffer('parseComplete', buffers.parseComplete())
   testErrorBuffer('commandComplete', buffers.commandComplete('f'))
+  testErrorBuffer('rowDescription', buffers.rowDescription())
+  testErrorBuffer('dataRow', buffers.dataRow())
+  testErrorBuffer('portalSuspended', buffers.portalSuspended())
+  testErrorBuffer('emptyQuery', buffers.emptyQuery())
+  testErrorBuffer('copyIn', buffers.copyIn(0))
+  testErrorBuffer('copyData', buffers.copyData(Buffer.from([1, 2, 3])))
 }

packages/pg/test/test-buffers.js

@@ -118,4 +118,20 @@ buffers.portalSuspended = function () {
   return new BufferList().join(true, 's')
 }
 
+buffers.copyIn = function (cols) {
+  const list = new BufferList()
+    // text mode
+    .add(Buffer.from([0]))
+    // column count
+    .addInt16(cols)
+  for (let i = 0; i < cols; i++) {
+    list.addInt16(i)
+  }
+  return list.join(true, 'G')
+}
+
+buffers.copyData = function (bytes) {
+  return new BufferList().add(bytes).join(true, 'd')
+}
+
 module.exports = buffers