fix(pg-cursor): handle exceptions from row type parsers

packages/pg-cursor/index.js

@@ -115,7 +115,16 @@ class Cursor extends EventEmitter {
   }
 
   handleDataRow(msg) {
-    const row = this._result.parseRow(msg.fields)
+    if (this._error) return
+
+    let row
+    try {
+      row = this._result.parseRow(msg.fields)
+    } catch (err) {
+      // Drain the current batch before closing the portal and reporting the error.
+      this._error = err
+      return
+    }
     this.emit('row', row, this._result)
     this._rows.push(row)
   }
@@ -144,10 +153,17 @@ class Cursor extends EventEmitter {
   }
 
   handlePortalSuspended() {
+    if (this._error) {
+      return this._closePortal()
+    }
     this._sendRows()
   }
 
   handleReadyForQuery() {
+    if (this._error) {
+      this.state = 'done'
+      return this._reportError(this._error)
+    }
     this._sendRows()
     this.state = 'done'
     this.emit('end', this._result)
@@ -174,10 +190,19 @@ class Cursor extends EventEmitter {
     }
 
     this.state = 'error'
+    this._reportError(this._error || msg)
+  }
+
+  _reportError(msg) {
+    // A timeout can report the error before ReadyForQuery.
+    if (this._errorReported) return
+    this._errorReported = true
     this._error = msg
     // satisfy any waiting callback
     if (this._cb) {
-      this._cb(msg)
+      const cb = this._cb
+      this._cb = null
+      cb(msg)
     }
     // dispatch error to all waiting callbacks
     for (let i = 0; i < this._queue.length; i++) {
@@ -250,7 +275,7 @@ class Cursor extends EventEmitter {
       this._getRows(rows, cb)
     } else if (this.state === 'busy' || this.state === 'initialized') {
       this._queue.push([rows, cb])
-    } else if (this.state === 'error') {
+    } else if (this.state === 'error' || (this.state === 'done' && this._error)) {
       setImmediate(() => cb(this._error))
     } else if (this.state === 'done') {
       setImmediate(() => cb(null, []))

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

@@ -51,6 +51,43 @@ describe('error handling', function () {
 
     client.end()
   })
+
+  it('reports a read timeout once after the server finishes the query', async () => {
+    const client = new pg.Client()
+    await client.connect()
+    try {
+      await client.query('BEGIN')
+      const cursor = new Cursor('SELECT 1 AS value')
+      const queryTimeout = 20
+      cursor.query_timeout = queryTimeout
+      const errors = []
+      const callbackErrors = []
+      cursor.on('error', (error) => errors.push(error))
+
+      // Hold server messages so the timeout precedes protocol completion.
+      client.connection.stream.pause()
+      client.query(cursor)
+      const timeoutError = await new Promise((resolve) => {
+        cursor.read(1, (error) => {
+          callbackErrors.push(error)
+          resolve(error)
+        })
+      })
+      assert.strictEqual(timeoutError.message, 'Query read timeout')
+      const completed = new Promise((resolve) => client.connection.once('readyForQuery', resolve))
+      client.connection.stream.resume()
+      await completed
+
+      const result = await client.query('SELECT 42 AS answer')
+      assert.deepStrictEqual(result.rows, [{ answer: 42 }])
+      assert.deepStrictEqual(callbackErrors, [timeoutError])
+      assert.deepStrictEqual(errors, [timeoutError])
+      await client.query('ROLLBACK')
+    } finally {
+      client.connection.stream.resume()
+      await client.end()
+    }
+  })
 })
 
 describe('read callback does not fire sync', () => {

packages/pg-cursor/test/type-parser-errors.js

@@ -0,0 +1,80 @@
+'use strict'
+const assert = require('assert')
+const Cursor = require('../')
+const pg = require('pg')
+
+describe('type parser errors', () => {
+  let client
+
+  beforeEach(async () => {
+    client = new pg.Client()
+    await client.connect()
+  })
+
+  afterEach(async () => {
+    await client.end()
+  })
+
+  it('preserves the parser error if a later row causes a server error', async () => {
+    const parserError = new Error('cannot decode value')
+    const types = {
+      getTypeParser: () => (value) => {
+        if (value === '3') throw parserError
+        return Number(value)
+      },
+    }
+    const text = 'SELECT value, 10 / (5 - value) AS quotient FROM generate_series(1, 6) value'
+    await assert.rejects(client.query({ text, types }), (error) => error === parserError)
+
+    const cursor = client.query(new Cursor(text, [], { types }))
+    await assert.rejects(cursor.read(20), (error) => error === parserError)
+    const result = await client.query('SELECT 42 AS answer')
+    assert.deepStrictEqual(result.rows, [{ answer: 42 }])
+  })
+
+  for (const rowMode of [undefined, 'array']) {
+    for (const batchSize of [2, 20]) {
+      it(`reports a parser error once with rowMode=${rowMode} and batchSize=${batchSize}`, async () => {
+        const parserError = new Error('cannot decode value')
+        const parsedValues = []
+        const types = {
+          getTypeParser: () => (value) => {
+            parsedValues.push(value)
+            if (value === '3') throw parserError
+            return Number(value)
+          },
+        }
+        const cursor = client.query(new Cursor('SELECT value FROM generate_series(1, 6) value', [], { types, rowMode }))
+        const emittedErrors = []
+        let endCount = 0
+        cursor.on('error', (error) => emittedErrors.push(error))
+        cursor.on('end', () => endCount++)
+
+        assert.deepStrictEqual(await cursor.read(1), rowMode === 'array' ? [[1]] : [{ value: 1 }])
+
+        let callbackCount = 0
+        const failedRead = new Promise((resolve, reject) => {
+          cursor.read(batchSize, (error) => {
+            callbackCount++
+            error ? reject(error) : resolve()
+          })
+        })
+        const queuedRead = cursor.read(1)
+        const results = await Promise.allSettled([failedRead, queuedRead])
+        for (const result of results) {
+          assert.strictEqual(result.status, 'rejected')
+          assert.strictEqual(result.reason, parserError)
+        }
+
+        await cursor.close()
+        await assert.rejects(cursor.read(1), (error) => error === parserError)
+        const result = await client.query('SELECT 42 AS answer, (SELECT count(*)::int FROM pg_cursors) AS cursors')
+        assert.deepStrictEqual(result.rows, [{ answer: 42, cursors: 0 }])
+        assert.strictEqual(callbackCount, 1)
+        assert.deepStrictEqual(emittedErrors, [parserError])
+        assert.strictEqual(endCount, 0)
+        assert.deepStrictEqual(parsedValues, ['1', '2', '3'])
+      })
+    }
+  }
+})

packages/pg-query-stream/test/type-parser-errors.ts

@@ -0,0 +1,82 @@
+import assert from 'assert'
+import pg from 'pg'
+import QueryStream from '../src'
+
+describe('type parser errors', () => {
+  it('closes a flowing stream once and allows the queued query to finish', async () => {
+    const client = new pg.Client()
+    await client.connect()
+    try {
+      const parserError = new Error('cannot decode value')
+      const stream = new QueryStream('SELECT value FROM generate_series(1, 6) value', [], {
+        batchSize: 2,
+        types: {
+          getTypeParser: () => (value) => {
+            if (value === '3') throw parserError
+            return Number(value)
+          },
+        },
+      })
+      const emittedErrors = []
+      const callbackErrors = []
+      let closeCount = 0
+      stream.on('data', (row) => assert(row.value < 3))
+      stream.on('error', (error) => emittedErrors.push(error))
+      stream.on('close', () => closeCount++)
+      const closed = new Promise<void>((resolve) => stream.once('close', resolve))
+      stream.callback = (error) => callbackErrors.push(error)
+      client.query(stream)
+      const queued = client.query('SELECT 42 AS answer, (SELECT count(*)::int FROM pg_cursors) AS cursors')
+      const [result] = await Promise.all([queued, closed])
+      assert.deepStrictEqual(result.rows, [{ answer: 42, cursors: 0 }])
+      assert.deepStrictEqual(callbackErrors, [parserError])
+      assert.deepStrictEqual(emittedErrors, [parserError])
+      assert.strictEqual(closeCount, 1)
+    } finally {
+      await client.end()
+    }
+  })
+
+  for (const batchSize of [2, 20]) {
+    it(`rejects iteration and releases the portal with batchSize=${batchSize}`, async () => {
+      const client = new pg.Client()
+      await client.connect()
+      try {
+        const parserError = new Error('cannot decode value')
+        const stream = new QueryStream('SELECT value FROM generate_series(1, 6) value', [], {
+          batchSize,
+          types: {
+            getTypeParser: () => (value) => {
+              if (value === '3') throw parserError
+              return Number(value)
+            },
+          },
+        })
+        const callbackErrors = []
+        const emittedErrors = []
+        let closeCount = 0
+        stream.on('error', (error) => emittedErrors.push(error))
+        stream.on('close', () => closeCount++)
+        stream.callback = (error) => callbackErrors.push(error)
+        client.query(stream)
+
+        await assert.rejects(
+          async () => {
+            for await (const row of stream) {
+              assert(row.value < 3)
+            }
+          },
+          (error) => error === parserError
+        )
+
+        const result = await client.query('SELECT 42 AS answer, (SELECT count(*)::int FROM pg_cursors) AS cursors')
+        assert.deepStrictEqual(result.rows, [{ answer: 42, cursors: 0 }])
+        assert.deepStrictEqual(callbackErrors, [parserError])
+        assert.deepStrictEqual(emittedErrors, [parserError])
+        assert.strictEqual(closeCount, 1)
+      } finally {
+        await client.end()
+      }
+    })
+  }
+})