fix: sync to the filesystem when a transaction ends

.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.

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
       }
     })

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<void> {
+    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(() => {})
+  })
+})