// Regression for the dev-db PGlite socket protocol-interleaving bug (patched// in patches/@electric-sql%2Fpglite-socket@0.1.4.patch).//// PGLiteSocketServer's QueryQueueManager used to enqueue each postgres wire// FRAME (Parse, Bind, Execute, Sync) as its own queue entry against the one// shared PGlite session. With more than one connection (the dev-db now allows// many — see scripts/dev-db.ts maxConnections), two clients' extended-protocol// pipelines interleaved: client A's Parse (5 params) ... client B's Parse// (1 param) ... A's Bind now hits B's unnamed statement://// PostgresError: bind message supplies 5 parameters, but prepared// statement "" requires 1//// which surfaced in e2e as random 500s ("Failed to load tools", StorageError)// on whichever request lost the race — the residual per-spec CI flakes after// the connection-storm fix. The patch batches all frames of one socket data// event into a single queue entry and adds handler affinity while a pipeline// is open, so one client's Parse..Sync executes atomically.//// This test drives concurrent clients issuing unprepared parameterized queries// with DIFFERENT parameter counts (the exact drizzle/postgres-js shape) through// one PGLiteSocketServer and asserts zero protocol corruption. import { setTimeout as sleep } from "node:timers/promises";import { connect, type Socket } from "node:net";import { describe, expect, it } from "@effect/vitest";import { PGlite } from "@electric-sql/pglite";import { PGLiteSocketServer } from "@electric-sql/pglite-socket";import postgres from "postgres"; const PORT = 45998;const CLIENTS = 6;const QUERIES_PER_CLIENT = 40; const makeClient = (port: number, connectTimeout = 5) => postgres(`postgres://postgres:postgres@127.0.0.1:${port}/postgres`, { max: 1, idle_timeout: 0, connect_timeout: connectTimeout, fetch_types: false, prepare: true, onnotice: () => undefined, }); // Hand-rolled wire client: connect and complete the trust-auth startup, so a// test can then speak raw protocol frames (e.g. a lone Parse) that postgres.js// would never emit on its own. Resolves after ReadyForQuery so the next write// is its own data event — and its own queue entry — on the server.const openWireClient = async (port: number): Promise<Socket> => { const socket: Socket = connect(port, "127.0.0.1"); await new Promise<void>((res, rej) => { socket.once("connect", res); socket.once("error", rej); }); const startupBody = Buffer.concat([ Buffer.from([0, 3, 0, 0]), Buffer.from("user\0postgres\0database\0postgres\0\0"), ]); const startup = Buffer.concat([Buffer.alloc(4), startupBody]); startup.writeInt32BE(startup.length, 0); socket.write(startup); await new Promise<void>((res) => { socket.on("data", (chunk: Buffer) => { if (chunk.includes(0x5a)) res(); // 'Z' = ReadyForQuery }); }); return socket;}; // A Parse frame for an unnamed statement: opens an extended-protocol pipeline// that only a later Sync (or the server's recovery) closes.const parseFrame = (query: string): Buffer => { const body = Buffer.concat([Buffer.from(`\0${query}\0`), Buffer.from([0, 0])]); const frame = Buffer.concat([Buffer.from("P"), Buffer.alloc(4), body]); frame.writeInt32BE(4 + body.length, 1); return frame;}; describe("dev-db PGlite socket under concurrent connections", () => { it( "serves interleaved multi-connection pipelines without protocol corruption", { timeout: 60_000 }, async () => { const db = await PGlite.create(); const server = new PGLiteSocketServer({ db, port: PORT, host: "127.0.0.1", maxConnections: 100, }); await server.start(); let ok = 0; const errors: string[] = []; const worker = async (id: number) => { const sql = makeClient(PORT, 10); // oxlint-disable-next-line executor/no-try-catch-or-throw -- test boundary: postgres.js is promise-native and the socket must be closed on every path try { for (let q = 0; q < QUERIES_PER_CLIENT; q++) { // Alternate 1-param and 5-param unprepared queries: maximally // collision-prone unnamed-statement shapes across connections. if ((id + q) % 2 === 0) { await sql.unsafe(`select $1::int as one`, [1]); } else { await sql.unsafe(`select $1::int, $2::text, $3::text, $4::text, $5::text`, [ 1, "b", "c", "d", "e", ]); } ok++; } } catch (cause) { // oxlint-disable-next-line executor/no-unknown-error-message -- test boundary: the raw PostgresError message IS the assertion payload errors.push(String(cause)); } finally { // oxlint-disable-next-line executor/no-promise-catch -- test boundary: postgres.js is promise-native; a failed teardown must not mask the assertion await sql.end({ timeout: 5 }).catch(() => {}); } }; await Promise.all(Array.from({ length: CLIENTS }, (_, i) => worker(i))); await server.stop(); await db.close(); expect(errors, `protocol corruption under concurrency:\n${errors.join("\n")}`).toEqual([]); expect(ok).toBe(CLIENTS * QUERIES_PER_CLIENT); }, ); // Regression for the CI e2e "cloud signIn: callback set no session (500)" // cascade: QueryQueueManager.processQueue used to `return` out of its drain // loop when a query REJECTED (as opposed to returning a wire-level // ErrorResponse), leaving `processing` latched true. From then on every // enqueue — including brand-new connections' startup packets — sat in the // queue forever: in-flight requests hung, postgres.js reconnects died with // CONNECT_TIMEOUT, and the whole dev stack was bricked until restart. The // patch rejects the one entry, drops pipeline affinity, and keeps draining. it( "a rejected query fails one client, not the whole socket server", { timeout: 30_000 }, async () => { const port = 45997; const db = await PGlite.create(); const server = new PGLiteSocketServer({ db, port, host: "127.0.0.1", maxConnections: 100 }); await server.start(); const first = makeClient(port); // oxlint-disable-next-line executor/no-try-catch-or-throw -- test boundary: sockets must be closed on every path try { expect((await first.unsafe(`select 1 as one`))[0]).toEqual({ one: 1 }); // Force the NEXT protocol exchange to reject at the JS level, the shape // PGlite produces when the shared session is broken mid-run. const real = db.execProtocolRawStream.bind(db); let arm = true; (db as { execProtocolRawStream: typeof real }).execProtocolRawStream = (...args) => { if (arm) { arm = false; // oxlint-disable-next-line executor/no-try-catch-or-throw, executor/no-error-constructor -- test boundary: simulating a PGlite internal failure requires a raw throw throw new Error("synthetic PGlite failure"); } return real(...args); }; await expect(first.unsafe(`select 2 as two`)).rejects.toThrow(); // The poisoned entry must take down only its own connection: a fresh // client (new socket, full startup handshake) still gets served. const second = makeClient(port); // oxlint-disable-next-line executor/no-try-catch-or-throw -- test boundary: sockets must be closed on every path try { expect((await second.unsafe(`select 3 as three`))[0]).toEqual({ three: 3 }); } finally { // oxlint-disable-next-line executor/no-promise-catch -- test boundary: a failed teardown must not mask the assertion await second.end({ timeout: 5 }).catch(() => {}); } } finally { // oxlint-disable-next-line executor/no-promise-catch -- test boundary: a failed teardown must not mask the assertion await first.end({ timeout: 5 }).catch(() => {}); await server.stop(); await db.close(); } }, ); // Regression for the sporadic `write CONNECTION_ENDED` 500s: the server's // idleTimeout backstop used to kill ANY connection with no traffic for the // window, which is the resting state of every healthy postgres.js pool // connection (idle_timeout: 0) held by a long-lived scope. The backstop now // only fires on a connection that is actually blocking the shared session — // an open pipeline or an open transaction. it("an idle-at-rest connection outlives the idle backstop", { timeout: 30_000 }, async () => { const port = 45996; const db = await PGlite.create(); const server = new PGLiteSocketServer({ db, port, host: "127.0.0.1", maxConnections: 100, idleTimeout: 250, }); await server.start(); const sql = makeClient(port); // oxlint-disable-next-line executor/no-try-catch-or-throw -- test boundary: sockets must be closed on every path try { expect((await sql.unsafe(`select 1 as one`))[0]).toEqual({ one: 1 }); await sleep(900); expect((await sql.unsafe(`select 2 as two`))[0]).toEqual({ two: 2 }); } finally { // oxlint-disable-next-line executor/no-promise-catch -- test boundary: a failed teardown must not mask the assertion await sql.end({ timeout: 5 }).catch(() => {}); await server.stop(); await db.close(); } }); // The backstop's actual job still works: a client that opens a pipeline // (Parse sent, never Sync) and goes silent holds queue affinity, which // starves every other connection. The idle timer must reap exactly that // client and hand the queue back. it( "a client stalled mid-pipeline is reaped and the queue recovers", { timeout: 30_000 }, async () => { const port = 45995; const db = await PGlite.create(); const server = new PGLiteSocketServer({ db, port, host: "127.0.0.1", maxConnections: 100, idleTimeout: 250, }); await server.start(); // Hand-rolled wire client: complete the trust-auth startup, then send a // lone Parse. Its last frame type ('P') marks the pipeline open, so the // handler takes affinity and every other connection queues behind it. const staller = await openWireClient(port); staller.write(parseFrame("select 1")); const bystander = makeClient(port, 10); // oxlint-disable-next-line executor/no-try-catch-or-throw -- test boundary: sockets must be closed on every path try { // Connects and queries only once the staller is reaped (~250ms). expect((await bystander.unsafe(`select 4 as four`))[0]).toEqual({ four: 4 }); } finally { // oxlint-disable-next-line executor/no-promise-catch -- test boundary: a failed teardown must not mask the assertion await bystander.end({ timeout: 5 }).catch(() => {}); staller.destroy(); await server.stop(); await db.close(); } }, ); // Regression for the second wedge mode behind the same CI cascade: a client // whose socket dies WHILE its pipeline-opening entry is executing. detach() // clears pipeline affinity before the entry finishes, so the queue then // assigned affinity to the already-dead handler — and nothing ever cleared // it: the dead handler has no timers left, and every other connection // (including fresh startups) queued behind the ghost forever. The queue now // tracks detached handlers and repairs affinity they can no longer release. it( "a client that dies mid-execution does not leave the queue pinned to its ghost", { timeout: 30_000 }, async () => { const port = 45994; const db = await PGlite.create(); // Hold the marker query in flight long enough that the disconnect below // reliably lands while the entry is EXECUTING (after detach's cleanup, // before the queue takes affinity for it). const real = db.execProtocolRawStream.bind(db); (db as { execProtocolRawStream: typeof real }).execProtocolRawStream = async (...args) => { if (Buffer.from(args[0]).includes("ghost_marker")) await sleep(300); return real(...args); }; const server = new PGLiteSocketServer({ db, port, host: "127.0.0.1", maxConnections: 100 }); await server.start(); const ghost = await openWireClient(port); ghost.write(parseFrame("select 'ghost_marker'")); // Give the data event time to reach the queue and start executing, then // die without a trace mid-flight. await sleep(100); ghost.destroy(); const bystander = makeClient(port); // oxlint-disable-next-line executor/no-try-catch-or-throw -- test boundary: sockets must be closed on every path try { expect((await bystander.unsafe(`select 5 as five`))[0]).toEqual({ five: 5 }); } finally { // oxlint-disable-next-line executor/no-promise-catch -- test boundary: a failed teardown must not mask the assertion await bystander.end({ timeout: 5 }).catch(() => {}); await server.stop(); await db.close(); } }, );});
// Regression for the dev-db PGlite socket protocol-interleaving bug (patched
// in patches/@electric-sql%2Fpglite-socket@0.1.4.patch).
//
// PGLiteSocketServer's QueryQueueManager used to enqueue each postgres wire
// FRAME (Parse, Bind, Execute, Sync) as its own queue entry against the one
// shared PGlite session. With more than one connection (the dev-db now allows
// many — see scripts/dev-db.ts maxConnections), two clients' extended-protocol
// pipelines interleaved: client A's Parse (5 params) ... client B's Parse
// (1 param) ... A's Bind now hits B's unnamed statement:
//
// PostgresError: bind message supplies 5 parameters, but prepared
// statement "" requires 1
//
// which surfaced in e2e as random 500s ("Failed to load tools", StorageError)
// on whichever request lost the race — the residual per-spec CI flakes after
// the connection-storm fix. The patch batches all frames of one socket data
// event into a single queue entry and adds handler affinity while a pipeline
// is open, so one client's Parse..Sync executes atomically.
//
// This test drives concurrent clients issuing unprepared parameterized queries
// with DIFFERENT parameter counts (the exact drizzle/postgres-js shape) through
// one PGLiteSocketServer and asserts zero protocol corruption.
import { setTimeout as sleep } from "node:timers/promises";
import { connect, type Socket } from "node:net";
import { describe, expect, it } from "@effect/vitest";
import { PGlite } from "@electric-sql/pglite";
import { PGLiteSocketServer } from "@electric-sql/pglite-socket";
import postgres from "postgres";
const PORT = 45998;
const CLIENTS = 6;
const QUERIES_PER_CLIENT = 40;
const makeClient = (port: number, connectTimeout = 5) =>
postgres(`postgres://postgres:postgres@127.0.0.1:${port}/postgres`, {
max: 1,
idle_timeout: 0,
connect_timeout: connectTimeout,
fetch_types: false,
prepare: true,
onnotice: () => undefined,
});
// Hand-rolled wire client: connect and complete the trust-auth startup, so a
// test can then speak raw protocol frames (e.g. a lone Parse) that postgres.js
// would never emit on its own. Resolves after ReadyForQuery so the next write
// is its own data event — and its own queue entry — on the server.
const openWireClient = async (port: number): Promise<Socket> => {
const socket: Socket = connect(port, "127.0.0.1");
await new Promise<void>((res, rej) => {
socket.once("connect", res);
socket.once("error", rej);
});
const startupBody = Buffer.concat([
Buffer.from([0, 3, 0, 0]),
Buffer.from("user\0postgres\0database\0postgres\0\0"),
]);
const startup = Buffer.concat([Buffer.alloc(4), startupBody]);
startup.writeInt32BE(startup.length, 0);
socket.write(startup);
await new Promise<void>((res) => {
socket.on("data", (chunk: Buffer) => {
if (chunk.includes(0x5a)) res(); // 'Z' = ReadyForQuery
});
});
return socket;
};
// A Parse frame for an unnamed statement: opens an extended-protocol pipeline
// that only a later Sync (or the server's recovery) closes.
const parseFrame = (query: string): Buffer => {
const body = Buffer.concat([Buffer.from(`\0${query}\0`), Buffer.from([0, 0])]);
const frame = Buffer.concat([Buffer.from("P"), Buffer.alloc(4), body]);
frame.writeInt32BE(4 + body.length, 1);
return frame;
};
describe("dev-db PGlite socket under concurrent connections", () => {
it(
"serves interleaved multi-connection pipelines without protocol corruption",
{ timeout: 60_000 },
async () => {
const db = await PGlite.create();
const server = new PGLiteSocketServer({
db,
port: PORT,
host: "127.0.0.1",
maxConnections: 100,
});
await server.start();
let ok = 0;
const errors: string[] = [];
const worker = async (id: number) => {
const sql = makeClient(PORT, 10);
// oxlint-disable-next-line executor/no-try-catch-or-throw -- test boundary: postgres.js is promise-native and the socket must be closed on every path
try {
for (let q = 0; q < QUERIES_PER_CLIENT; q++) {
// Alternate 1-param and 5-param unprepared queries: maximally
// collision-prone unnamed-statement shapes across connections.
if ((id + q) % 2 === 0) {
await sql.unsafe(`select $1::int as one`, [1]);
} else {
await sql.unsafe(`select $1::int, $2::text, $3::text, $4::text, $5::text`, [
1,
"b",
"c",
"d",
"e",
]);
}
ok++;
}
} catch (cause) {
// oxlint-disable-next-line executor/no-unknown-error-message -- test boundary: the raw PostgresError message IS the assertion payload
errors.push(String(cause));
} finally {
// oxlint-disable-next-line executor/no-promise-catch -- test boundary: postgres.js is promise-native; a failed teardown must not mask the assertion
await sql.end({ timeout: 5 }).catch(() => {});
}
};
await Promise.all(Array.from({ length: CLIENTS }, (_, i) => worker(i)));
await server.stop();
await db.close();
expect(errors, `protocol corruption under concurrency:\n${errors.join("\n")}`).toEqual([]);
expect(ok).toBe(CLIENTS * QUERIES_PER_CLIENT);
},
);
// Regression for the CI e2e "cloud signIn: callback set no session (500)"
// cascade: QueryQueueManager.processQueue used to `return` out of its drain
// loop when a query REJECTED (as opposed to returning a wire-level
// ErrorResponse), leaving `processing` latched true. From then on every
// enqueue — including brand-new connections' startup packets — sat in the
// queue forever: in-flight requests hung, postgres.js reconnects died with
// CONNECT_TIMEOUT, and the whole dev stack was bricked until restart. The
// patch rejects the one entry, drops pipeline affinity, and keeps draining.
it(
"a rejected query fails one client, not the whole socket server",
{ timeout: 30_000 },
async () => {
const port = 45997;
const db = await PGlite.create();
const server = new PGLiteSocketServer({ db, port, host: "127.0.0.1", maxConnections: 100 });
await server.start();
const first = makeClient(port);
// oxlint-disable-next-line executor/no-try-catch-or-throw -- test boundary: sockets must be closed on every path
try {
expect((await first.unsafe(`select 1 as one`))[0]).toEqual({ one: 1 });
// Force the NEXT protocol exchange to reject at the JS level, the shape
// PGlite produces when the shared session is broken mid-run.
const real = db.execProtocolRawStream.bind(db);
let arm = true;
(db as { execProtocolRawStream: typeof real }).execProtocolRawStream = (...args) => {
if (arm) {
arm = false;
// oxlint-disable-next-line executor/no-try-catch-or-throw, executor/no-error-constructor -- test boundary: simulating a PGlite internal failure requires a raw throw
throw new Error("synthetic PGlite failure");
}
return real(...args);
};
await expect(first.unsafe(`select 2 as two`)).rejects.toThrow();
// The poisoned entry must take down only its own connection: a fresh
// client (new socket, full startup handshake) still gets served.
const second = makeClient(port);
// oxlint-disable-next-line executor/no-try-catch-or-throw -- test boundary: sockets must be closed on every path
try {
expect((await second.unsafe(`select 3 as three`))[0]).toEqual({ three: 3 });
} finally {
// oxlint-disable-next-line executor/no-promise-catch -- test boundary: a failed teardown must not mask the assertion
await second.end({ timeout: 5 }).catch(() => {});
}
} finally {
// oxlint-disable-next-line executor/no-promise-catch -- test boundary: a failed teardown must not mask the assertion
await first.end({ timeout: 5 }).catch(() => {});
await server.stop();
await db.close();
}
},
);
// Regression for the sporadic `write CONNECTION_ENDED` 500s: the server's
// idleTimeout backstop used to kill ANY connection with no traffic for the
// window, which is the resting state of every healthy postgres.js pool
// connection (idle_timeout: 0) held by a long-lived scope. The backstop now
// only fires on a connection that is actually blocking the shared session —
// an open pipeline or an open transaction.
it("an idle-at-rest connection outlives the idle backstop", { timeout: 30_000 }, async () => {
const port = 45996;
const db = await PGlite.create();
const server = new PGLiteSocketServer({
db,
port,
host: "127.0.0.1",
maxConnections: 100,
idleTimeout: 250,
});
await server.start();
const sql = makeClient(port);
// oxlint-disable-next-line executor/no-try-catch-or-throw -- test boundary: sockets must be closed on every path
try {
expect((await sql.unsafe(`select 1 as one`))[0]).toEqual({ one: 1 });
await sleep(900);
expect((await sql.unsafe(`select 2 as two`))[0]).toEqual({ two: 2 });
} finally {
// oxlint-disable-next-line executor/no-promise-catch -- test boundary: a failed teardown must not mask the assertion
await sql.end({ timeout: 5 }).catch(() => {});
await server.stop();
await db.close();
}
});
// The backstop's actual job still works: a client that opens a pipeline
// (Parse sent, never Sync) and goes silent holds queue affinity, which
// starves every other connection. The idle timer must reap exactly that
// client and hand the queue back.
it(
"a client stalled mid-pipeline is reaped and the queue recovers",
{ timeout: 30_000 },
async () => {
const port = 45995;
const db = await PGlite.create();
const server = new PGLiteSocketServer({
db,
port,
host: "127.0.0.1",
maxConnections: 100,
idleTimeout: 250,
});
await server.start();
// Hand-rolled wire client: complete the trust-auth startup, then send a
// lone Parse. Its last frame type ('P') marks the pipeline open, so the
// handler takes affinity and every other connection queues behind it.
const staller = await openWireClient(port);
staller.write(parseFrame("select 1"));
const bystander = makeClient(port, 10);
// oxlint-disable-next-line executor/no-try-catch-or-throw -- test boundary: sockets must be closed on every path
try {
// Connects and queries only once the staller is reaped (~250ms).
expect((await bystander.unsafe(`select 4 as four`))[0]).toEqual({ four: 4 });
} finally {
// oxlint-disable-next-line executor/no-promise-catch -- test boundary: a failed teardown must not mask the assertion
await bystander.end({ timeout: 5 }).catch(() => {});
staller.destroy();
await server.stop();
await db.close();
}
},
);
// Regression for the second wedge mode behind the same CI cascade: a client
// whose socket dies WHILE its pipeline-opening entry is executing. detach()
// clears pipeline affinity before the entry finishes, so the queue then
// assigned affinity to the already-dead handler — and nothing ever cleared
// it: the dead handler has no timers left, and every other connection
// (including fresh startups) queued behind the ghost forever. The queue now
// tracks detached handlers and repairs affinity they can no longer release.
it(
"a client that dies mid-execution does not leave the queue pinned to its ghost",
{ timeout: 30_000 },
async () => {
const port = 45994;
const db = await PGlite.create();
// Hold the marker query in flight long enough that the disconnect below
// reliably lands while the entry is EXECUTING (after detach's cleanup,
// before the queue takes affinity for it).
const real = db.execProtocolRawStream.bind(db);
(db as { execProtocolRawStream: typeof real }).execProtocolRawStream = async (...args) => {
if (Buffer.from(args[0]).includes("ghost_marker")) await sleep(300);
return real(...args);
};
const server = new PGLiteSocketServer({ db, port, host: "127.0.0.1", maxConnections: 100 });
await server.start();
const ghost = await openWireClient(port);
ghost.write(parseFrame("select 'ghost_marker'"));
// Give the data event time to reach the queue and start executing, then
// die without a trace mid-flight.
await sleep(100);
ghost.destroy();
const bystander = makeClient(port);
// oxlint-disable-next-line executor/no-try-catch-or-throw -- test boundary: sockets must be closed on every path
try {
expect((await bystander.unsafe(`select 5 as five`))[0]).toEqual({ five: 5 });
} finally {
// oxlint-disable-next-line executor/no-promise-catch -- test boundary: a failed teardown must not mask the assertion
await bystander.end({ timeout: 5 }).catch(() => {});
await server.stop();
await db.close();
}
},
);
});