Reformat and changeset

Author: simolus3

.changeset/violet-falcons-build.md

@@ -0,0 +1,5 @@
+---
+'@powersync/common': patch
+---
+
+Fix a race condition causing sync changes during uploads not to be applied.

packages/common/src/client/sync/stream/AbstractStreamingSyncImplementation.ts

@@ -717,21 +717,24 @@ The next upload iteration will be delayed.`);
 
   private async applyCheckpoint(checkpoint: Checkpoint, abort: AbortSignal) {
     let result = await this.options.adapter.syncLocalDatabase(checkpoint);
+    const pending = this.pendingCrudUpload;
+
     if (!result.checkpointValid) {
       this.logger.debug('Checksum mismatch in checkpoint, will reconnect');
       // This means checksums failed. Start again with a new checkpoint.
       // TODO: better back-off
       await new Promise((resolve) => setTimeout(resolve, 50));
       return { applied: false, endIteration: true };
-    } else if (!result.ready) {
+    } else if (!result.ready && pending != null) {
       // We have pending entries in the local upload queue or are waiting to confirm a write
-      // checkpoint. See if that is happening right now.
-      const pending = this.pendingCrudUpload;
-      if (pending != null) {
-        await Promise.race([pending, onAbortPromise(abort)]);
-      }
-
-      if (abort.aborted || pending == null) {
+      // checkpoint, which prevented this checkpoint from applying. Wait for that to complete and
+      // try again.
+      this.logger.debug(
+        'Could not apply checkpoint due to local data. Waiting for in-progress upload before retrying.'
+      );
+      await Promise.race([pending, onAbortPromise(abort)]);
+
+      if (abort.aborted) {
         return { applied: false, endIteration: true };
       }
 
@@ -752,7 +755,7 @@ The next upload iteration will be delayed.`);
 
       return { applied: true, endIteration: false };
     } else {
-      this.logger.debug('Could not apply checkpoint even after waiting for uploads. Waiting for next sync complete line.');
+      this.logger.debug('Could not apply checkpoint. Waiting for next sync complete line.');
       return { applied: false, endIteration: false };
     }
   }

packages/common/src/utils/async.ts

@@ -56,5 +56,5 @@ export function onAbortPromise(signal: AbortSignal): Promise<void> {
     } else {
       signal.onabort = () => resolve();
     }
-  })
+  });
 }

packages/web/tests/stream.test.ts

@@ -54,14 +54,14 @@ describe('Streaming', { sequential: true }, () => {
     let resolveUploadPromise: () => void;
     let resolveUploadStartedPromise: () => void;
     const completeUploadPromise = new Promise<void>((resolve) => {
-      resolveUploadPromise = resolve
+      resolveUploadPromise = resolve;
     });
     const uploadStartedPromise = new Promise<void>((resolve) => {
-      resolveUploadStartedPromise = resolve
+      resolveUploadStartedPromise = resolve;
     });
 
     async function expectUserRows(amount: number) {
-      const row = await powersync.get<{r: number}>('SELECT COUNT(*) AS r FROM users');
+      const row = await powersync.get<{ r: number }>('SELECT COUNT(*) AS r FROM users');
       expect(row.r).toBe(amount);
     }
 
@@ -90,7 +90,7 @@ describe('Streaming', { sequential: true }, () => {
     remote.generateCheckpoint.mockImplementation(() => {
       return {
         data: {
-          write_checkpoint: '1',
+          write_checkpoint: '1'
         }
       };
     });
@@ -114,7 +114,7 @@ describe('Streaming', { sequential: true }, () => {
             object_id: '2',
             object_type: 'users',
             data: '{"id": "test1", "name": "additional entry"}'
-          },
+          }
         ]
       }
     });

packages/web/tests/utils/MockStreamOpenFactory.ts

@@ -47,7 +47,7 @@ export class MockRemote extends AbstractRemote {
     this.generateCheckpoint = vi.fn(() => {
       return {
         data: {
-          write_checkpoint: '1000',
+          write_checkpoint: '1000'
         }
       };
     });