Skip to content

Commit 454f88a

Browse files
ChriztiaanChriztiaan
committed
Handling edge cases where subsets are updating while a mutation is pending.
1 parent 74eef39 commit 454f88a

2 files changed

Lines changed: 286 additions & 51 deletions

File tree

‎packages/powersync-db-collection/src/powersync.ts‎

Lines changed: 71 additions & 51 deletions
Original file line numberDiff line numberDiff line change
@@ -358,6 +358,55 @@ export function powerSyncCollectionOptions<
358358
})
359359
}
360360

361+
async function flushDiffRecords(): Promise<void> {
362+
await database
363+
.writeTransaction(async (context) => {
364+
begin()
365+
const operations = await context.getAll<TriggerDiffRecord>(
366+
`SELECT * FROM ${trackedTableName} ORDER BY timestamp ASC`,
367+
)
368+
const pendingOperations: Array<PendingOperation> = []
369+
370+
for (const op of operations) {
371+
const { id, operation, timestamp, value } = op
372+
const parsedValue = deserializeSyncRow({
373+
id,
374+
...JSON.parse(value),
375+
})
376+
const parsedPreviousValue =
377+
op.operation == DiffTriggerOperation.UPDATE
378+
? deserializeSyncRow({
379+
id,
380+
...JSON.parse(op.previous_value),
381+
})
382+
: undefined
383+
write({
384+
type: mapOperation(operation),
385+
value: parsedValue,
386+
previousValue: parsedPreviousValue,
387+
})
388+
pendingOperations.push({
389+
id,
390+
operation,
391+
timestamp,
392+
tableName: viewName,
393+
})
394+
}
395+
396+
// clear the current operations
397+
await context.execute(`DELETE FROM ${trackedTableName}`)
398+
399+
commit()
400+
pendingOperationStore.resolvePendingFor(pendingOperations)
401+
})
402+
.catch((error) => {
403+
database.logger.error(
404+
`An error has been detected in the sync handler`,
405+
error,
406+
)
407+
})
408+
}
409+
361410
// The sync function needs to be synchronous.
362411
async function start(afterOnChangeRegistered?: () => Promise<void>) {
363412
database.logger.info(
@@ -366,52 +415,7 @@ export function powerSyncCollectionOptions<
366415
database.onChangeWithCallback(
367416
{
368417
onChange: async () => {
369-
await database
370-
.writeTransaction(async (context) => {
371-
begin()
372-
const operations = await context.getAll<TriggerDiffRecord>(
373-
`SELECT * FROM ${trackedTableName} ORDER BY timestamp ASC`,
374-
)
375-
const pendingOperations: Array<PendingOperation> = []
376-
377-
for (const op of operations) {
378-
const { id, operation, timestamp, value } = op
379-
const parsedValue = deserializeSyncRow({
380-
id,
381-
...JSON.parse(value),
382-
})
383-
const parsedPreviousValue =
384-
op.operation == DiffTriggerOperation.UPDATE
385-
? deserializeSyncRow({
386-
id,
387-
...JSON.parse(op.previous_value),
388-
})
389-
: undefined
390-
write({
391-
type: mapOperation(operation),
392-
value: parsedValue,
393-
previousValue: parsedPreviousValue,
394-
})
395-
pendingOperations.push({
396-
id,
397-
operation,
398-
timestamp,
399-
tableName: viewName,
400-
})
401-
}
402-
403-
// clear the current operations
404-
await context.execute(`DELETE FROM ${trackedTableName}`)
405-
406-
commit()
407-
pendingOperationStore.resolvePendingFor(pendingOperations)
408-
})
409-
.catch((error) => {
410-
database.logger.error(
411-
`An error has been detected in the sync handler`,
412-
error,
413-
)
414-
})
418+
await flushDiffRecords()
415419
},
416420
},
417421
{
@@ -487,7 +491,11 @@ export function powerSyncCollectionOptions<
487491
// Tracks all active WHERE expressions for on-demand sync filtering.
488492
// Each loadSubset call pushes its predicate; unloadSubset removes it.
489493
const activeWhereExpressions: Array<LoadSubsetOptions['where']> = []
490-
const mutex = new Mutex()
494+
// Mutex for loadSubset() and unloadSubset() calls invoked by subset changes.
495+
const subsetMutex = new Mutex()
496+
497+
// Mutex for flushDiffRecords() and disposeTracking() calls
498+
const operationsMutex = new Mutex()
491499

492500
const loadSubset = async (
493501
options?: LoadSubsetOptions,
@@ -497,7 +505,13 @@ export function powerSyncCollectionOptions<
497505
}
498506

499507
if (activeWhereExpressions.length === 0) {
500-
await disposeTracking?.()
508+
await operationsMutex.runExclusive(async () => {
509+
await flushDiffRecords()
510+
})
511+
512+
await operationsMutex.runExclusive(async () => {
513+
await disposeTracking?.()
514+
})
501515
return
502516
}
503517

@@ -526,7 +540,13 @@ export function powerSyncCollectionOptions<
526540
const oldDataWhenClause = toInlinedWhereClause(compiledOldData)
527541
const viewWhereClause = toInlinedWhereClause(compiledView)
528542

529-
await disposeTracking?.()
543+
await operationsMutex.runExclusive(async () => {
544+
await flushDiffRecords()
545+
})
546+
547+
await operationsMutex.runExclusive(async () => {
548+
await disposeTracking?.()
549+
})
530550

531551
disposeTracking = await createDiffTrigger({
532552
when: {
@@ -615,9 +635,9 @@ export function powerSyncCollectionOptions<
615635
abortController.abort()
616636
},
617637
loadSubset: (options: LoadSubsetOptions) =>
618-
mutex.runExclusive(() => loadSubset(options)),
638+
subsetMutex.runExclusive(() => loadSubset(options)),
619639
unloadSubset: (options: LoadSubsetOptions) =>
620-
mutex.runExclusive(() => unloadSubset(options)),
640+
subsetMutex.runExclusive(() => unloadSubset(options)),
621641
}
622642
}
623643
},

‎packages/powersync-db-collection/tests/on-demand-sync.test.ts‎

Lines changed: 215 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1550,4 +1550,219 @@ describe(`On-Demand Sync Mode`, () => {
15501550
)
15511551
})
15521552
})
1553+
1554+
describe(`Pending mutations during filter changes`, () => {
1555+
it(`should resolve isPersisted when loadSubset is called during a pending mutation`, async () => {
1556+
const db = await createDatabase()
1557+
await createTestProducts(db)
1558+
1559+
const collection = createCollection(
1560+
powerSyncCollectionOptions({
1561+
database: db,
1562+
table: APP_SCHEMA.props.products,
1563+
syncMode: `on-demand`,
1564+
}),
1565+
)
1566+
onTestFinished(() => collection.cleanup())
1567+
await collection.stateWhenReady()
1568+
1569+
// LQ1: electronics category
1570+
const electronicsQuery = createLiveQueryCollection({
1571+
query: (q) =>
1572+
q
1573+
.from({ product: collection })
1574+
.where(({ product }) => eq(product.category, `electronics`))
1575+
.select(({ product }) => ({
1576+
id: product.id,
1577+
name: product.name,
1578+
price: product.price,
1579+
category: product.category,
1580+
})),
1581+
})
1582+
onTestFinished(() => electronicsQuery.cleanup())
1583+
1584+
await electronicsQuery.preload()
1585+
1586+
await vi.waitFor(
1587+
() => {
1588+
expect(electronicsQuery.size).toBe(3)
1589+
},
1590+
{ timeout: 2000 },
1591+
)
1592+
1593+
// Insert a new electronics product — creates a pending mutation
1594+
const insertResult = collection.insert({
1595+
id: randomUUID(),
1596+
name: `New Gadget`,
1597+
price: 99,
1598+
category: `electronics`,
1599+
})
1600+
1601+
// Immediately create a second live query for clothing — triggers loadSubset
1602+
// which rebuilds the diff trigger, potentially dropping unprocessed diff records
1603+
const clothingQuery = createLiveQueryCollection({
1604+
query: (q) =>
1605+
q
1606+
.from({ product: collection })
1607+
.where(({ product }) => eq(product.category, `clothing`))
1608+
.select(({ product }) => ({
1609+
id: product.id,
1610+
name: product.name,
1611+
price: product.price,
1612+
category: product.category,
1613+
})),
1614+
})
1615+
onTestFinished(() => clothingQuery.cleanup())
1616+
1617+
await clothingQuery.preload()
1618+
1619+
// isPersisted.promise should resolve — if the bug is present, this hangs forever
1620+
await vi.waitFor(
1621+
async () => {
1622+
await insertResult.isPersisted.promise
1623+
},
1624+
{ timeout: 5000 },
1625+
)
1626+
})
1627+
1628+
it(`should resolve isPersisted when unloadSubset is called during a pending mutation`, async () => {
1629+
const db = await createDatabase()
1630+
await createTestProducts(db)
1631+
1632+
const collection = createCollection(
1633+
powerSyncCollectionOptions({
1634+
database: db,
1635+
table: APP_SCHEMA.props.products,
1636+
syncMode: `on-demand`,
1637+
}),
1638+
)
1639+
onTestFinished(() => collection.cleanup())
1640+
await collection.stateWhenReady()
1641+
1642+
// LQ1: electronics category
1643+
const electronicsQuery = createLiveQueryCollection({
1644+
query: (q) =>
1645+
q
1646+
.from({ product: collection })
1647+
.where(({ product }) => eq(product.category, `electronics`))
1648+
.select(({ product }) => ({
1649+
id: product.id,
1650+
name: product.name,
1651+
price: product.price,
1652+
category: product.category,
1653+
})),
1654+
})
1655+
onTestFinished(() => electronicsQuery.cleanup())
1656+
1657+
await electronicsQuery.preload()
1658+
1659+
await vi.waitFor(
1660+
() => {
1661+
expect(electronicsQuery.size).toBe(3)
1662+
},
1663+
{ timeout: 2000 },
1664+
)
1665+
1666+
// LQ2: clothing category
1667+
const clothingQuery = createLiveQueryCollection({
1668+
query: (q) =>
1669+
q
1670+
.from({ product: collection })
1671+
.where(({ product }) => eq(product.category, `clothing`))
1672+
.select(({ product }) => ({
1673+
id: product.id,
1674+
name: product.name,
1675+
price: product.price,
1676+
category: product.category,
1677+
})),
1678+
})
1679+
1680+
await clothingQuery.preload()
1681+
1682+
await vi.waitFor(
1683+
() => {
1684+
expect(clothingQuery.size).toBe(2)
1685+
},
1686+
{ timeout: 2000 },
1687+
)
1688+
1689+
// Insert a new electronics product — creates a pending mutation
1690+
const insertResult = collection.insert({
1691+
id: randomUUID(),
1692+
name: `New Gadget`,
1693+
price: 99,
1694+
category: `electronics`,
1695+
})
1696+
1697+
// Immediately clean up the clothing query — triggers unloadSubset → loadSubset
1698+
// which rebuilds the diff trigger, potentially dropping unprocessed diff records
1699+
clothingQuery.cleanup()
1700+
1701+
// isPersisted.promise should resolve — if the bug is present, this hangs forever
1702+
await vi.waitFor(
1703+
async () => {
1704+
await insertResult.isPersisted.promise
1705+
},
1706+
{ timeout: 5000 },
1707+
)
1708+
})
1709+
1710+
it(`should resolve isPersisted when all live queries are cleaned up during a pending mutation`, async () => {
1711+
const db = await createDatabase()
1712+
await createTestProducts(db)
1713+
1714+
const collection = createCollection(
1715+
powerSyncCollectionOptions({
1716+
database: db,
1717+
table: APP_SCHEMA.props.products,
1718+
syncMode: `on-demand`,
1719+
}),
1720+
)
1721+
onTestFinished(() => collection.cleanup())
1722+
await collection.stateWhenReady()
1723+
1724+
// Start with 1 live query (electronics)
1725+
const electronicsQuery = createLiveQueryCollection({
1726+
query: (q) =>
1727+
q
1728+
.from({ product: collection })
1729+
.where(({ product }) => eq(product.category, `electronics`))
1730+
.select(({ product }) => ({
1731+
id: product.id,
1732+
name: product.name,
1733+
price: product.price,
1734+
category: product.category,
1735+
})),
1736+
})
1737+
1738+
await electronicsQuery.preload()
1739+
1740+
await vi.waitFor(
1741+
() => {
1742+
expect(electronicsQuery.size).toBe(3)
1743+
},
1744+
{ timeout: 2000 },
1745+
)
1746+
1747+
// Insert a new electronics product — creates a pending mutation
1748+
const insertResult = collection.insert({
1749+
id: randomUUID(),
1750+
name: `New Gadget`,
1751+
price: 99,
1752+
category: `electronics`,
1753+
})
1754+
1755+
// Immediately clean up the only live query — triggers unloadSubset → loadSubset
1756+
// with 0 predicates (early-return path), which must still call resolveAllPendingFor
1757+
electronicsQuery.cleanup()
1758+
1759+
// isPersisted.promise should resolve — if the bug is present, this hangs forever
1760+
await vi.waitFor(
1761+
async () => {
1762+
await insertResult.isPersisted.promise
1763+
},
1764+
{ timeout: 5000 },
1765+
)
1766+
})
1767+
})
15531768
})

0 commit comments

Comments
 (0)