fix: double client.end() hang

packages/pg/lib/client.js

@@ -38,6 +38,7 @@ class Client extends EventEmitter {
     this._Promise = c.Promise || global.Promise
     this._types = new TypeOverrides(c.types)
     this._ending = false
+    this._ended = false
     this._connecting = false
     this._connected = false
     this._connectionError = false
@@ -133,6 +134,7 @@ class Client extends EventEmitter {
 
       clearTimeout(this.connectionTimeoutHandle)
       this._errorAllQueries(error)
+      this._ended = true
 
       if (!this._ending) {
         // if the connection is ended without us calling .end()
@@ -589,7 +591,7 @@ class Client extends EventEmitter {
     this._ending = true
 
     // if we have never connected, then end is a noop, callback immediately
-    if (!this.connection._connecting) {
+    if (!this.connection._connecting || this._ended) {
       if (cb) {
         cb()
       } else {

packages/pg/lib/connection.js

@@ -103,9 +103,6 @@ class Connection extends EventEmitter {
   }
 
   attachListeners(stream) {
-    stream.on('end', () => {
-      this.emit('end')
-    })
     parse(stream, (msg) => {
       var eventName = msg.name === 'error' ? 'errorMessage' : msg.name
       if (this._emitMessage) {

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

@@ -0,0 +1,38 @@
+'use strict'
+const helper = require('../test-helper')
+
+const suite = new helper.Suite()
+
+// https://github.com/brianc/node-postgres/issues/2716
+suite.testAsync('client.end() should resolve if already ended', async () => {
+  const client = new helper.pg.Client()
+  await client.connect()
+
+  // this should resolve only when the underlying socket is fully closed, both
+  // the readable part ("end" event) & writable part ("close" event).
+
+  // https://nodejs.org/docs/latest-v16.x/api/net.html#event-end
+  // > Emitted when the other end of the socket signals the end of
+  // > transmission, thus ending the readable side of the socket.
+
+  // https://nodejs.org/docs/latest-v16.x/api/net.html#event-close_1
+  // > Emitted once the socket is fully closed.
+
+  // here: stream = socket
+
+  await client.end()
+  // connection.end()
+  //   stream.end()
+  // ...
+  // stream emits "end"
+  //   not listening to this event anymore so the promise doesn't resolve yet
+  // stream emits "close"; no more events will be emitted from the stream
+  //   connection emits "end"
+  //     promise resolved
+
+  // This should now resolve immediately, rather than wait for connection.on('end')
+  await client.end()
+
+  // this should resolve immediately, rather than waiting forever
+  await client.end()
+})