[BUG]: pglite-socket: concurrent extended-protocol batches interleave across connections; one internal throw permanently deadlocks the query queue

[alfredmarte]

Describe the bug

Two related defects in QueryQueueManager.processQueue() in pglite-socket (0.2.7, still present on main). Both bite any client that opens more than one connection; together they make a multi-connection test substrate flake under load. Deterministic, minimal repros for each below.

Defect A — extended-protocol batches from two connections interleave, clobbering the unnamed prepared statement

The queue serializes per message, and applies connection affinity only while db.isInTransaction():

pglite/packages/pglite-socket/src/index.ts

Lines 82 to 98 in 25d0a55

	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,
	)
	query = null
	} else {
	query = this.queue.splice(i, 1)[0]
	}
	} else {
	query = this.queue.shift()
	}

Outside a transaction there is no affinity at all. But an extended-protocol batch (Parse/Bind/Describe/Execute/Sync) is not in a transaction between its messages — so when connections A and B send batches concurrently, their messages are dequeued interleaved into the single shared PGlite session. B's Parse overwrites the unnamed prepared statement between A's Parse and A's Bind. A then binds against B's statement: bind/type errors, or — when the shapes happen to be compatible — silently wrong rows. Neither client did anything unusual; this is plain non-pipelined use of two connections.

Real Postgres gives each connection its own session, so per-connection unnamed statements can never cross.

Repro (node repro-interleave.mjs, deps: @electric-sql/pglite@0.5.4 @electric-sql/pglite-socket@0.2.7 postgres@3.4.9) — two clients issue differently-shaped one-row SELECTs concurrently; prepare: false makes postgres.js use the unnamed statement:

import { PGlite } from '@electric-sql/pglite'
import { PGLiteSocketServer } from '@electric-sql/pglite-socket'
import postgres from 'postgres'

const db = await PGlite.create()
const server = new PGLiteSocketServer({ db, host: '127.0.0.1', port: 0, maxConnections: 8 })
await server.start()
const port = Number(server.getServerConn().split(':').pop())
const url = `postgres://postgres:postgres@127.0.0.1:${port}/postgres`

const a = postgres(url, { max: 1, prepare: false, onnotice: () => {} })
const b = postgres(url, { max: 1, prepare: false, onnotice: () => {} })

try {
  for (let i = 0; i < 500; i++) {
    const [ra, rb] = await Promise.all([
      a`SELECT ${'client-a'}::text AS who, ${i}::int AS i`,
      b`SELECT ${i + 1000}::int * 2 AS doubled, ${'x'}::text || 'y' AS s`,
    ]).catch((err) => {
      throw new Error(`iteration ${i}: ${err.message ?? err}`)
    })
    if (ra[0].who !== 'client-a' || Number(ra[0].i) !== i)
      throw new Error(`iteration ${i}: client A got wrong row: ${JSON.stringify(ra[0])}`)
    if (Number(rb[0].doubled) !== (i + 1000) * 2 || rb[0].s !== 'xy')
      throw new Error(`iteration ${i}: client B got wrong row: ${JSON.stringify(rb[0])}`)
  }
  console.log('OK: 500 concurrent iterations, no cross-connection interleave')
  process.exit(0)
} catch (err) {
  console.error('REPRODUCED:', err.message ?? err)
  process.exit(1)
}

Observed output (fails immediately, every run):

REPRODUCED: iteration 0: invalid input syntax for type integer: "client-a"

Client A's 'client-a' text parameter was bound into client B's statement (whose $1 is an int) — direct proof the batches crossed.

Fix sketch: treat the first extended-protocol message (Parse/Bind/Describe/Execute/Flush/Close/Copy*) from a handler as opening a batch owned by that handler, and dequeue only that handler's messages until its Sync completes (releasing ownership on handler disconnect). We run with exactly this patch downstream and it holds up under heavy concurrent load.

Defect B — one execProtocolRawStream throw leaves processing stuck true, permanently deadlocking the queue for all connections

pglite/packages/pglite-socket/src/index.ts

Lines 106 to 132 in 25d0a55

	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
	}
	this.log(
	`query from handler #${query.handlerId} completed, ${result} bytes`,
	)
	this.lastHandlerId = query.handlerId
	query.resolve(result)
	}
	this.processing = false
	this.log(`queue processing complete, queue length is`, this.queue.length)
	}

      } catch (error) {
        this.log(`query from handler #${query.handlerId} failed:`, error)
        query.reject(error as Error)
        return                        // <-- exits processQueue() without ever reaching…
      }
      ...
    }
    this.processing = false           // <-- …this line

