fix: Prevent closing the portal twice (#2609)

Fixes #2119

Author: matthieusieben

Diff for: packages/pg-cursor/index.js

@@ -86,6 +86,8 @@ class Cursor extends EventEmitter {
   }
 
   _closePortal() {
+    if (this.state === 'done') return
+
     // because we opened a named portal to stream results
     // we need to close the same named portal.  Leaving a named portal
     // open can lock tables for modification if inside a transaction.
@@ -97,6 +99,8 @@ class Cursor extends EventEmitter {
     if (this.state !== 'error') {
       this.connection.sync()
     }
+
+    this.state = 'done'
   }
 
   handleRowDescription(msg) {
@@ -213,7 +217,6 @@ class Cursor extends EventEmitter {
     }
 
     this._closePortal()
-    this.state = 'done'
     this.connection.once('readyForQuery', function () {
       cb()
     })

Diff for: packages/pg-query-stream/test/async-iterator.ts

@@ -117,5 +117,17 @@ if (!process.version.startsWith('v8')) {
       client.release()
       await pool.end()
     })
+
+    it('supports breaking with low watermark', async function () {
+      const pool = new pg.Pool({ max: 1 })
+      const client = await pool.connect()
+
+      for await (const _ of client.query(new QueryStream('select TRUE', [], { highWaterMark: 1 }))) break
+      for await (const _ of client.query(new QueryStream('select TRUE', [], { highWaterMark: 1 }))) break
+      for await (const _ of client.query(new QueryStream('select TRUE', [], { highWaterMark: 1 }))) break
+
+      client.release()
+      await pool.end()
+    })
   })
 }