From 68a7c06337c7a93af898d901b08df6b27a2d3999 Mon Sep 17 00:00:00 2001 From: Brown-Sage Date: Sun, 23 Aug 2026 13:35:12 +0530 Subject: [PATCH] fix(pglite-socket): recover query queue after an internal execution error A single throw from execProtocolRawStream (e.g. a WASM abort or a race with db.close()) left processQueue() with processing stuck at true, so enqueue() never restarted the loop and every subsequent query from any connection hung forever. Reset the flag in a finally block and continue draining the queue after a failed execution. Transaction affinity semantics are unchanged. Fixes #1046 (Defect B) --- .changeset/socket-queue-deadlock.md | 5 + packages/pglite-socket/src/index.ts | 92 +++++++------ .../tests/queue-recovery.test.ts | 130 ++++++++++++++++++ 3 files changed, 183 insertions(+), 44 deletions(-) create mode 100644 .changeset/socket-queue-deadlock.md create mode 100644 packages/pglite-socket/tests/queue-recovery.test.ts diff --git a/.changeset/socket-queue-deadlock.md b/.changeset/socket-queue-deadlock.md new file mode 100644 index 000000000..429ca9c93 --- /dev/null +++ b/.changeset/socket-queue-deadlock.md @@ -0,0 +1,5 @@ +--- +'@electric-sql/pglite-socket': patch +--- + +Fix a permanent deadlock in the query queue: a single internal failure thrown while executing one query (e.g. a WASM abort or a race with `db.close()`) left the queue's `processing` flag stuck, so no further queries from any connection were ever processed. The queue now recovers and keeps serving queued queries (#1046). diff --git a/packages/pglite-socket/src/index.ts b/packages/pglite-socket/src/index.ts index 19ab6db05..431e27e48 100644 --- a/packages/pglite-socket/src/index.ts +++ b/packages/pglite-socket/src/index.ts @@ -76,59 +76,63 @@ class QueryQueueManager { this.processing = true - while (this.queue.length > 0) { - let query + try { + while (this.queue.length > 0) { + let query - if (this.db.isInTransaction() && this.lastHandlerId) { - const i = this.queue.findIndex( - (q) => q.handlerId === this.lastHandlerId, - ) - if (i === -1) { - // we didn't find any other query from the same client! - this.log( - `transaction started, but no query from the same handler id found in queue`, - this.lastHandlerId, + if (this.db.isInTransaction() && this.lastHandlerId) { + const i = this.queue.findIndex( + (q) => q.handlerId === this.lastHandlerId, ) - query = null + if (i === -1) { + // we didn't find any other query from the same client! + this.log( + `transaction started, but no query from the same handler id found in queue`, + this.lastHandlerId, + ) + query = null + } else { + query = this.queue.splice(i, 1)[0] + } } else { - query = this.queue.splice(i, 1)[0] + query = this.queue.shift() } - } else { - query = this.queue.shift() - } - if (!query) break + if (!query) break - const waitTime = Date.now() - query.timestamp - this.log( - `processing query from handler #${query.handlerId} (waited ${waitTime}ms)`, - ) + const waitTime = Date.now() - query.timestamp + this.log( + `processing query from handler #${query.handlerId} (waited ${waitTime}ms)`, + ) - let result = 0 - try { - // Execute the query with exclusive access to PGlite - await this.db.runExclusive(async () => { - return await this.db.execProtocolRawStream(query.message, { - onRawData: (data) => { - result += data.length - query.onData(data) - }, + let result = 0 + try { + // Execute the query with exclusive access to PGlite + await this.db.runExclusive(async () => { + return await this.db.execProtocolRawStream(query.message, { + onRawData: (data) => { + result += data.length + query.onData(data) + }, + }) }) - }) - } catch (error) { - this.log(`query from handler #${query.handlerId} failed:`, error) - query.reject(error as Error) - return - } + } catch (error) { + this.log(`query from handler #${query.handlerId} failed:`, error) + query.reject(error as Error) + // continue processing the remaining queued queries so a single + // failed execution does not deadlock the queue + continue + } - this.log( - `query from handler #${query.handlerId} completed, ${result} bytes`, - ) - this.lastHandlerId = query.handlerId - query.resolve(result) + this.log( + `query from handler #${query.handlerId} completed, ${result} bytes`, + ) + this.lastHandlerId = query.handlerId + query.resolve(result) + } + } finally { + this.processing = false + this.log(`queue processing complete, queue length is`, this.queue.length) } - - this.processing = false - this.log(`queue processing complete, queue length is`, this.queue.length) } getQueueLength(): number { diff --git a/packages/pglite-socket/tests/queue-recovery.test.ts b/packages/pglite-socket/tests/queue-recovery.test.ts new file mode 100644 index 000000000..c23df614e --- /dev/null +++ b/packages/pglite-socket/tests/queue-recovery.test.ts @@ -0,0 +1,130 @@ +import { describe, it, expect, afterAll } from 'vitest' +import { Client } from 'pg' +import { PGlite } from '@electric-sql/pglite' +import { PGLiteSocketServer } from '../src' + +/** + * Regression test for https://github.com/electric-sql/pglite/issues/1046 (Defect B) + * + * A single internal failure thrown from `execProtocolRawStream` (e.g. a WASM + * abort or a race with `db.close()`) used to leave the query queue's + * `processing` flag stuck at `true`, permanently deadlocking the queue for + * all connections. The queue must recover and keep serving queries. + */ +describe('QueryQueueManager recovers after an internal error', () => { + let db: PGlite + let server: PGLiteSocketServer + + const startServer = async () => { + db = await PGlite.create() + await db.waitReady + + server = new PGLiteSocketServer({ + db, + host: '127.0.0.1', + port: 0, // OS-assigned port + maxConnections: 100, + }) + await server.start() + + const port = (server as any).port as number + return { + port, + config: { + host: '127.0.0.1', + port, + database: 'postgres', + user: 'postgres', + password: 'postgres', + connectionTimeoutMillis: 10000, + }, + } + } + + afterAll(async () => { + if (server) { + await server.stop().catch(() => {}) + } + if (db) { + await db.close() + } + }) + + it('should keep processing queued queries after one query fails internally', async () => { + const { config } = await startServer() + + // Fault injection: the FIRST Parse message rejects, everything else + // behaves normally — simulating a single transient internal failure. + const originalExec = db.execProtocolRawStream.bind(db) + let injected = false + db.execProtocolRawStream = async (message: Uint8Array, options: any) => { + if (!injected && message[0] === 0x50 /* Parse */) { + injected = true + throw new Error('injected internal failure') + } + return originalExec(message, options) + } + + // Client 1 trips the injected failure — its query must fail... + const failingClient = new Client(config) + // Suppress the expected "Connection terminated unexpectedly" error + failingClient.on('error', () => {}) + await failingClient.connect() + await expect( + failingClient.query({ text: 'SELECT $1::int AS one', values: [1] }), + ).rejects.toThrow() + + // ...but the queue must not be deadlocked: subsequent queries from a + // fresh connection must still be processed. + const healthyClient = new Client(config) + await healthyClient.connect() + try { + const result = await healthyClient.query('SELECT 42 AS answer') + expect(result.rows[0].answer).toBe(42) + } finally { + await healthyClient.end() + await failingClient.end().catch(() => {}) + } + }, 30000) + + it('should recover when a queued query is rejected while other queries are pending', async () => { + const { config } = await startServer() + + const originalExec = db.execProtocolRawStream.bind(db) + let injected = false + db.execProtocolRawStream = async (message: Uint8Array, options: any) => { + if (!injected && message[0] === 0x50 /* Parse */) { + injected = true + throw new Error('injected internal failure') + } + return originalExec(message, options) + } + + const clientA = new Client(config) + // Suppress the expected "Connection terminated unexpectedly" error + clientA.on('error', () => {}) + await clientA.connect() + const clientB = new Client(config) + await clientB.connect() + + try { + // Fire both concurrently so B is enqueued while/after A fails + const resultA = clientA.query({ + text: 'SELECT $1::int AS one', + values: [1], + }) + const resultB = clientB.query('SELECT 2 AS two') + + // A's connection hits the injected failure — either the query errors + // or the server closes the socket; both surface as a rejection. + await expect(resultA).rejects.toThrow() + + // B must complete regardless of what happened to A + const resB = await resultB + expect(resB.rows[0].two).toBe(2) + } finally { + await clientA.end().catch(() => {}) + await clientB.end().catch(() => {}) + } + }, 30000) +})