enqueue() only kicks the loop if (!this.processing), so after a single throw no message from any connection is ever processed again. New connections' startup messages sit in the queue forever — in a test suite this surfaces as CONNECT_TIMEOUT on connections that have nothing to do with the original failure.

SQL errors don't take this path (they come back as ErrorResponse bytes), but internal failures do — e.g. a WASM abort, or any in-flight query racing db.close() (PGlite is closed). Related but distinct from #985: that one is the idx === -1 wedge with an idle in-transaction handler; this one needs no transaction at all, just one rejected execProtocolRawStream call.

Repro (node repro-queue-stall.mjs) — injects a single rejection at the exact seam (first Parse rejects, everything else untouched), then shows a fresh connection against the now-healthy db hangs forever:

import net from 'node:net'
import { PGlite } from '@electric-sql/pglite'
import { PGLiteSocketServer } from '@electric-sql/pglite-socket'
import postgres from 'postgres'

const db = await PGlite.create()

// Fault injection: the FIRST Parse message ('P') rejects, everything else
// behaves normally — simulating a single transient internal failure.
const original = db.execProtocolRawStream.bind(db)
let injected = false
db.execProtocolRawStream = async (message, options) => {
  if (!injected && message[0] === 0x50 /* Parse */) {
    injected = true
    throw new Error('injected internal failure (stand-in for a WASM abort / closed-db race)')
  }
  return original(message, options)
}

const server = new PGLiteSocketServer({ db, host: '127.0.0.1', port: 0, maxConnections: 8 })
await server.start()
const port = Number(server.getServerConn().split(':').pop())
const url = `postgres://postgres:postgres@127.0.0.1:${port}/postgres`

// --- connection 1: raw socket, trips the injected throw -------------------
function msg(type, body) {
  const b = Buffer.concat([Buffer.alloc(4), body])
  b.writeInt32BE(body.length + 4, 0)
  return type ? Buffer.concat([Buffer.from(type), b]) : b
}
const cstr = (s) => Buffer.from(s + '\0')

await new Promise((resolve) => {
  const sock = net.connect(port, '127.0.0.1', () => {
    sock.write(msg(null, Buffer.concat([Buffer.from([0, 3, 0, 0]), cstr('user'), cstr('postgres'), cstr('database'), cstr('postgres'), Buffer.from([0])])))
  })
  let sent = false
  sock.on('data', (d) => {
    if (!sent && d.includes(0x5a /* ReadyForQuery */)) {
      sent = true
      sock.write(Buffer.concat([
        msg('P', Buffer.concat([cstr(''), cstr('SELECT 1'), Buffer.from([0, 0])])),
        msg('S', Buffer.alloc(0)),
      ]))
    }
  })
  sock.on('close', () => {
    console.log('connection 1: socket closed after the injected throw (expected)')
    resolve()
  })
  sock.on('error', () => {})
})

// --- connection 2: fresh client, healthy db — must work on a sane server --
const sql = postgres(url, { max: 1, prepare: false, connect_timeout: 10, onnotice: () => {} })
const outcome = await Promise.race([
  sql`SELECT 42 AS answer`.then(
    (r) => ({ code: 0, text: `OK: queue recovered, got ${r[0].answer}` }),
    (err) => ({ code: 1, text: `REPRODUCED (as error): healthy connection failed after one earlier throw: ${err.message}` }),
  ),
  new Promise((r) => setTimeout(() => r({ code: 1, text: 'REPRODUCED: queue is deadlocked — a healthy connection hangs forever after one earlier throw' }), 12_000)),
])
console.log(outcome.text)
process.exit(outcome.code)

Observed output:

connection 1: socket closed after the injected throw (expected)
REPRODUCED (as error): healthy connection failed after one earlier throw: write CONNECT_TIMEOUT 127.0.0.1:50849

Fix sketch: wrap the loop body's failure path so it continues (or wrap the whole loop in try { … } finally { this.processing = false }), rejecting only the failed item instead of abandoning the queue.

Logs

