Skip to content

Commit 91bd413

Browse files
fix: sync to the filesystem when a transaction ends
Every top-level query/exec ends with syncToFs(), gated on not being inside a transaction — but transaction() executed its terminal COMMIT/ROLLBACK while #inTransaction still suppressed that gate and cleared the flag only afterwards. A resolved transaction() had therefore neither performed nor scheduled any filesystem sync: with relaxedDurability: false the awaited durability promise was silently broken, and with relaxedDurability: true no background sync was even scheduled until some later unrelated query ran. Clear the flag before the terminal statement so #runExec ends the transaction with the same synchronization as a top-level exec, and sync explicitly on the tx.rollback() path, which exits the wrapper without running a terminal statement. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
1 parent 25d0a55 commit 91bd413

3 files changed

Lines changed: 154 additions & 2 deletions

File tree

.changeset/transaction-end-sync.md

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,5 @@
1+
---
2+
'@electric-sql/pglite': patch
3+
---
4+
5+
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

Lines changed: 22 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -511,17 +511,37 @@ export abstract class BasePGlite
511511

512512
try {
513513
const result = await callback(tx)
514+
// Clear the flag before the terminal statement so #runExec ends the
515+
// transaction with the same syncToFs() as any top-level exec —
516+
// otherwise a committed transaction is not persisted (or scheduled
517+
// for persistence) until some later unrelated query runs.
518+
this.#inTransaction = false
514519
if (!closed) {
515520
closed = true
516521
await this.#runExec('COMMIT')
522+
} else {
523+
// The transaction was closed by an explicit tx.rollback(), which
524+
// ran under the in-transaction gate; sync its result now.
525+
await this.syncToFs()
517526
}
518-
this.#inTransaction = false
519527
return result
520528
} catch (e) {
529+
this.#inTransaction = false
521530
if (!closed) {
522531
await this.#runExec('ROLLBACK')
532+
} else {
533+
// The transaction already ended without reaching a sync: either an
534+
// explicit tx.rollback() ran under the in-transaction gate, or the
535+
// terminal COMMIT threw before #runExec reached its syncToFs().
536+
// Still end at an awaited sync boundary, but never mask the
537+
// original error with a sync failure — a failing filesystem
538+
// surfaces again on the next operation's own sync.
539+
try {
540+
await this.syncToFs()
541+
} catch {
542+
// the original error takes precedence
543+
}
523544
}
524-
this.#inTransaction = false
525545
throw e
526546
}
527547
})
Lines changed: 127 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,127 @@
1+
import { describe, expect, it } from 'vitest'
2+
import { MemoryFS, PGlite } from '../dist/index.js'
3+
4+
class CountingFS extends MemoryFS {
5+
syncCalls = 0
6+
failSyncs = false
7+
8+
override async syncToFs(relaxedDurability?: boolean): Promise<void> {
9+
this.syncCalls += 1
10+
if (this.failSyncs) {
11+
throw new Error('sync failed')
12+
}
13+
await super.syncToFs(relaxedDurability)
14+
}
15+
}
16+
17+
describe('transaction end synchronization', () => {
18+
it('syncs to the filesystem after COMMIT before transaction() resolves', async () => {
19+
const fs = new CountingFS()
20+
const pg = await PGlite.create({ fs })
21+
await pg.exec('CREATE TABLE t (v int)')
22+
23+
let syncsAtCallbackEnd = -1
24+
await pg.transaction(async (tx) => {
25+
await tx.exec('INSERT INTO t VALUES (1)')
26+
syncsAtCallbackEnd = fs.syncCalls
27+
})
28+
// The terminal COMMIT must end with the same awaited sync as a top-level
29+
// exec; without it a committed transaction is not persisted until some
30+
// later unrelated query runs.
31+
expect(fs.syncCalls).toBeGreaterThan(syncsAtCallbackEnd)
32+
await pg.close()
33+
})
34+
35+
it('syncs after an explicit tx.rollback()', async () => {
36+
const fs = new CountingFS()
37+
const pg = await PGlite.create({ fs })
38+
await pg.exec('CREATE TABLE t (v int)')
39+
40+
let syncsAtCallbackEnd = -1
41+
await pg.transaction(async (tx) => {
42+
await tx.exec('INSERT INTO t VALUES (1)')
43+
await tx.rollback()
44+
syncsAtCallbackEnd = fs.syncCalls
45+
})
46+
expect(fs.syncCalls).toBeGreaterThan(syncsAtCallbackEnd)
47+
await pg.close()
48+
})
49+
50+
it('syncs after the ROLLBACK issued for a throwing callback', async () => {
51+
const fs = new CountingFS()
52+
const pg = await PGlite.create({ fs })
53+
await pg.exec('CREATE TABLE t (v int)')
54+
55+
let syncsAtCallbackEnd = -1
56+
await expect(
57+
pg.transaction(async (tx) => {
58+
await tx.exec('INSERT INTO t VALUES (1)')
59+
syncsAtCallbackEnd = fs.syncCalls
60+
throw new Error('force rollback')
61+
}),
62+
).rejects.toThrow('force rollback')
63+
expect(fs.syncCalls).toBeGreaterThan(syncsAtCallbackEnd)
64+
await pg.close()
65+
})
66+
67+
it('syncs after an explicit tx.rollback() followed by a throwing callback', async () => {
68+
const fs = new CountingFS()
69+
const pg = await PGlite.create({ fs })
70+
await pg.exec('CREATE TABLE t (v int)')
71+
72+
let syncsAtCallbackEnd = -1
73+
await expect(
74+
pg.transaction(async (tx) => {
75+
await tx.exec('INSERT INTO t VALUES (1)')
76+
await tx.rollback()
77+
syncsAtCallbackEnd = fs.syncCalls
78+
throw new Error('after explicit rollback')
79+
}),
80+
).rejects.toThrow('after explicit rollback')
81+
expect(fs.syncCalls).toBeGreaterThan(syncsAtCallbackEnd)
82+
await pg.close()
83+
})
84+
85+
it('syncs when the terminal COMMIT itself fails', async () => {
86+
const fs = new CountingFS()
87+
const pg = await PGlite.create({ fs })
88+
await pg.exec(`
89+
CREATE TABLE parent (id int PRIMARY KEY);
90+
CREATE TABLE child (
91+
pid int REFERENCES parent (id) DEFERRABLE INITIALLY DEFERRED
92+
);
93+
`)
94+
95+
let syncsAtCallbackEnd = -1
96+
await expect(
97+
pg.transaction(async (tx) => {
98+
// Violates the deferred constraint only at COMMIT, so the terminal
99+
// COMMIT throws and Postgres rolls the transaction back implicitly.
100+
await tx.exec('INSERT INTO child VALUES (42)')
101+
syncsAtCallbackEnd = fs.syncCalls
102+
}),
103+
).rejects.toThrow(/violates foreign key constraint/)
104+
expect(fs.syncCalls).toBeGreaterThan(syncsAtCallbackEnd)
105+
await pg.close()
106+
})
107+
108+
it('does not mask the callback error when the terminal sync fails', async () => {
109+
const fs = new CountingFS()
110+
const pg = await PGlite.create({ fs })
111+
await pg.exec('CREATE TABLE t (v int)')
112+
113+
await expect(
114+
pg.transaction(async (tx) => {
115+
await tx.rollback()
116+
fs.failSyncs = true
117+
throw new Error('callback cause')
118+
}),
119+
).rejects.toThrow('callback cause')
120+
121+
// The sync failure surfaces on the next operation rather than masking
122+
// the callback error above.
123+
await expect(pg.query('SELECT 1')).rejects.toThrow('sync failed')
124+
fs.failSyncs = false
125+
await pg.close().catch(() => {})
126+
})
127+
})

0 commit comments

Comments
 (0)