pg-cursor: report an error that arrives after CommandComplete

packages/pg-cursor/index.js

@@ -159,7 +159,28 @@ class Cursor extends EventEmitter {
 
   handleError(msg) {
     // If this cursor has already closed, don't try to handle the error.
-    if (this.state === 'done') return
+    if (this.state === 'done') {
+      // An error can still arrive after CommandComplete (for example a deferred
+      // constraint failing when the implicit transaction commits on Sync). A
+      // pending read is only answered on ReadyForQuery, which the client does
+      // not forward after an error, so settle it here. Later reads get the
+      // error too instead of an empty result.
+      this._error = msg
+      if (this._cb) {
+        const cb = this._cb
+        this._cb = null
+        this._rows = []
+        cb(msg)
+      }
+      for (let i = 0; i < this._queue.length; i++) {
+        this._queue[i][1].call(this, msg)
+      }
+      this._queue.length = 0
+      if (this.listenerCount('error') > 0) {
+        this.emit('error', msg)
+      }
+      return
+    }
 
     // If we're in an initialized state we've never been submitted
     // and don't have a connection instance reference yet.
@@ -253,7 +274,11 @@ class Cursor extends EventEmitter {
     } else if (this.state === 'error') {
       setImmediate(() => cb(this._error))
     } else if (this.state === 'done') {
-      setImmediate(() => cb(null, []))
+      if (this._error) {
+        setImmediate(() => cb(this._error))
+      } else {
+        setImmediate(() => cb(null, []))
+      }
     } else {
       throw new Error('Unknown state: ' + this.state)
     }

packages/pg-cursor/test/error-handling.js

@@ -115,4 +115,85 @@ describe('proper cleanup', function () {
       })
     })
   })
+
+  it('rejects a pending read when the error arrives after CommandComplete', async () => {
+    const client = new pg.Client()
+    await client.connect()
+    try {
+      await client.query('CREATE TEMP TABLE cursor_parent (id int PRIMARY KEY)')
+      await client.query(
+        'CREATE TEMP TABLE cursor_child (id int, pid int REFERENCES cursor_parent (id) DEFERRABLE INITIALLY DEFERRED)'
+      )
+      await client.query(
+        'CREATE FUNCTION pg_temp.cursor_insert_child(i int) RETURNS int LANGUAGE plpgsql AS $$ BEGIN INSERT INTO cursor_child VALUES (i, 999); RETURN i; END $$'
+      )
+      const settleWithin = (promise, ms) => {
+        let timer
+        const timeout = new Promise((resolve) => {
+          timer = setTimeout(() => resolve('timeout'), ms)
+        })
+        return Promise.race([promise, timeout]).finally(() => clearTimeout(timer))
+      }
+      const outcome = (promise) =>
+        promise.then(
+          () => 'rows',
+          (err) => err
+        )
+      const cursor = client.query(new Cursor('SELECT pg_temp.cursor_insert_child(g) FROM generate_series(1, 3) g'))
+      const result = await settleWithin(outcome(cursor.read(100)), 1500)
+      assert.notStrictEqual(result, 'timeout', 'read never settled')
+      assert.notStrictEqual(result, 'rows', 'expected the commit-time error')
+      assert(result instanceof Error)
+      const second = await outcome(cursor.read(100))
+      assert.strictEqual(second, result, 'a later read should get the same error')
+      await cursor.close()
+      const { rows } = await client.query('SELECT 1 AS one')
+      assert.strictEqual(rows[0].one, 1)
+    } finally {
+      await client.end()
+    }
+  })
+
+  it('rejects queued reads when the error arrives after CommandComplete', async () => {
+    const client = new pg.Client()
+    await client.connect()
+    try {
+      await client.query('CREATE TEMP TABLE cursor_parent (id int PRIMARY KEY)')
+      await client.query(
+        'CREATE TEMP TABLE cursor_child (id int, pid int REFERENCES cursor_parent (id) DEFERRABLE INITIALLY DEFERRED)'
+      )
+      await client.query(
+        'CREATE FUNCTION pg_temp.cursor_insert_child(i int) RETURNS int LANGUAGE plpgsql AS $$ BEGIN INSERT INTO cursor_child VALUES (i, 999); RETURN i; END $$'
+      )
+      const settleWithin = (promise, ms) => {
+        let timer
+        const timeout = new Promise((resolve) => {
+          timer = setTimeout(() => resolve('timeout'), ms)
+        })
+        return Promise.race([promise, timeout]).finally(() => clearTimeout(timer))
+      }
+      const outcome = (promise) =>
+        promise.then(
+          () => 'rows',
+          (err) => err
+        )
+      // a query ahead of the cursor keeps it in the 'initialized' state, so both reads are queued
+      const blocker = client.query('SELECT 1')
+      const cursor = client.query(new Cursor('SELECT pg_temp.cursor_insert_child(g) FROM generate_series(1, 3) g'))
+      const r1 = outcome(cursor.read(100))
+      const r2 = outcome(cursor.read(100))
+      const results = await settleWithin(Promise.all([r1, r2]), 1500)
+      assert.notStrictEqual(results, 'timeout', 'queued reads never settled')
+      results.forEach((result) => {
+        assert.notStrictEqual(result, 'rows', 'expected the commit-time error')
+        assert(result instanceof Error)
+      })
+      await blocker
+      await cursor.close()
+      const { rows } = await client.query('SELECT 1 AS one')
+      assert.strictEqual(rows[0].one, 1)
+    } finally {
+      await client.end()
+    }
+  })
 })