Included inline above (both scripts print REPRODUCED: … and exit 1 on current versions).

Details

• PGlite version: @electric-sql/pglite 0.5.4, @electric-sql/pglite-socket 0.2.7 (both latest; defect sites unchanged on main @ 25d0a55)
• using any extensions? which ones?: No
• OS version: macOS 26.5.1 (arm64)
• node, bun, deno or browser version: Node 22.18.0; client is postgres.js (postgres) 3.4.9 with prepare: false

Additional context

Found while running a multi-connection integration-test substrate (postgres.js pool → pglite-socket → PGlite). Under load the two defects compound: an interleave from defect A can make execProtocolRawStream throw, and defect B then wedges the whole server, surfacing as unrelated CONNECT_TIMEOUTs.

Also related (but a separate, already-reported defect): the spurious extra ReadyForQuery on errored extended-protocol messages, #958 — I'll add a deterministic repro there.

We currently work around all of the above by patching processQueue at test-setup time (batch affinity from first extended-protocol message until Sync; try/finally on processing), and are happy to upstream a PR along those lines if the approach sounds right to the maintainers.

[gelias]

Corroboration from adopting pglite-socket as a test database

We're evaluating PGlite via pglite-socket as an in-memory Postgres substitute for an ORM-backed service's test suite (one shared PGlite instance per test worker; the app connects through the pg driver and a normal connection pool, exactly as it does against real Postgres). It's great for single-connection tests, but any path that uses more than one connection hits the defects described here. Adding one more facet and the failure signatures we observe.

Facet C — transaction affinity turns concurrency into a permanent hang

Because affinity is applied only while db.isInTransaction(), once connection A is inside a transaction the queue pins to A; a query dequeued for connection B then blocks until A commits — but if the in-flight logic on A only completes after B returns (a transaction that triggers a query on a second pooled connection), A waits on B and B waits on A → the worker hangs with no error and no timeout.

Scenario	Result
single client: BEGIN / query / COMMIT	ok
pool max: 1: acquire / BEGIN / query / COMMIT	ok
2 connections: BEGIN on A (held open) + any query on B	permanent hang

Minimal, generic repro:

import { PGlite } from '@electric-sql/pglite'
import { PGLiteSocketServer } from '@electric-sql/pglite-socket'
import pg from 'pg'

const db = await PGlite.create()
const server = new PGLiteSocketServer({ db, host: '127.0.0.1', port: 55599, maxConnections: 64 })
await server.start()

const cfg = { host: '127.0.0.1', port: 55599, database: 'xxx', user: 'x', password: 'x' }
const pool = new pg.Pool({ ...cfg, max: 2 })

const a = await pool.connect(); await a.query('BEGIN'); await a.query('SELECT 1')
const b = await pool.connect(); await b.query('SELECT 2')   // never resolves
await a.query('COMMIT')

pool.max tuning doesn't help (max: 2 also deadlocks). Reproduces across versions.

The other mode: the pg client crashing

A response arrives for a connection whose activeQuery is already null, consistent with Defect A/B corrupting per-connection state:

TypeError: Cannot read properties of null (reading 'handleCommandComplete')
    at Client._handleCommandComplete (pg/lib/client.js:382:22)
    at pg/lib/connection.js:117:12
    at Parser.parse (pg-protocol/src/parser.ts:103:9)
    at Socket.<anonymous> (pg-protocol/src/index.ts)

A single test file that issues two requests in sequence generated these by the thousand.

Blocker shapes this produces

All pass on real Postgres; only fail on pglite-socket:

1. Schema setup — a migration that wraps DDL in a transaction which then issues a follow-up query hangs during setup, before any test runs.
2. ORM transactions — a write that opens a transaction and performs a nested read crashes with the trace above.
3. Concurrent requests — a test that fires two requests that each take a pooled connection interleaves/crashes.

Net

For any app using a real connection pool + transactions (i.e. behaving as it does against production Postgres), pglite-socket currently can't be a drop-in substitute for integration/end-to-end testing — it's single-session semantics behind a multi-connection wire protocol. Per-connection session isolation — or at minimum full per-connection affinity for the entire lifetime of an extended-protocol batch and any open transaction, with fair queue progress — would unblock it.

[tdrz]

@gelias Thank you for the additional info.

Here is what I tried:

Terminal 1, start pglite-server:

$ pglite-server -m 10

