More correct pg wire protocol error handling

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

@@ -5,7 +5,7 @@
 // They are only available as ESM, and if we import them normally,
 // our tsconfig will convert them to requires, which will cause errors
 // during module resolution.
 const { dynamicImport } = require(true && "../../dynamicImport");
 import * as net from "node:net";
 import * as fs from "fs";
 
@@ -41,10 +41,10 @@
   public db: PGlite | undefined = undefined;
   private server: net.Server | undefined = undefined;
 
   public async createPGServer(host: string = "127.0.0.1", port: number): Promise<net.Server> {
     const getDb = this.getDb.bind(this);
 
     const server = net.createServer(async (socket) => {
       const connection: PostgresConnection = await fromNodeSocket(socket, {
         serverVersion: "16.3 (PGlite 0.2.0)",
         auth: { method: "trust" },
@@ -63,11 +63,11 @@
           const result = await db.execProtocolRaw(data);
           // Extended query patch removes the extra Ready for Query messages that
           // pglite wrongly sends.
           return extendedQueryPatch.filterResponse(data, result);
         },
       });
 
       const extendedQueryPatch: PGliteExtendedQueryPatch = new PGliteExtendedQueryPatch(connection);
 
       socket.on("end", () => {
         logger.debug("Postgres client disconnected");
@@ -95,8 +95,8 @@
       }
       // Not all schemas will need vector installed, but we don't have an good way
       // to swap extensions after starting PGLite, so we always include it.
       const vector = (await dynamicImport("@electric-sql/pglite/vector")).vector;
       const uuidOssp = (await dynamicImport("@electric-sql/pglite/contrib/uuid_ossp")).uuid_ossp;
       const pgliteArgs: PGliteOptions = {
         debug: this.debug,
         extensions: {
@@ -164,6 +164,7 @@
 // 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 @@
     // '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) {