From 4aaa87118e33c55a4fd13579d687b75dfd79b91d Mon Sep 17 00:00:00 2001 From: riderx Date: Tue, 7 Jul 2026 14:08:06 +0200 Subject: [PATCH 1/2] Enable logical replication slots - Start Postgres with wal_level=logical in defaultStartParams so pg_create_logical_replication_slot works out of the box. - Surface unexpected WASM exceptions from PostgresMainLoopOnce instead of swallowing them: previously a crash (e.g. a call into a missing symbol) was silently ignored, leaving the session returning empty results until it hit 'ERRORDATA_STACK_SIZE exceeded'. - Add tests covering slot creation/listing/drop, decoding changes with pgoutput, error recovery, and slot persistence across dump/load. Requires the postgres-pglite change that exports the symbols pgoutput.so imports from the main module (electric-sql/postgres-pglite side of this fix). Fixes #880 Co-Authored-By: Claude Fable 5 --- .changeset/heavy-pandas-repeat.md | 5 + docs/docs/api.md | 2 +- packages/pglite/src/pglite.ts | 8 ++ .../pglite/tests/replication-slots.test.ts | 134 ++++++++++++++++++ 4 files changed, 148 insertions(+), 1 deletion(-) create mode 100644 .changeset/heavy-pandas-repeat.md create mode 100644 packages/pglite/tests/replication-slots.test.ts diff --git a/.changeset/heavy-pandas-repeat.md b/.changeset/heavy-pandas-repeat.md new file mode 100644 index 000000000..5ca3898a4 --- /dev/null +++ b/.changeset/heavy-pandas-repeat.md @@ -0,0 +1,5 @@ +--- +'@electric-sql/pglite': patch +--- + +Support logical replication slots: start Postgres with `wal_level=logical` by default, export the symbols the `pgoutput` output plugin needs from the WASM main module, and surface unexpected WASM crashes as errors instead of silently corrupting the session. diff --git a/docs/docs/api.md b/docs/docs/api.md index c2e7d2785..461afb1d5 100644 --- a/docs/docs/api.md +++ b/docs/docs/api.md @@ -91,7 +91,7 @@ Path to the directory for storing the Postgres database. You can provide a URI s ``` - `startParams?: string[]`
- An array of strings that will be passed to the Postgres process as start parameters. This is the set of parameters one would pass to a native PostgreSQL instance. + An array of strings that will be passed to the Postgres process as start parameters. This is the set of parameters one would pass to a native PostgreSQL instance. When not provided, `PGlite.defaultStartParams` is used, which starts Postgres with `wal_level=logical` so that logical replication slots (e.g. `pg_create_logical_replication_slot` with the `pgoutput` plugin) work out of the box. ```ts import { PGlite } from '@electric-sql/pglite' diff --git a/packages/pglite/src/pglite.ts b/packages/pglite/src/pglite.ts index e23ac32c4..200e88bf6 100644 --- a/packages/pglite/src/pglite.ts +++ b/packages/pglite/src/pglite.ts @@ -166,6 +166,8 @@ export class PGlite 'io_method=sync', '-c', 'max_parallel_maintenance_workers=0', + '-c', + 'wal_level=logical', // enable logical decoding / replication slots ] /** @@ -940,6 +942,12 @@ export class PGlite // that we call whenever the exception longjmp is executed // like this we also just need to setjmp only once, in a similar fashion to the original code. mod._PostgresMainLongJmp() + } else if (typeof e?.status !== 'number') { + // not an emscripten ExitStatus: the wasm instance crashed + // (e.g. a trap or a call to a missing symbol) and its internal + // state can no longer be trusted. Swallowing it here would leave + // the session silently corrupted, so surface it to the caller. + throw e } // even if there is an exception caused by one of the batched queries, // we need to continue processing the rest without throwing. diff --git a/packages/pglite/tests/replication-slots.test.ts b/packages/pglite/tests/replication-slots.test.ts new file mode 100644 index 000000000..44b8a3485 --- /dev/null +++ b/packages/pglite/tests/replication-slots.test.ts @@ -0,0 +1,134 @@ +import { describe, it, expect, beforeEach, afterEach } from 'vitest' +import { PGlite } from '../dist/index.js' + +describe('replication slots', () => { + let db: PGlite + + beforeEach(async () => { + db = await PGlite.create() + }) + + afterEach(async () => { + if (!db.closed) { + await db.close() + } + }) + + it('wal_level defaults to logical', async () => { + const res = await db.query<{ wal_level: string }>(`SHOW wal_level`) + expect(res.rows[0].wal_level).toBe('logical') + }) + + it('can create, list and drop a logical replication slot', async () => { + const created = await db.query<{ slot_name: string; lsn: string }>( + `SELECT * FROM pg_create_logical_replication_slot('test_slot', 'pgoutput')`, + ) + expect(created.rows[0].slot_name).toBe('test_slot') + expect(created.rows[0].lsn).toBeTruthy() + + const slots = await db.query<{ slot_name: string; plugin: string }>( + `SELECT slot_name, plugin FROM pg_replication_slots`, + ) + expect(slots.rows).toEqual([{ slot_name: 'test_slot', plugin: 'pgoutput' }]) + + await db.query(`SELECT pg_drop_replication_slot('test_slot')`) + const after = await db.query(`SELECT slot_name FROM pg_replication_slots`) + expect(after.rows).toEqual([]) + }) + + it('decodes changes via pgoutput', async () => { + await db.exec(` + CREATE TABLE rep_test (id int PRIMARY KEY, value text); + CREATE PUBLICATION rep_pub FOR ALL TABLES; + `) + // slot creation is not allowed in a transaction that has performed + // writes, so it can't be part of the exec batch above + await db.query( + `SELECT pg_create_logical_replication_slot('rep_slot', 'pgoutput')`, + ) + + await db.exec(` + INSERT INTO rep_test VALUES (1, 'one'), (2, 'two'); + UPDATE rep_test SET value = 'uno' WHERE id = 1; + DELETE FROM rep_test WHERE id = 2; + `) + + // peek first - changes must remain available afterwards + const peeked = await db.query<{ data: Uint8Array }>( + `SELECT data FROM pg_logical_slot_peek_binary_changes( + 'rep_slot', NULL, NULL, + 'proto_version', '1', 'publication_names', 'rep_pub')`, + ) + expect(peeked.rows.length).toBeGreaterThan(0) + + // messages start with a tag byte; expect insert (I), update (U) and delete (D) + const tags = peeked.rows.map((r) => String.fromCharCode(r.data[0])) + expect(tags).toContain('I') + expect(tags).toContain('U') + expect(tags).toContain('D') + + // consuming advances the slot + const consumed = await db.query( + `SELECT lsn FROM pg_logical_slot_get_binary_changes( + 'rep_slot', NULL, NULL, + 'proto_version', '1', 'publication_names', 'rep_pub')`, + ) + expect(consumed.rows.length).toBe(peeked.rows.length) + + const empty = await db.query( + `SELECT lsn FROM pg_logical_slot_get_binary_changes( + 'rep_slot', NULL, NULL, + 'proto_version', '1', 'publication_names', 'rep_pub')`, + ) + expect(empty.rows).toEqual([]) + }) + + it('recovers cleanly from a decoding error', async () => { + await db.query( + `SELECT pg_create_logical_replication_slot('err_slot', 'pgoutput')`, + ) + + // pgoutput requires publication_names; this must fail with a proper error + await expect( + db.query( + `SELECT * FROM pg_logical_slot_get_binary_changes('err_slot', NULL, NULL, 'proto_version', '1')`, + ), + ).rejects.toThrow(/publication_names/) + + // and the session must still be usable afterwards + const res = await db.query<{ one: number }>(`SELECT 1 AS one`) + expect(res.rows[0].one).toBe(1) + + const slots = await db.query<{ slot_name: string; active: boolean }>( + `SELECT slot_name, active FROM pg_replication_slots`, + ) + expect(slots.rows).toEqual([{ slot_name: 'err_slot', active: false }]) + }) + + it('slots survive a dump/load cycle', async () => { + await db.exec(` + CREATE TABLE rep_persist (id int PRIMARY KEY); + CREATE PUBLICATION persist_pub FOR ALL TABLES; + `) + await db.query( + `SELECT pg_create_logical_replication_slot('persist_slot', 'pgoutput')`, + ) + const dump = await db.dumpDataDir('none') + await db.close() + + db = await PGlite.create({ loadDataDir: dump }) + const slots = await db.query<{ slot_name: string }>( + `SELECT slot_name FROM pg_replication_slots`, + ) + expect(slots.rows).toEqual([{ slot_name: 'persist_slot' }]) + + // the restored slot must still be usable + await db.exec(`INSERT INTO rep_persist VALUES (1)`) + const changes = await db.query( + `SELECT lsn FROM pg_logical_slot_get_binary_changes( + 'persist_slot', NULL, NULL, + 'proto_version', '1', 'publication_names', 'persist_pub')`, + ) + expect(changes.rows.length).toBeGreaterThan(0) + }) +}) From 6d55f6ccdcc19ad7cecfe26e547afd1bf0302c14 Mon Sep 17 00:00:00 2001 From: riderx Date: Tue, 7 Jul 2026 15:00:32 +0200 Subject: [PATCH 2/2] Bump postgres-pglite to include pgoutput symbol exports Points at the head of electric-sql/postgres-pglite#79 (served by the upstream repo via refs/pull/79/head, so submodule checkout by SHA works). Should be updated to the merge commit once that PR lands. Co-Authored-By: Claude Fable 5 --- postgres-pglite | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/postgres-pglite b/postgres-pglite index 7b4ee5086..04f1bea67 160000 --- a/postgres-pglite +++ b/postgres-pglite @@ -1 +1 @@ -Subproject commit 7b4ee5086055dc5e54ae1e13e487888249438e68 +Subproject commit 04f1bea671eb6c285a3eb2f6c45ab2d545866d7f