diff --git a/.changeset/transaction-end-sync.md b/.changeset/transaction-end-sync.md new file mode 100644 index 000000000..3b449b95e --- /dev/null +++ b/.changeset/transaction-end-sync.md @@ -0,0 +1,5 @@ +--- +'@electric-sql/pglite': patch +--- + +Sync to the filesystem when a transaction ends. `transaction()` executed its terminal `COMMIT`/`ROLLBACK` while the in-transaction flag still suppressed the per-exec `syncToFs()`, and cleared the flag only afterwards — so a resolved `transaction()` had neither performed nor scheduled any filesystem sync, and a committed transaction was not persisted until some later unrelated query ran. The transaction now ends with the same synchronization as a top-level exec, on every terminal path: commit, rollback, explicit `tx.rollback()` (whether the callback then returns or throws), and a terminal `COMMIT` that itself fails (e.g. a deferred constraint violation). A failure in that final sync never masks the transaction's own error. diff --git a/packages/pglite/src/base.ts b/packages/pglite/src/base.ts index e18efb911..364b78a22 100644 --- a/packages/pglite/src/base.ts +++ b/packages/pglite/src/base.ts @@ -511,17 +511,37 @@ export abstract class BasePGlite try { const result = await callback(tx) + // Clear the flag before the terminal statement so #runExec ends the + // transaction with the same syncToFs() as any top-level exec — + // otherwise a committed transaction is not persisted (or scheduled + // for persistence) until some later unrelated query runs. + this.#inTransaction = false if (!closed) { closed = true await this.#runExec('COMMIT') + } else { + // The transaction was closed by an explicit tx.rollback(), which + // ran under the in-transaction gate; sync its result now. + await this.syncToFs() } - this.#inTransaction = false return result } catch (e) { + this.#inTransaction = false if (!closed) { await this.#runExec('ROLLBACK') + } else { + // The transaction already ended without reaching a sync: either an + // explicit tx.rollback() ran under the in-transaction gate, or the + // terminal COMMIT threw before #runExec reached its syncToFs(). + // Still end at an awaited sync boundary, but never mask the + // original error with a sync failure — a failing filesystem + // surfaces again on the next operation's own sync. + try { + await this.syncToFs() + } catch { + // the original error takes precedence + } } - this.#inTransaction = false throw e } }) diff --git a/packages/pglite/tests/transaction-sync.test.ts b/packages/pglite/tests/transaction-sync.test.ts new file mode 100644 index 000000000..733e66e51 --- /dev/null +++ b/packages/pglite/tests/transaction-sync.test.ts @@ -0,0 +1,127 @@ +import { describe, expect, it } from 'vitest' +import { MemoryFS, PGlite } from '../dist/index.js' + +class CountingFS extends MemoryFS { + syncCalls = 0 + failSyncs = false + + override async syncToFs(relaxedDurability?: boolean): Promise { + this.syncCalls += 1 + if (this.failSyncs) { + throw new Error('sync failed') + } + await super.syncToFs(relaxedDurability) + } +} + +describe('transaction end synchronization', () => { + it('syncs to the filesystem after COMMIT before transaction() resolves', async () => { + const fs = new CountingFS() + const pg = await PGlite.create({ fs }) + await pg.exec('CREATE TABLE t (v int)') + + let syncsAtCallbackEnd = -1 + await pg.transaction(async (tx) => { + await tx.exec('INSERT INTO t VALUES (1)') + syncsAtCallbackEnd = fs.syncCalls + }) + // The terminal COMMIT must end with the same awaited sync as a top-level + // exec; without it a committed transaction is not persisted until some + // later unrelated query runs. + expect(fs.syncCalls).toBeGreaterThan(syncsAtCallbackEnd) + await pg.close() + }) + + it('syncs after an explicit tx.rollback()', async () => { + const fs = new CountingFS() + const pg = await PGlite.create({ fs }) + await pg.exec('CREATE TABLE t (v int)') + + let syncsAtCallbackEnd = -1 + await pg.transaction(async (tx) => { + await tx.exec('INSERT INTO t VALUES (1)') + await tx.rollback() + syncsAtCallbackEnd = fs.syncCalls + }) + expect(fs.syncCalls).toBeGreaterThan(syncsAtCallbackEnd) + await pg.close() + }) + + it('syncs after the ROLLBACK issued for a throwing callback', async () => { + const fs = new CountingFS() + const pg = await PGlite.create({ fs }) + await pg.exec('CREATE TABLE t (v int)') + + let syncsAtCallbackEnd = -1 + await expect( + pg.transaction(async (tx) => { + await tx.exec('INSERT INTO t VALUES (1)') + syncsAtCallbackEnd = fs.syncCalls + throw new Error('force rollback') + }), + ).rejects.toThrow('force rollback') + expect(fs.syncCalls).toBeGreaterThan(syncsAtCallbackEnd) + await pg.close() + }) + + it('syncs after an explicit tx.rollback() followed by a throwing callback', async () => { + const fs = new CountingFS() + const pg = await PGlite.create({ fs }) + await pg.exec('CREATE TABLE t (v int)') + + let syncsAtCallbackEnd = -1 + await expect( + pg.transaction(async (tx) => { + await tx.exec('INSERT INTO t VALUES (1)') + await tx.rollback() + syncsAtCallbackEnd = fs.syncCalls + throw new Error('after explicit rollback') + }), + ).rejects.toThrow('after explicit rollback') + expect(fs.syncCalls).toBeGreaterThan(syncsAtCallbackEnd) + await pg.close() + }) + + it('syncs when the terminal COMMIT itself fails', async () => { + const fs = new CountingFS() + const pg = await PGlite.create({ fs }) + await pg.exec(` + CREATE TABLE parent (id int PRIMARY KEY); + CREATE TABLE child ( + pid int REFERENCES parent (id) DEFERRABLE INITIALLY DEFERRED + ); + `) + + let syncsAtCallbackEnd = -1 + await expect( + pg.transaction(async (tx) => { + // Violates the deferred constraint only at COMMIT, so the terminal + // COMMIT throws and Postgres rolls the transaction back implicitly. + await tx.exec('INSERT INTO child VALUES (42)') + syncsAtCallbackEnd = fs.syncCalls + }), + ).rejects.toThrow(/violates foreign key constraint/) + expect(fs.syncCalls).toBeGreaterThan(syncsAtCallbackEnd) + await pg.close() + }) + + it('does not mask the callback error when the terminal sync fails', async () => { + const fs = new CountingFS() + const pg = await PGlite.create({ fs }) + await pg.exec('CREATE TABLE t (v int)') + + await expect( + pg.transaction(async (tx) => { + await tx.rollback() + fs.failSyncs = true + throw new Error('callback cause') + }), + ).rejects.toThrow('callback cause') + + // The sync failure surfaces on the next operation rather than masking + // the callback error above. + await expect(pg.query('SELECT 1')).rejects.toThrow('sync failed') + fs.failSyncs = false + await pg.close().catch(() => {}) + }) +})