More correct pg wire protocol error handling (#8348)

Author: joehan

CHANGELOG.md

@@ -11,3 +11,4 @@
 - Fixed an issue where credentials from `firebase login` would not be correctly provided to the Data Connect emulator.
 - Fixed misleading comments in `firebase init dataconnect` `connector.yaml` template.
 - Improved Data Connect SQL permissions to better handle tables owned by IAM roles. (#8339)
+- Fixed an issue where the Data Connect emulator would crash after some SQL errors.

src/emulator/dataconnect/pgliteServer.ts

@@ -164,6 +164,7 @@ export class PostgresServer {
 // TODO: Remove this code once https://github.com/electric-sql/pglite/pull/294 is released in PGLite
 export class PGliteExtendedQueryPatch {
   isExtendedQuery = false;
+  eqpErrored = false;
 
   constructor(public connection: PostgresConnection) {}
 
@@ -184,16 +185,21 @@ export class PGliteExtendedQueryPatch {
     // 'Sync' indicates the end of an extended query
     if (message[0] === FrontendMessageCode.Sync) {
       this.isExtendedQuery = false;
+      this.eqpErrored = false;
 
       // Manually inject 'ReadyForQuery' message at the end
       return this.connection.createReadyForQuery();
     }
 
     // A PGlite response can contain multiple messages
     for await (const message of getMessages(response)) {
-      // If a prepared statement leads to an error message, we need to end the pipeline.
-      if (message[0] === BackendMessageCode.ErrorMessage) {
-        this.isExtendedQuery = false;
+      // After an ErrorMessage in extended query protocol, we should throw away messages until the next Sync
+      // (per https://www.postgresql.org/docs/current/protocol-flow.html#PROTOCOL-FLOW-EXT-QUERY:~:text=When%20an%20error,for%20each%20Sync.))
+      if (this.eqpErrored) {
+        continue;
+      }
+      if (this.isExtendedQuery && message[0] === BackendMessageCode.ErrorMessage) {
+        this.eqpErrored = true;
       }
       // Filter out incorrect `ReadyForQuery` messages during the extended query protocol
       if (this.isExtendedQuery && message[0] === BackendMessageCode.ReadyForQuery) {