[Fix] Keep uploading when a write races the write checkpoint

.changeset/lucky-donkeys-wander.md

@@ -0,0 +1,5 @@
+---
+'@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 instead, immediately on the first attempt and throttled by `crudUploadThrottleMs` after that.

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.
@@ -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,6 +807,45 @@ function defineSyncTests(bson: boolean) {
     });
   });
 
+  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;
+
+    // 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;
+    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();
+
+    // Let the initial upload iteration settle, so the loop is parked waiting for a notification.
+    database.connect(connector, { ...options, crudUploadThrottleMs: 100 });
+    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);
+
+    await vi.waitFor(async () => expect(await pendingCrud()).toBe(0), { timeout: 5000 });
+    expect(connector.uploadDataInvocations).toBeGreaterThanOrEqual(1);
+  });
+
   mockSyncServiceTest('should update sync state incrementally', async ({ syncService }) => {
     const powersync = await syncService.createDatabase();
     powersync.connect(new TestConnector(), options);

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<BucketStorageListener>, Disposable {
   hasMigratedSubkeys(): Promise<boolean>;
   migrateToFixedSubkeys(): Promise<void>;
@@ -43,7 +58,7 @@ export interface BucketStorageAdapter extends BaseObserverInterface<BucketStorag
   hasCrud(): Promise<boolean>;
   getCrudBatch(limit?: number): Promise<CrudBatch | null>;
 
-  updateLocalTarget(cb: () => Promise<string>): Promise<boolean>;
+  updateLocalTarget(cb: () => Promise<string>): Promise<UpdateLocalTargetResult>;
   handleCrudCheckpoint(lastClientId: number, writeCheckpoint?: string): Promise<void>;
 
   /**

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<BucketStorageListener> imp
     return this._clientId!;
   }
 
-  async updateLocalTarget(cb: () => Promise<string>): Promise<boolean> {
+  async updateLocalTarget(cb: () => Promise<string>): Promise<UpdateLocalTargetResult> {
     const sequenceBefore = await this.db.readTransaction(async (tx): Promise<number | undefined> => {
       const currentCheckpoint = await targetCheckpointRequestId(tx);
       if (currentCheckpoint != MAX_OP_ID) return;
@@ -68,7 +69,7 @@ export class SqliteBucketStorage extends BaseObserver<BucketStorageListener> 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<BucketStorageListener> 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,15 +96,15 @@ export class SqliteBucketStorage extends BaseObserver<BucketStorageListener> imp
         });
 
         // New crud data may have been uploaded since we got the checkpoint. Abort.
-        return false;
+        return 'sequence_changed';
       }
 
       this.logger.log({
         level: LogLevels.debug,
         message: `Updating target write checkpoint to ${opId}`
       });
       await targetCheckpointRequestId(tx, opId);
-      return true;
+      return 'updated_checkpoint';
     });
   }
 

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,14 +240,29 @@ The next upload iteration will be delayed.`
 
               checkedCrudItem = nextCrudItem;
               await this.options.uploadCrud();
+              unseenWriteRetries = 0;
               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 (checkedCrudItem != null) {
-                // Only log this if there was something to upload
+              } 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. 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()`
+                // 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 (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;