Terminal 2, start, connect and begin transaction with one psql client:

$ psql --host=localhost
psql (16.14 (Ubuntu 16.14-0ubuntu0.24.04.1), server 18.3)
WARNING: psql major version 16, server major version 18.
         Some psql features might not work.
Type "help" for help.

tdr=# BEGIN;
BEGIN
tdr=*# SELECT 1;
 ?column? 
----------
        1
(1 row)

Terminal 3, start, connect and query with a second psql client:

$ psql --host=localhost
psql (16.14 (Ubuntu 16.14-0ubuntu0.24.04.1), server 18.3)
WARNING: psql major version 16, server major version 18.
         Some psql features might not work.
Type "help" for help.

tdr=# SELECT 2;

// this client hangs here for the moment because there is an ongoing transaction on the other connection.

Back to Terminal 2, commit transaction:

tdr=*# END;
COMMIT

At this point, terminal 3 receives data:

 ?column? 
----------
        2
(1 row)

So this:

It's great for single-connection tests, but any path that uses more than one connection hits the defects described here.

is not correct. The issue seems to be with pg-pool <> pglite-socket rather than just with pglite-socket.

I don't know how pg-pool handles multiple connections, but have a look if it is not possible to force a new connection for each client, or have a look at other parameters to see why my example works and pg-pool doesn't.

[tdrz]

@gelias

I just had a look at your example:

import { PGlite } from '@electric-sql/pglite'
import { PGLiteSocketServer } from '@electric-sql/pglite-socket'
import pg from 'pg'

const db = await PGlite.create()
const server = new PGLiteSocketServer({ db, host: '127.0.0.1', port: 55599, maxConnections: 64 })
await server.start()

const cfg = { host: '127.0.0.1', port: 55599, database: 'xxx', user: 'x', password: 'x' }
const pool = new pg.Pool({ ...cfg, max: 2 })

const a = await pool.connect(); await a.query('BEGIN'); await a.query('SELECT 1')
const b = await pool.connect(); await b.query('SELECT 2')   // never resolves
await a.query('COMMIT')

Re await b.query('SELECT 2') // never resolves this is obvious, as you're awaiting for a query which is blocked by the transaction that is started on a, which you only commit after the SELECT 2. This could never work anyway.

I've just tried without that await, but the connection errors.

Will have a closer look soon, would be great to have pg-pool working with PGlite.

[tdrz]

This works fine:

import { PGlite } from '@electric-sql/pglite'
import { PGLiteSocketServer } from '@electric-sql/pglite-socket'
import pg from 'pg'

const db = await PGlite.create()
console.log('pglite is ready')

const server = new PGLiteSocketServer({ db, host: '127.0.0.1', port: 55599, maxConnections: 64 })
await server.start()
console.log('server has started')

const cfg = { host: '127.0.0.1', port: 55599, database: 'xxx', user: 'x', password: 'x' }
const pool = new pg.Pool({ ...cfg, max: 200 })
console.log('pgpool is up')

const a = await pool.connect(); 
console.log('a connected')

const b = await pool.connect();
console.log('b connected')

await a.query('BEGIN');
console.log('a beginned')
await a.query('SELECT 1')
console.log('a selected')

const bResult = b.query('SELECT 2') // do not await here

await a.query('COMMIT')
console.log('a commited')

const y = await bResult
console.log(y)

await a.release()
await b.release()
await pool.end()

await server.stop()
db.close()

Result:

pglite is ready
server has started
pgpool is up
a connected
b connected
a beginned
a selected
a commited
Result {
  command: 'SELECT',
  rowCount: 1,
  oid: null,
  rows: [ { '?column?': 2 } ],
  fields: [
    Field {
      name: '?column?',
      tableID: 0,
      columnID: 0,
      dataTypeID: 23,
      dataTypeSize: 4,
      dataTypeModifier: -1,
      format: 'text'
    }
  ],
  _parsers: [ [Function: parseInteger] ],
  _types: TypeOverrides {
    _types: {
      getTypeParser: [Function: getTypeParser],
      setTypeParser: [Function: setTypeParser],
      arrayParser: [Object],
      builtins: [Object]
    },
    text: {},
    binary: {}
  },
  RowCtor: null,
  rowAsArray: false,
  _prebuiltEmptyResultObject: { '?column?': null }
}