fix(parser): handle exceptions within handlePacket

Exception occuring within the handlePacket will bubble up to pg as
uncatchables exceptions errors. Adding a try/catch and returning
an handled error allow the caller program to catch such exceptions
and decide what to do with them.

Fixes brianc#2653

Author: avallete

packages/pg-protocol/src/messages.ts

@@ -120,6 +120,32 @@ export class DatabaseError extends Error implements NoticeOrError {
   }
 }
 
+export class ParserError extends Error implements NoticeOrError {
+  public severity: string | undefined
+  public code: string | undefined
+  public detail: string | undefined
+  public hint: string | undefined
+  public position: string | undefined
+  public internalPosition: string | undefined
+  public internalQuery: string | undefined
+  public where: string | undefined
+  public schema: string | undefined
+  public table: string | undefined
+  public column: string | undefined
+  public dataType: string | undefined
+  public constraint: string | undefined
+  public file: string | undefined
+  public line: string | undefined
+  public routine: string | undefined
+  constructor(
+    message: string,
+    public readonly length: number,
+    public readonly name: MessageName
+  ) {
+    super(message)
+  }
+}
+
 export class CopyDataMessage {
   public readonly name = 'copyData'
   constructor(

packages/pg-protocol/src/parser.ts

@@ -25,6 +25,7 @@ import {
   MessageName,
   AuthenticationMD5Password,
   NoticeMessage,
+  ParserError,
 } from './messages'
 import { BufferReader } from './buffer-reader'
 
@@ -79,13 +80,11 @@ export class Parser {
   private bufferLength: number = 0
   private bufferOffset: number = 0
   private reader = new BufferReader()
-  private mode: Mode
 
   constructor(opts?: StreamOptions) {
     if (opts?.mode === 'binary') {
       throw new Error('Binary mode not supported yet')
     }
-    this.mode = opts?.mode || 'text'
   }
 
   public parse(buffer: Buffer, callback: MessageCallback) {
@@ -152,53 +151,57 @@ export class Parser {
   }
 
   private handlePacket(offset: number, code: number, length: number, bytes: Buffer): BackendMessage {
-    switch (code) {
-      case MessageCodes.BindComplete:
-        return bindComplete
-      case MessageCodes.ParseComplete:
-        return parseComplete
-      case MessageCodes.CloseComplete:
-        return closeComplete
-      case MessageCodes.NoData:
-        return noData
-      case MessageCodes.PortalSuspended:
-        return portalSuspended
-      case MessageCodes.CopyDone:
-        return copyDone
-      case MessageCodes.ReplicationStart:
-        return replicationStart
-      case MessageCodes.EmptyQuery:
-        return emptyQuery
-      case MessageCodes.DataRow:
-        return this.parseDataRowMessage(offset, length, bytes)
-      case MessageCodes.CommandComplete:
-        return this.parseCommandCompleteMessage(offset, length, bytes)
-      case MessageCodes.ReadyForQuery:
-        return this.parseReadyForQueryMessage(offset, length, bytes)
-      case MessageCodes.NotificationResponse:
-        return this.parseNotificationMessage(offset, length, bytes)
-      case MessageCodes.AuthenticationResponse:
-        return this.parseAuthenticationResponse(offset, length, bytes)
-      case MessageCodes.ParameterStatus:
-        return this.parseParameterStatusMessage(offset, length, bytes)
-      case MessageCodes.BackendKeyData:
-        return this.parseBackendKeyData(offset, length, bytes)
-      case MessageCodes.ErrorMessage:
-        return this.parseErrorMessage(offset, length, bytes, 'error')
-      case MessageCodes.NoticeMessage:
-        return this.parseErrorMessage(offset, length, bytes, 'notice')
-      case MessageCodes.RowDescriptionMessage:
-        return this.parseRowDescriptionMessage(offset, length, bytes)
-      case MessageCodes.ParameterDescriptionMessage:
-        return this.parseParameterDescriptionMessage(offset, length, bytes)
-      case MessageCodes.CopyIn:
-        return this.parseCopyInMessage(offset, length, bytes)
-      case MessageCodes.CopyOut:
-        return this.parseCopyOutMessage(offset, length, bytes)
-      case MessageCodes.CopyData:
-        return this.parseCopyData(offset, length, bytes)
-      default:
-        return new DatabaseError('received invalid response: ' + code.toString(16), length, 'error')
+    try {
+      switch (code) {
+        case MessageCodes.BindComplete:
+          return bindComplete
+        case MessageCodes.ParseComplete:
+          return parseComplete
+        case MessageCodes.CloseComplete:
+          return closeComplete
+        case MessageCodes.NoData:
+          return noData
+        case MessageCodes.PortalSuspended:
+          return portalSuspended
+        case MessageCodes.CopyDone:
+          return copyDone
+        case MessageCodes.ReplicationStart:
+          return replicationStart
+        case MessageCodes.EmptyQuery:
+          return emptyQuery
+        case MessageCodes.DataRow:
+          return this.parseDataRowMessage(offset, length, bytes)
+        case MessageCodes.CommandComplete:
+          return this.parseCommandCompleteMessage(offset, length, bytes)
+        case MessageCodes.ReadyForQuery:
+          return this.parseReadyForQueryMessage(offset, length, bytes)
+        case MessageCodes.NotificationResponse:
+          return this.parseNotificationMessage(offset, length, bytes)
+        case MessageCodes.AuthenticationResponse:
+          return this.parseAuthenticationResponse(offset, length, bytes)
+        case MessageCodes.ParameterStatus:
+          return this.parseParameterStatusMessage(offset, length, bytes)
+        case MessageCodes.BackendKeyData:
+          return this.parseBackendKeyData(offset, length, bytes)
+        case MessageCodes.ErrorMessage:
+          return this.parseErrorMessage(offset, length, bytes, 'error')
+        case MessageCodes.NoticeMessage:
+          return this.parseErrorMessage(offset, length, bytes, 'notice')
+        case MessageCodes.RowDescriptionMessage:
+          return this.parseRowDescriptionMessage(offset, length, bytes)
+        case MessageCodes.ParameterDescriptionMessage:
+          return this.parseParameterDescriptionMessage(offset, length, bytes)
+        case MessageCodes.CopyIn:
+          return this.parseCopyInMessage(offset, length, bytes)
+        case MessageCodes.CopyOut:
+          return this.parseCopyOutMessage(offset, length, bytes)
+        case MessageCodes.CopyData:
+          return this.parseCopyData(offset, length, bytes)
+        default:
+          return new DatabaseError('received invalid response: ' + code.toString(16), length, 'error')
+      }
+    } catch (error) {
+      return new ParserError(`unexpected error handling packet: ${error}`, length, 'error')
     }
   }