Add test

Author: simolus3

packages/node/src/db/PowerSyncDatabase.ts

@@ -78,7 +78,8 @@ export class PowerSyncDatabase extends AbstractPowerSyncDatabase {
     connector: PowerSyncBackendConnector,
     options: NodeAdditionalConnectionOptions
   ): AbstractStreamingSyncImplementation {
-    const remote = new NodeRemote(connector, this.options.logger, {
+    const logger = this.options.logger;
+    const remote = new NodeRemote(connector, logger, {
       dispatcher: options.dispatcher,
       ...(this.options as NodePowerSyncDatabaseOptions).remoteOptions
     });
@@ -92,7 +93,8 @@ export class PowerSyncDatabase extends AbstractPowerSyncDatabase {
       },
       retryDelayMs: this.options.retryDelayMs,
       crudUploadThrottleMs: this.options.crudUploadThrottleMs,
-      identifier: this.database.name
+      identifier: this.database.name,
+      logger: logger
     });
   }
 }

packages/node/tests/sync.test.ts

@@ -1,9 +1,11 @@
 import { describe, vi, expect, beforeEach } from 'vitest';
+import util from 'node:util';
 
 import { MockSyncService, mockSyncServiceTest, TestConnector, waitForSyncStatus } from './utils';
 import {
   AbstractPowerSyncDatabase,
   BucketChecksum,
+  createLogger,
   OplogEntryJSON,
   PowerSyncConnectionOptions,
   ProgressWithOperations,
@@ -536,6 +538,84 @@ function defineSyncTests(impl: SyncClientImplementation) {
     expect(rows).toStrictEqual([{ name: 'from server' }]);
   });
 
+  mockSyncServiceTest('handles uploads across checkpoints', async ({ syncService }) => {
+    const logger = createLogger('test', { logLevel: Logger.TRACE });
+    const logMessages: string[] = [];
+    (logger as any).invoke = (level, args) => {
+      console.log(...args);
+      logMessages.push(util.format(...args));
+    };
+
+    // Regression test for https://github.com/powersync-ja/powersync-js/pull/665
+    let database = await syncService.createDatabase({ logger });
+    const connector = new TestConnector();
+    let finishUpload: () => void;
+    const finishUploadPromise = new Promise<void>((resolve, reject) => {
+      finishUpload = resolve;
+    });
+    connector.uploadData = async (db) => {
+      const batch = await db.getCrudBatch();
+      if (batch != null) {
+        await finishUploadPromise;
+        await batch.complete();
+      }
+    };
+
+    await database.execute('INSERT INTO lists (id, name) VALUES (uuid(), ?);', ['local']);
+    database.connect(connector, options);
+    await vi.waitFor(() => expect(syncService.connectedListeners).toHaveLength(1));
+
+    syncService.pushLine({ checkpoint: { last_op_id: '1', write_checkpoint: '1', buckets: [bucket('a', 1)] } });
+    syncService.pushLine({
+      data: {
+        bucket: 'a',
+        data: [
+          {
+            checksum: 0,
+            op_id: '1',
+            op: 'PUT',
+            object_id: '1',
+            object_type: 'lists',
+            data: '{"name": "s1"}'
+          }
+        ]
+      }
+    });
+    // 1. Could not apply checkpoint due to local data. We will retry [...] after that upload is completed.
+    syncService.pushLine({ checkpoint_complete: { last_op_id: '1' } });
+    await vi.waitFor(() => {
+      expect(logMessages).toEqual(expect.arrayContaining([expect.stringContaining('due to local data')]));
+    });
+
+    // 2. Send additional checkpoint while we're still busy uploading
+    syncService.pushLine({ checkpoint: { last_op_id: '2', write_checkpoint: '2', buckets: [bucket('a', 2)] } });
+    syncService.pushLine({
+      data: {
+        bucket: 'a',
+        data: [
+          {
+            checksum: 0,
+            op_id: '2',
+            op: 'PUT',
+            object_id: '2',
+            object_type: 'lists',
+            data: '{"name": "s2"}'
+          }
+        ]
+      }
+    });
+    syncService.pushLine({ checkpoint_complete: { last_op_id: '2' } });
+
+    // 3. Crud upload complete
+    finishUpload!();
+
+    // 4. Ensure the database is applying the second checkpoint
+    await vi.waitFor(async () => {
+      const rows = await database.getAll('SELECT * FROM lists WHERE name = ?', ['s2']);
+      expect(rows).toHaveLength(1);
+    });
+  });
+
   mockSyncServiceTest('should update sync state incrementally', async ({ syncService }) => {
     const powersync = await syncService.createDatabase();
     powersync.connect(new TestConnector(), options);

packages/node/tests/utils.ts

@@ -9,7 +9,6 @@ import {
   column,
   NodePowerSyncDatabaseOptions,
   PowerSyncBackendConnector,
-  PowerSyncConnectionOptions,
   PowerSyncCredentials,
   PowerSyncDatabase,
   Schema,
@@ -128,8 +127,9 @@ export const mockSyncServiceTest = tempDirectoryTest.extend<{
       }
     };
 
-    const newConnection = async () => {
+    const newConnection = async (options: Partial<NodePowerSyncDatabaseOptions> = {}) => {
       const db = await createDatabase(tmpdir, {
+        ...options,
         remoteOptions: {
           fetchImplementation: inMemoryFetch
         }
@@ -156,7 +156,7 @@ export const mockSyncServiceTest = tempDirectoryTest.extend<{
 export interface MockSyncService {
   pushLine: (line: StreamingSyncLine) => void;
   connectedListeners: any[];
-  createDatabase: () => Promise<PowerSyncDatabase>;
+  createDatabase: (options?: Partial<NodePowerSyncDatabaseOptions>) => Promise<PowerSyncDatabase>;
 }
 
 export class TestConnector implements PowerSyncBackendConnector {