From ccb68bae25b35a99560fe0c3c67adcd4aac13542 Mon Sep 17 00:00:00 2001 From: bean1352 Date: Wed, 29 Jul 2026 12:52:29 +0200 Subject: [PATCH 1/4] fix: keep uploading when a write races the write checkpoint --- .changeset/lucky-donkeys-wander.md | 5 +++ packages/node/tests/sync.test.ts | 43 +++++++++++++++++++ .../AbstractStreamingSyncImplementation.ts | 4 ++ 3 files changed, 52 insertions(+) create mode 100644 .changeset/lucky-donkeys-wander.md diff --git a/.changeset/lucky-donkeys-wander.md b/.changeset/lucky-donkeys-wander.md new file mode 100644 index 000000000..868a9510a --- /dev/null +++ b/.changeset/lucky-donkeys-wander.md @@ -0,0 +1,5 @@ +--- +'@powersync/shared-internals': patch +--- + +Fixed an issue where a local write made while the client was requesting a write checkpoint could leave the upload queue stuck, blocking uploads and downloads until the next local write. diff --git a/packages/node/tests/sync.test.ts b/packages/node/tests/sync.test.ts index a3b770bab..09e55dd60 100644 --- a/packages/node/tests/sync.test.ts +++ b/packages/node/tests/sync.test.ts @@ -722,6 +722,49 @@ function defineSyncTests(bson: boolean) { }); }); + mockSyncServiceTest('uploads writes made while requesting a write checkpoint', async ({ syncService }) => { + let holdWriteCheckpoint = false; + let onCheckpointRequested!: () => void; + const checkpointRequested = new Promise((resolve) => (onCheckpointRequested = resolve)); + let releaseCheckpoint!: () => void; + const checkpointReleased = new Promise((resolve) => (releaseCheckpoint = resolve)); + + syncService.installRequestInterceptor(async (request) => { + if (!request.url.includes('/write-checkpoint2.json') || !holdWriteCheckpoint) { + return undefined; + } + + onCheckpointRequested(); + await checkpointReleased; + return new Response(JSON.stringify({ data: { write_checkpoint: '1' } }), { status: 200 }); + }); + + const database = await syncService.createDatabase(); + const connector = new TestConnector(); + const pendingCrud = async () => (await database.get<{ c: number }>('SELECT count(*) AS c FROM ps_crud')).c; + + // Complete a transaction outside of the upload loop, which leaves the local write target set with an empty queue. + await database.execute('INSERT INTO lists (id, name) VALUES (uuid(), ?)', ['completed outside the loop']); + const transaction = await database.getNextCrudTransaction(); + await transaction!.complete(); + + // The first upload iteration finds nothing to upload and requests a write checkpoint, which we hold open. + holdWriteCheckpoint = true; + database.connect(connector, { ...options, crudUploadThrottleMs: 100 }); + await checkpointRequested; + + // Ignore CRUD notifications, so that the loop can only recover by noticing the queue itself. + const sync = (database as BasePowerSyncDatabase).syncStreamImplementation!; + sync.triggerCrudUpload = () => {}; + + await database.execute('INSERT INTO lists (id, name) VALUES (uuid(), ?)', ['raced write']); + expect(await pendingCrud()).toBe(1); + releaseCheckpoint(); + + await vi.waitFor(async () => expect(await pendingCrud()).toBe(0)); + expect(connector.uploadDataInvocations).toBeGreaterThanOrEqual(1); + }); + mockSyncServiceTest('should update sync state incrementally', async ({ syncService }) => { const powersync = await syncService.createDatabase(); powersync.connect(new TestConnector(), options); diff --git a/packages/shared-internals/src/client/sync/stream/AbstractStreamingSyncImplementation.ts b/packages/shared-internals/src/client/sync/stream/AbstractStreamingSyncImplementation.ts index 7f593eb73..ab300d498 100644 --- a/packages/shared-internals/src/client/sync/stream/AbstractStreamingSyncImplementation.ts +++ b/packages/shared-internals/src/client/sync/stream/AbstractStreamingSyncImplementation.ts @@ -241,6 +241,10 @@ The next upload iteration will be delayed.` const neededUpdate = await this.options.adapter.updateLocalTarget(() => this.getWriteCheckpoint()); if (neededUpdate) { this.notifyCompletedUploads?.(); + } else if (await this.options.adapter.hasCrud()) { + // `updateLocalTarget` also returns false when a local write raced the write checkpoint request. That + // write still needs to be uploaded, and no checkpoint can be applied until it is. + continue; } else if (checkedCrudItem != null) { // Only log this if there was something to upload this.logger.log({ level: LogLevels.debug, message: 'Upload complete, no write checkpoint needed.' }); From 4223808a4e1712574d841e757dc2b2a7ad770c1f Mon Sep 17 00:00:00 2001 From: bean1352 Date: Tue, 11 Aug 2026 15:58:47 +0200 Subject: [PATCH 2/4] Throttle the raced-write upload retry and fix its test --- .changeset/lucky-donkeys-wander.md | 2 +- packages/node/tests/sync.test.ts | 85 +++++++++++++------ .../AbstractStreamingSyncImplementation.ts | 9 +- 3 files changed, 67 insertions(+), 29 deletions(-) diff --git a/.changeset/lucky-donkeys-wander.md b/.changeset/lucky-donkeys-wander.md index 868a9510a..ac469982c 100644 --- a/.changeset/lucky-donkeys-wander.md +++ b/.changeset/lucky-donkeys-wander.md @@ -2,4 +2,4 @@ '@powersync/shared-internals': patch --- -Fixed an issue where a local write made while the client was requesting a write checkpoint could leave the upload queue stuck, blocking uploads and downloads until the next local write. +Fixed an issue where the upload loop could stop with a local write still queued. When `updateLocalTarget` saw a CRUD entry that the preceding queue read had missed, the loop discarded that information and parked, blocking uploads and downloads until the next local write. It now retries, throttled by `crudUploadThrottleMs`. diff --git a/packages/node/tests/sync.test.ts b/packages/node/tests/sync.test.ts index 115a41615..5a264c521 100644 --- a/packages/node/tests/sync.test.ts +++ b/packages/node/tests/sync.test.ts @@ -20,7 +20,7 @@ import { waitForSyncStatus } from './utils.js'; import { BucketChecksum, OplogEntryJSON } from '@powersync/shared-internals/internal/sync_protocol'; -import { BasePowerSyncDatabase } from '@powersync/shared-internals'; +import { BasePowerSyncDatabase, BucketStorageAdapter } from '@powersync/shared-internals'; const defaultConnectOptions: SyncOptions = { // This might help with test stability/timeouts if a retry is needed. @@ -726,48 +726,81 @@ function defineSyncTests(bson: boolean) { }); mockSyncServiceTest('uploads writes made while requesting a write checkpoint', async ({ syncService }) => { - let holdWriteCheckpoint = false; - let onCheckpointRequested!: () => void; - const checkpointRequested = new Promise((resolve) => (onCheckpointRequested = resolve)); - let releaseCheckpoint!: () => void; - const checkpointReleased = new Promise((resolve) => (releaseCheckpoint = resolve)); - - syncService.installRequestInterceptor(async (request) => { - if (!request.url.includes('/write-checkpoint2.json') || !holdWriteCheckpoint) { - return undefined; - } - - onCheckpointRequested(); - await checkpointReleased; - return new Response(JSON.stringify({ data: { write_checkpoint: '1' } }), { status: 200 }); - }); - const database = await syncService.createDatabase(); const connector = new TestConnector(); const pendingCrud = async () => (await database.get<{ c: number }>('SELECT count(*) AS c FROM ps_crud')).c; + // `updateLocalTarget` checks the CRUD queue in a write transaction, while `nextCrudItem` reads it outside of one. + // Make that read miss the row once, reproducing an upload iteration that finds nothing to upload even though the + // queue is not empty. CRUD notifications are deliberately left intact: the write's own notification is what wakes + // this iteration, so it is already consumed by the time the loop would park again. + const adapter = (database as any).bucketStorageAdapter as BucketStorageAdapter; + const nextCrudItem = adapter.nextCrudItem.bind(adapter); + let missNextRead = false; + adapter.nextCrudItem = async () => { + if (missNextRead) { + missNextRead = false; + return undefined; + } + return nextCrudItem(); + }; + // Complete a transaction outside of the upload loop, which leaves the local write target set with an empty queue. await database.execute('INSERT INTO lists (id, name) VALUES (uuid(), ?)', ['completed outside the loop']); const transaction = await database.getNextCrudTransaction(); await transaction!.complete(); - // The first upload iteration finds nothing to upload and requests a write checkpoint, which we hold open. - holdWriteCheckpoint = true; + // Let the initial upload iteration settle, so the loop is parked waiting for a notification. database.connect(connector, { ...options, crudUploadThrottleMs: 100 }); - await checkpointRequested; - - // Ignore CRUD notifications, so that the loop can only recover by noticing the queue itself. - const sync = (database as BasePowerSyncDatabase).syncStreamImplementation!; - sync.triggerCrudUpload = () => {}; + await vi.waitFor(() => expect(syncService.connectedListeners).toHaveLength(1)); + await vi.waitFor(async () => expect(await pendingCrud()).toBe(0)); + missNextRead = true; await database.execute('INSERT INTO lists (id, name) VALUES (uuid(), ?)', ['raced write']); expect(await pendingCrud()).toBe(1); - releaseCheckpoint(); - await vi.waitFor(async () => expect(await pendingCrud()).toBe(0)); + await vi.waitFor(async () => expect(await pendingCrud()).toBe(0), { timeout: 5000 }); expect(connector.uploadDataInvocations).toBeGreaterThanOrEqual(1); }); + mockSyncServiceTest('throttles the upload retry when the queue read keeps missing', async ({ syncService }) => { + const database = await syncService.createDatabase(); + const connector = new TestConnector(); + + // The retry exits once `nextCrudItem` observes the row. Keep it missing indefinitely to check that the retry is + // rate-limited rather than busy-looping on the write checkpoint endpoint. + const adapter = (database as any).bucketStorageAdapter as BucketStorageAdapter; + const nextCrudItem = adapter.nextCrudItem.bind(adapter); + let alwaysMiss = false; + let missedReads = 0; + adapter.nextCrudItem = async () => { + if (alwaysMiss) { + missedReads++; + return undefined; + } + return nextCrudItem(); + }; + + await database.execute('INSERT INTO lists (id, name) VALUES (uuid(), ?)', ['completed outside the loop']); + const transaction = await database.getNextCrudTransaction(); + await transaction!.complete(); + + const throttleMs = 100; + database.connect(connector, { ...options, crudUploadThrottleMs: throttleMs }); + await vi.waitFor(() => expect(syncService.connectedListeners).toHaveLength(1)); + await vi.waitFor(async () => + expect((await database.get<{ c: number }>('SELECT count(*) AS c FROM ps_crud')).c).toBe(0) + ); + + const observeMs = 1000; + alwaysMiss = true; + await database.execute('INSERT INTO lists (id, name) VALUES (uuid(), ?)', ['raced write']); + await new Promise((resolve) => setTimeout(resolve, observeMs)); + + // Without throttling this loops as fast as the event loop allows (thousands of iterations per second). + expect(missedReads).toBeLessThan((observeMs / throttleMs) * 4); + }); + mockSyncServiceTest('should update sync state incrementally', async ({ syncService }) => { const powersync = await syncService.createDatabase(); powersync.connect(new TestConnector(), options); diff --git a/packages/shared-internals/src/client/sync/stream/AbstractStreamingSyncImplementation.ts b/packages/shared-internals/src/client/sync/stream/AbstractStreamingSyncImplementation.ts index 61dd53102..abde33e26 100644 --- a/packages/shared-internals/src/client/sync/stream/AbstractStreamingSyncImplementation.ts +++ b/packages/shared-internals/src/client/sync/stream/AbstractStreamingSyncImplementation.ts @@ -243,8 +243,13 @@ The next upload iteration will be delayed.` if (neededUpdate) { this.notifyCompletedUploads?.(); } else if (await this.options.adapter.hasCrud()) { - // `updateLocalTarget` also returns false when a local write raced the write checkpoint request. That - // write still needs to be uploaded, and no checkpoint can be applied until it is. + // `updateLocalTarget` compares the CRUD queue inside a write transaction, so it can see a local write + // that `nextCrudItem()` did not. When that happens the write still needs to be uploaded and no + // checkpoint can be applied until it is, so retry instead of parking the loop. + // + // The retry is throttled because its exit condition is `nextCrudItem()` observing the row. If the two + // reads keep disagreeing, an immediate `continue` busy-loops and floods the write checkpoint endpoint. + await this.delayRetry(signal, options.crudUploadThrottleMs); continue; } else if (checkedCrudItem != null) { // Only log this if there was something to upload From eb76f810c037c9d61b1595dac66d5483b3d25c06 Mon Sep 17 00:00:00 2001 From: bean1352 Date: Wed, 12 Aug 2026 09:32:45 +0200 Subject: [PATCH 3/4] Retry a raced write immediately, throttle only repeats --- .changeset/lucky-donkeys-wander.md | 2 +- packages/node/tests/sync.test.ts | 36 +++++++++++++++++++ .../AbstractStreamingSyncImplementation.ts | 15 ++++++-- 3 files changed, 49 insertions(+), 4 deletions(-) diff --git a/.changeset/lucky-donkeys-wander.md b/.changeset/lucky-donkeys-wander.md index ac469982c..2b2a9d25a 100644 --- a/.changeset/lucky-donkeys-wander.md +++ b/.changeset/lucky-donkeys-wander.md @@ -2,4 +2,4 @@ '@powersync/shared-internals': patch --- -Fixed an issue where the upload loop could stop with a local write still queued. When `updateLocalTarget` saw a CRUD entry that the preceding queue read had missed, the loop discarded that information and parked, blocking uploads and downloads until the next local write. It now retries, throttled by `crudUploadThrottleMs`. +Fixed an issue where the upload loop could stop with a local write still queued. When `updateLocalTarget` saw a CRUD entry that the preceding queue read had missed, the loop discarded that information and parked, blocking uploads and downloads until the next local write. It now retries instead, immediately on the first attempt and throttled by `crudUploadThrottleMs` after that. diff --git a/packages/node/tests/sync.test.ts b/packages/node/tests/sync.test.ts index 5a264c521..c823eff0b 100644 --- a/packages/node/tests/sync.test.ts +++ b/packages/node/tests/sync.test.ts @@ -801,6 +801,42 @@ function defineSyncTests(bson: boolean) { expect(missedReads).toBeLessThan((observeMs / throttleMs) * 4); }); + mockSyncServiceTest('retries a raced write without waiting for the upload throttle', async ({ syncService }) => { + const database = await syncService.createDatabase(); + const connector = new TestConnector(); + const pendingCrud = async () => (await database.get<{ c: number }>('SELECT count(*) AS c FROM ps_crud')).c; + + const adapter = (database as any).bucketStorageAdapter as BucketStorageAdapter; + const nextCrudItem = adapter.nextCrudItem.bind(adapter); + let missNextRead = false; + adapter.nextCrudItem = async () => { + if (missNextRead) { + missNextRead = false; + return undefined; + } + return nextCrudItem(); + }; + + await database.execute('INSERT INTO lists (id, name) VALUES (uuid(), ?)', ['completed outside the loop']); + const transaction = await database.getNextCrudTransaction(); + await transaction!.complete(); + + // A throttle much longer than the upload itself, so waiting one interval would show up in the timing below. + const throttleMs = 1500; + database.connect(connector, { ...options, crudUploadThrottleMs: throttleMs }); + await vi.waitFor(() => expect(syncService.connectedListeners).toHaveLength(1)); + // Let the first iteration finish its throttle and park, so the measurement below covers only the retry. + await new Promise((resolve) => setTimeout(resolve, throttleMs + 300)); + + missNextRead = true; + const startedAt = performance.now(); + await database.execute('INSERT INTO lists (id, name) VALUES (uuid(), ?)', ['raced write']); + await vi.waitFor(async () => expect(await pendingCrud()).toBe(0), { timeout: 5000 }); + + // The first retry runs immediately, so the queue drains well inside one throttle interval. + expect(performance.now() - startedAt).toBeLessThan(throttleMs / 2); + }); + mockSyncServiceTest('should update sync state incrementally', async ({ syncService }) => { const powersync = await syncService.createDatabase(); powersync.connect(new TestConnector(), options); diff --git a/packages/shared-internals/src/client/sync/stream/AbstractStreamingSyncImplementation.ts b/packages/shared-internals/src/client/sync/stream/AbstractStreamingSyncImplementation.ts index abde33e26..45e86e13e 100644 --- a/packages/shared-internals/src/client/sync/stream/AbstractStreamingSyncImplementation.ts +++ b/packages/shared-internals/src/client/sync/stream/AbstractStreamingSyncImplementation.ts @@ -212,6 +212,10 @@ export abstract class AbstractStreamingSyncImplementation * Keep track of the first item in the CRUD queue for the last `uploadCrud` iteration. */ let checkedCrudItem: CrudEntry | undefined; + /** + * Consecutive retries caused by a local write that `nextCrudItem` did not see, reset once an upload succeeds. + */ + let unseenWriteRetries = 0; while (!signal.aborted) { try { @@ -236,6 +240,7 @@ The next upload iteration will be delayed.` checkedCrudItem = nextCrudItem; await this.options.uploadCrud(); + unseenWriteRetries = 0; this.updateJsSyncState({ uploadError: undefined }); } else { // Uploading is completed @@ -247,9 +252,13 @@ The next upload iteration will be delayed.` // that `nextCrudItem()` did not. When that happens the write still needs to be uploaded and no // checkpoint can be applied until it is, so retry instead of parking the loop. // - // The retry is throttled because its exit condition is `nextCrudItem()` observing the row. If the two - // reads keep disagreeing, an immediate `continue` busy-loops and floods the write checkpoint endpoint. - await this.delayRetry(signal, options.crudUploadThrottleMs); + // The first retry runs immediately, because the row is normally visible by then and delaying it would + // add latency to an ordinary upload. Later retries wait: the exit condition is `nextCrudItem()` + // observing the row, so if the two reads keep disagreeing an unthrottled loop would request write + // checkpoints as fast as the event loop allows. + if (unseenWriteRetries++ > 0) { + await this.delayRetry(signal, options.crudUploadThrottleMs); + } continue; } else if (checkedCrudItem != null) { // Only log this if there was something to upload From fd15846753a013b0bbeffc256b7544879f1598b4 Mon Sep 17 00:00:00 2001 From: bean1352 Date: Wed, 12 Aug 2026 12:12:54 +0200 Subject: [PATCH 4/4] Return a result type from updateLocalTarget instead of a boolean --- packages/node/tests/sync.test.ts | 167 +++++++++--------- .../sync/bucket/BucketStorageAdapter.ts | 17 +- .../client/sync/bucket/SqliteBucketStorage.ts | 13 +- .../AbstractStreamingSyncImplementation.ts | 15 +- 4 files changed, 119 insertions(+), 93 deletions(-) diff --git a/packages/node/tests/sync.test.ts b/packages/node/tests/sync.test.ts index c823eff0b..91f61294c 100644 --- a/packages/node/tests/sync.test.ts +++ b/packages/node/tests/sync.test.ts @@ -149,6 +149,88 @@ describe('Sync', () => { await database.waitForStatus((s) => !s.connected); await vi.waitFor(() => expect(syncService.connectedListeners).toHaveLength(1)); }); + + mockSyncServiceTest('throttles the upload retry when the queue read keeps missing', async ({ syncService }) => { + const database = await syncService.createDatabase(); + const connector = new TestConnector(); + + // The retry exits once `nextCrudItem` observes the row. Keep it missing indefinitely to check that the retry is + // rate-limited rather than busy-looping on the write checkpoint endpoint. + const adapter = (database as any).bucketStorageAdapter as BucketStorageAdapter; + const nextCrudItem = adapter.nextCrudItem.bind(adapter); + let alwaysMiss = false; + let missedReads = 0; + adapter.nextCrudItem = async () => { + if (alwaysMiss) { + missedReads++; + return undefined; + } + return nextCrudItem(); + }; + + await database.execute('INSERT INTO lists (id, name) VALUES (uuid(), ?)', ['completed outside the loop']); + const transaction = await database.getNextCrudTransaction(); + await transaction!.complete(); + + const throttleMs = 100; + database.connect(connector, { + ...defaultConnectOptions, + connectionMethod: SyncStreamConnectionMethod.HTTP, + crudUploadThrottleMs: throttleMs + }); + await vi.waitFor(() => expect(syncService.connectedListeners).toHaveLength(1)); + await vi.waitFor(async () => + expect((await database.get<{ c: number }>('SELECT count(*) AS c FROM ps_crud')).c).toBe(0) + ); + + const observeMs = 1000; + alwaysMiss = true; + await database.execute('INSERT INTO lists (id, name) VALUES (uuid(), ?)', ['raced write']); + await new Promise((resolve) => setTimeout(resolve, observeMs)); + + // Without throttling this loops as fast as the event loop allows (thousands of iterations per second). + expect(missedReads).toBeLessThan((observeMs / throttleMs) * 4); + }); + + mockSyncServiceTest('retries a raced write without waiting for the upload throttle', async ({ syncService }) => { + const database = await syncService.createDatabase(); + const connector = new TestConnector(); + const pendingCrud = async () => (await database.get<{ c: number }>('SELECT count(*) AS c FROM ps_crud')).c; + + const adapter = (database as any).bucketStorageAdapter as BucketStorageAdapter; + const nextCrudItem = adapter.nextCrudItem.bind(adapter); + let missNextRead = false; + adapter.nextCrudItem = async () => { + if (missNextRead) { + missNextRead = false; + return undefined; + } + return nextCrudItem(); + }; + + await database.execute('INSERT INTO lists (id, name) VALUES (uuid(), ?)', ['completed outside the loop']); + const transaction = await database.getNextCrudTransaction(); + await transaction!.complete(); + + // A throttle much longer than the upload itself, so waiting one interval would show up in the timing below. + const throttleMs = 1500; + database.connect(connector, { + ...defaultConnectOptions, + connectionMethod: SyncStreamConnectionMethod.HTTP, + crudUploadThrottleMs: throttleMs + }); + await vi.waitFor(() => expect(syncService.connectedListeners).toHaveLength(1)); + // Let the first iteration finish its throttle and park, so the measurement below covers only the retry. + await new Promise((resolve) => setTimeout(resolve, throttleMs + 300)); + + missNextRead = true; + const startedAt = performance.now(); + await database.execute('INSERT INTO lists (id, name) VALUES (uuid(), ?)', ['raced write']); + await vi.waitFor(async () => expect(await pendingCrud()).toBe(0), { timeout: 5000 }); + + // The first retry runs immediately, so the queue drains well inside one throttle interval. + expect(performance.now() - startedAt).toBeLessThan(throttleMs / 2); + }); }); function defineSyncTests(bson: boolean) { @@ -725,15 +807,16 @@ function defineSyncTests(bson: boolean) { }); }); - mockSyncServiceTest('uploads writes made while requesting a write checkpoint', async ({ syncService }) => { + mockSyncServiceTest('retries when the queue read misses a write updateLocalTarget sees', async ({ syncService }) => { const database = await syncService.createDatabase(); const connector = new TestConnector(); const pendingCrud = async () => (await database.get<{ c: number }>('SELECT count(*) AS c FROM ps_crud')).c; - // `updateLocalTarget` checks the CRUD queue in a write transaction, while `nextCrudItem` reads it outside of one. - // Make that read miss the row once, reproducing an upload iteration that finds nothing to upload even though the - // queue is not empty. CRUD notifications are deliberately left intact: the write's own notification is what wakes - // this iteration, so it is already consumed by the time the loop would park again. + // Stubbing `nextCrudItem` puts the upload loop into the state seen in a customer's TRACE logs: an iteration that + // uploaded nothing, while `updateLocalTarget` reported new CRUD from its own write transaction. Why the two reads + // disagree in the field is not yet established, so this pins the loop's behaviour in that state rather than + // demonstrating how the state arises. CRUD notifications are deliberately left working, because the ordinary + // interleaving (a write landing during the write checkpoint request) is already recovered by the notification. const adapter = (database as any).bucketStorageAdapter as BucketStorageAdapter; const nextCrudItem = adapter.nextCrudItem.bind(adapter); let missNextRead = false; @@ -763,80 +846,6 @@ function defineSyncTests(bson: boolean) { expect(connector.uploadDataInvocations).toBeGreaterThanOrEqual(1); }); - mockSyncServiceTest('throttles the upload retry when the queue read keeps missing', async ({ syncService }) => { - const database = await syncService.createDatabase(); - const connector = new TestConnector(); - - // The retry exits once `nextCrudItem` observes the row. Keep it missing indefinitely to check that the retry is - // rate-limited rather than busy-looping on the write checkpoint endpoint. - const adapter = (database as any).bucketStorageAdapter as BucketStorageAdapter; - const nextCrudItem = adapter.nextCrudItem.bind(adapter); - let alwaysMiss = false; - let missedReads = 0; - adapter.nextCrudItem = async () => { - if (alwaysMiss) { - missedReads++; - return undefined; - } - return nextCrudItem(); - }; - - await database.execute('INSERT INTO lists (id, name) VALUES (uuid(), ?)', ['completed outside the loop']); - const transaction = await database.getNextCrudTransaction(); - await transaction!.complete(); - - const throttleMs = 100; - database.connect(connector, { ...options, crudUploadThrottleMs: throttleMs }); - await vi.waitFor(() => expect(syncService.connectedListeners).toHaveLength(1)); - await vi.waitFor(async () => - expect((await database.get<{ c: number }>('SELECT count(*) AS c FROM ps_crud')).c).toBe(0) - ); - - const observeMs = 1000; - alwaysMiss = true; - await database.execute('INSERT INTO lists (id, name) VALUES (uuid(), ?)', ['raced write']); - await new Promise((resolve) => setTimeout(resolve, observeMs)); - - // Without throttling this loops as fast as the event loop allows (thousands of iterations per second). - expect(missedReads).toBeLessThan((observeMs / throttleMs) * 4); - }); - - mockSyncServiceTest('retries a raced write without waiting for the upload throttle', async ({ syncService }) => { - const database = await syncService.createDatabase(); - const connector = new TestConnector(); - const pendingCrud = async () => (await database.get<{ c: number }>('SELECT count(*) AS c FROM ps_crud')).c; - - const adapter = (database as any).bucketStorageAdapter as BucketStorageAdapter; - const nextCrudItem = adapter.nextCrudItem.bind(adapter); - let missNextRead = false; - adapter.nextCrudItem = async () => { - if (missNextRead) { - missNextRead = false; - return undefined; - } - return nextCrudItem(); - }; - - await database.execute('INSERT INTO lists (id, name) VALUES (uuid(), ?)', ['completed outside the loop']); - const transaction = await database.getNextCrudTransaction(); - await transaction!.complete(); - - // A throttle much longer than the upload itself, so waiting one interval would show up in the timing below. - const throttleMs = 1500; - database.connect(connector, { ...options, crudUploadThrottleMs: throttleMs }); - await vi.waitFor(() => expect(syncService.connectedListeners).toHaveLength(1)); - // Let the first iteration finish its throttle and park, so the measurement below covers only the retry. - await new Promise((resolve) => setTimeout(resolve, throttleMs + 300)); - - missNextRead = true; - const startedAt = performance.now(); - await database.execute('INSERT INTO lists (id, name) VALUES (uuid(), ?)', ['raced write']); - await vi.waitFor(async () => expect(await pendingCrud()).toBe(0), { timeout: 5000 }); - - // The first retry runs immediately, so the queue drains well inside one throttle interval. - expect(performance.now() - startedAt).toBeLessThan(throttleMs / 2); - }); - mockSyncServiceTest('should update sync state incrementally', async ({ syncService }) => { const powersync = await syncService.createDatabase(); powersync.connect(new TestConnector(), options); diff --git a/packages/shared-internals/src/client/sync/bucket/BucketStorageAdapter.ts b/packages/shared-internals/src/client/sync/bucket/BucketStorageAdapter.ts index 594d377c0..ef0d6be7d 100644 --- a/packages/shared-internals/src/client/sync/bucket/BucketStorageAdapter.ts +++ b/packages/shared-internals/src/client/sync/bucket/BucketStorageAdapter.ts @@ -35,6 +35,21 @@ export interface BucketStorageListener extends BaseListener { crudUpdate: () => void; } +/** + * The outcome of an attempt to record a write checkpoint as the local target. + * + * - `updated_checkpoint`: the target was recorded, so uploads are fully acknowledged. + * - `new_data`: the CRUD queue was not empty when the checkpoint came back, so a local write raced the request and + * still needs uploading. + * - `sequence_changed`: the queue is empty, but its sequence moved while the checkpoint was in flight, so the + * checkpoint is already outdated and a new one is needed. + * - `no_crud_sequence`: there was nothing to record, either because no CRUD has ever been written or because a target + * checkpoint is already pending. + * + * @internal + */ +export type UpdateLocalTargetResult = 'updated_checkpoint' | 'new_data' | 'sequence_changed' | 'no_crud_sequence'; + export interface BucketStorageAdapter extends BaseObserverInterface, Disposable { hasMigratedSubkeys(): Promise; migrateToFixedSubkeys(): Promise; @@ -43,7 +58,7 @@ export interface BucketStorageAdapter extends BaseObserverInterface; getCrudBatch(limit?: number): Promise; - updateLocalTarget(cb: () => Promise): Promise; + updateLocalTarget(cb: () => Promise): Promise; handleCrudCheckpoint(lastClientId: number, writeCheckpoint?: string): Promise; /** diff --git a/packages/shared-internals/src/client/sync/bucket/SqliteBucketStorage.ts b/packages/shared-internals/src/client/sync/bucket/SqliteBucketStorage.ts index dcaf241e8..f088d30a3 100644 --- a/packages/shared-internals/src/client/sync/bucket/SqliteBucketStorage.ts +++ b/packages/shared-internals/src/client/sync/bucket/SqliteBucketStorage.ts @@ -15,7 +15,8 @@ import { PowerSyncControlCommand, PSInternalTable, rawPowerSyncControl, - targetCheckpointRequestId + targetCheckpointRequestId, + UpdateLocalTargetResult } from './BucketStorageAdapter.js'; import { CrudEntryImpl, CrudEntryJSON } from './CrudEntry.js'; import { MAX_OP_ID } from '../../../constants.js'; @@ -57,7 +58,7 @@ export class SqliteBucketStorage extends BaseObserver imp return this._clientId!; } - async updateLocalTarget(cb: () => Promise): Promise { + async updateLocalTarget(cb: () => Promise): Promise { const sequenceBefore = await this.db.readTransaction(async (tx): Promise => { const currentCheckpoint = await targetCheckpointRequestId(tx); if (currentCheckpoint != MAX_OP_ID) return; @@ -68,7 +69,7 @@ export class SqliteBucketStorage extends BaseObserver imp if (sequenceBefore == null) { // Nothing to update - return false; + return 'no_crud_sequence'; } const opId = await cb(); @@ -81,7 +82,7 @@ export class SqliteBucketStorage extends BaseObserver imp level: LogLevels.debug, message: `New data uploaded since write checkpoint ${opId} - need new write checkpoint` }); - return false; + return 'new_data'; } const { seq: seqAfter } = await tx.get<{ seq: number }>( @@ -95,7 +96,7 @@ export class SqliteBucketStorage extends BaseObserver imp }); // New crud data may have been uploaded since we got the checkpoint. Abort. - return false; + return 'sequence_changed'; } this.logger.log({ @@ -103,7 +104,7 @@ export class SqliteBucketStorage extends BaseObserver imp message: `Updating target write checkpoint to ${opId}` }); await targetCheckpointRequestId(tx, opId); - return true; + return 'updated_checkpoint'; }); } diff --git a/packages/shared-internals/src/client/sync/stream/AbstractStreamingSyncImplementation.ts b/packages/shared-internals/src/client/sync/stream/AbstractStreamingSyncImplementation.ts index 45e86e13e..b549c0c46 100644 --- a/packages/shared-internals/src/client/sync/stream/AbstractStreamingSyncImplementation.ts +++ b/packages/shared-internals/src/client/sync/stream/AbstractStreamingSyncImplementation.ts @@ -244,13 +244,13 @@ The next upload iteration will be delayed.` this.updateJsSyncState({ uploadError: undefined }); } else { // Uploading is completed - const neededUpdate = await this.options.adapter.updateLocalTarget(() => this.getWriteCheckpoint()); - if (neededUpdate) { + const localTarget = await this.options.adapter.updateLocalTarget(() => this.getWriteCheckpoint()); + if (localTarget == 'updated_checkpoint') { this.notifyCompletedUploads?.(); - } else if (await this.options.adapter.hasCrud()) { + } else if (localTarget == 'new_data') { // `updateLocalTarget` compares the CRUD queue inside a write transaction, so it can see a local write - // that `nextCrudItem()` did not. When that happens the write still needs to be uploaded and no - // checkpoint can be applied until it is, so retry instead of parking the loop. + // that `nextCrudItem()` did not. That write still needs to be uploaded and no checkpoint can be + // applied until it is, so retry instead of parking the loop. // // The first retry runs immediately, because the row is normally visible by then and delaying it would // add latency to an ordinary upload. Later retries wait: the exit condition is `nextCrudItem()` @@ -260,8 +260,9 @@ The next upload iteration will be delayed.` await this.delayRetry(signal, options.crudUploadThrottleMs); } continue; - } else if (checkedCrudItem != null) { - // Only log this if there was something to upload + } else if (localTarget == 'no_crud_sequence' && checkedCrudItem != null) { + // Only log this if there was something to upload. `sequence_changed` is excluded because + // `updateLocalTarget` has already reported that a new write checkpoint is needed. this.logger.log({ level: LogLevels.debug, message: 'Upload complete, no write checkpoint needed.' }); } break;