From 5508c0ee6bc751ea2474202d12fb36b4f21089a3 Mon Sep 17 00:00:00 2001
From: Matthieu <matthieusieben@users.noreply.github.com>
Date: Fri, 28 Jan 2022 19:59:45 +0100
Subject: [PATCH] fix: Prevent closing the portal twice (#2609)

Fixes brianc/node-postgres#2119
---
 packages/pg-cursor/index.js                     |  5 ++++-
 packages/pg-query-stream/test/async-iterator.ts | 12 ++++++++++++
 2 files changed, 16 insertions(+), 1 deletion(-)

diff --git a/packages/pg-cursor/index.js b/packages/pg-cursor/index.js
index ddfb2b4c..9bbda641 100644
--- a/packages/pg-cursor/index.js
+++ b/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 --git a/packages/pg-query-stream/test/async-iterator.ts b/packages/pg-query-stream/test/async-iterator.ts
index d47ede16..e2f8a755 100644
--- a/packages/pg-query-stream/test/async-iterator.ts
+++ b/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()
+    })
   })
 }