fix(pg-protocol): fail the stream when the parser throws on an unsupported message

packages/pg-protocol/src/inbound-parser.test.ts

@@ -582,4 +582,30 @@ describe('PgPacketStream', function () {
     parser.parse(oneFieldBuf, () => {})
     assert.strictEqual((parser as any).reader.buffer.byteLength, 0)
   })
+
+  it('fails the stream instead of throwing when the parser throws', async function () {
+    const stream = new PassThrough()
+    const msgs: BackendMessage[] = []
+    const errors: Error[] = []
+    stream.on('error', (err) => errors.push(err))
+    parse(stream, (msg) => msgs.push(msg))
+    // AuthenticationGSS (code 7) is not supported by the parser
+    const gss = new BufferList().addInt32(7).join(true, 'R')
+    stream.write(Buffer.concat([authOkBuffer, gss]))
+    await new Promise((resolve) => stream.once('close', resolve))
+    assert.strictEqual(msgs.length, 1)
+    assert.strictEqual(errors.length, 1)
+    assert.match(errors[0].message, /Unknown authenticationOk message type 7/)
+  })
+
+  it('does not swallow errors thrown by the callback', function () {
+    const stream = new PassThrough()
+    const errors: Error[] = []
+    stream.on('error', (err) => errors.push(err))
+    parse(stream, () => {
+      throw new Error('boom')
+    })
+    assert.throws(() => stream.emit('data', authOkBuffer), /boom/)
+    assert.strictEqual(errors.length, 0)
+  })
 })

packages/pg-protocol/src/index.ts

@@ -1,10 +1,35 @@
+import { Readable } from 'stream'
 import { DatabaseError } from './messages'
 import { serialize } from './serializer'
 import { Parser, MessageCallback } from './parser'
 
 export function parse(stream: NodeJS.ReadableStream, callback: MessageCallback): Promise<void> {
   const parser = new Parser()
-  stream.on('data', (buffer: Buffer) => parser.parse(buffer, callback))
+  stream.on('data', (buffer: Buffer) => {
+    // Errors thrown by the callback are the caller's business and must keep propagating.
+    // Errors thrown by the parser mean the server sent something we cannot handle: fail
+    // the stream instead of letting the exception escape the 'data' listener.
+    let callbackThrew = false
+    try {
+      parser.parse(buffer, (message) => {
+        try {
+          callback(message)
+        } catch (err) {
+          callbackThrew = true
+          throw err
+        }
+      })
+    } catch (err) {
+      if (callbackThrew) {
+        throw err
+      }
+      if (typeof (stream as Readable).destroy === 'function') {
+        ;(stream as Readable).destroy(err as Error)
+      } else {
+        stream.emit('error', err)
+      }
+    }
+  })
   return new Promise((resolve) => stream.on('end', () => resolve()))
 }
 

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

@@ -0,0 +1,33 @@
+'use strict'
+const helper = require('./../test-helper')
+const assert = require('assert')
+const net = require('net')
+const pg = require('../../../lib')
+
+const suite = new helper.Suite()
+
+// a backend that answers the startup message with an authentication request
+// pg-protocol does not know how to parse (code 7 is AuthenticationGSS)
+suite.test('unsupported authentication request rejects connect', (done) => {
+  const server = net.createServer((socket) => {
+    socket.once('data', () => {
+      const message = Buffer.alloc(9)
+      message.write('R', 0)
+      message.writeInt32BE(8, 1)
+      message.writeInt32BE(7, 5)
+      socket.write(message)
+    })
+    socket.on('error', () => {})
+    socket.on('close', () => server.close())
+  })
+  server.listen(0, '127.0.0.1', () => {
+    const port = server.address().port
+    const client = new pg.Client({ host: '127.0.0.1', port, user: 'app', database: 'app' })
+    client.on('error', () => {})
+    client.connect((err) => {
+      assert(err instanceof Error)
+      assert(/Unknown authenticationOk message type 7/.test(err.message))
+      done()
+    })
+  })
+})