Test partial priority with data

Author: simolus3

packages/common/src/client/sync/bucket/SqliteBucketStorage.ts

@@ -205,8 +205,7 @@ export class SqliteBucketStorage extends BaseObserver<BucketStorageListener> imp
   async validateChecksums(checkpoint: Checkpoint, priority: number | undefined): Promise<SyncLocalDatabaseResult> {
     if (priority !== undefined) {
       // Only validate the buckets within the priority we care about
-      const newBuckets = [...checkpoint.buckets];
-      newBuckets.filter((cs) => cs.priority <= priority);
+      const newBuckets = checkpoint.buckets.filter((cs) => cs.priority <= priority);
       checkpoint = {...checkpoint, buckets: newBuckets};
     }
 
@@ -215,7 +214,7 @@ export class SqliteBucketStorage extends BaseObserver<BucketStorageListener> imp
     ]);
 
     const resultItem = rs.rows?.item(0);
-    this.logger.debug('validateChecksums result item', resultItem);
+    this.logger.debug('validateChecksums priority, checkpoint, result item', priority, checkpoint, resultItem);
     if (!resultItem) {
       return {
         checkpointValid: false,

packages/web/tests/stream.test.ts

@@ -165,34 +165,100 @@ describe('Streaming', () => {
 
       const buckets: BucketChecksum[] = [];
       for (let prio = 0; prio <= 3; prio++) {
-        buckets.push({bucket: `prio${prio}`, priority: prio, checksum: 0});
+        buckets.push({bucket: `prio${prio}`, priority: prio, checksum: 10 + prio});
       }
       remote.enqueueLine({
         checkpoint: {
-          last_op_id: '0',
+          last_op_id: '4',
           buckets,
         },
       });
 
+      let operationId = 1;
+      const addRow = (prio: number) => {
+        remote.enqueueLine({
+          data: {
+            bucket: `prio${prio}`,
+            data: [{
+              checksum: prio + 10,
+              data: JSON.stringify({'name': 'row'}),
+              op: 'PUT',
+              op_id: (operationId++).toString(),
+              object_id: `prio${prio}`,
+              object_type: 'users'
+            }]
+          },
+        });
+      }
+
+      const syncCompleted = vi.fn();
+      powersync.waitForFirstSync().then(syncCompleted);
+
       // Emit partial sync complete for each priority but the last.
       for (var prio = 0; prio < 3; prio++) {
+        const partialSyncCompleted = vi.fn();
+        powersync.waitForFirstSync({priority: prio}).then(partialSyncCompleted);
         expect(powersync.currentStatus.statusForPriority(prio).hasSynced).toBe(false);
+        expect(partialSyncCompleted).not.toHaveBeenCalled();
+        expect(syncCompleted).not.toHaveBeenCalled();
 
+        addRow(prio);
         remote.enqueueLine({
           partial_checkpoint_complete: {
-            last_op_id: '0',
+            last_op_id: operationId.toString(),
             priority: prio,
           }
         });
 
         await powersync.syncStreamImplementation!.waitUntilStatusMatches((status) => {
           return status.statusForPriority(prio).hasSynced === true;
         });
+        await new Promise(r => setTimeout(r));
+        expect(partialSyncCompleted).toHaveBeenCalledOnce();
+
+        expect(await powersync.getAll('select * from users')).toHaveLength(prio + 1);
       }
 
       // Then, complete the sync.
-      remote.enqueueLine({checkpoint_complete: {last_op_id: '0'}});
-      await powersync.waitForFirstSync();
+      addRow(3);
+      remote.enqueueLine({checkpoint_complete: {last_op_id: operationId.toString()}});
+      await vi.waitFor(() => expect(syncCompleted).toHaveBeenCalledOnce(), 500);
+      expect(await powersync.getAll('select * from users')).toHaveLength(4);
+    });
+
+    itWithGenerators('Should remember sync state', async (createConnectedDatabase) => {
+      const { powersync, remote, openAnother } = await createConnectedDatabase();
+      expect(powersync.currentStatus.dataFlowStatus.downloading).toBe(false);
+
+      const buckets: BucketChecksum[] = [];
+      for (let prio = 0; prio <= 3; prio++) {
+        buckets.push({bucket: `prio${prio}`, priority: prio, checksum: 0});
+      }
+      remote.enqueueLine({
+        checkpoint: {
+          last_op_id: '0',
+          buckets,
+        },
+      });
+      remote.enqueueLine({
+        partial_checkpoint_complete: {
+          last_op_id: '0',
+          priority: 0,
+        }
+      });
+
+      await powersync.waitForFirstSync({priority: 0});
+
+      // Open another database instance.
+      const another = openAnother();
+      onTestFinished(async () => {
+        await another.close();
+      });
+      await another.init();
+
+      expect(another.currentStatus.statusInPriority).toHaveLength(1);
+      expect(another.currentStatus.statusForPriority(0).hasSynced).toBeTruthy();
+      await another.waitForFirstSync({priority: 0});
     });
   });
 });

packages/web/tests/utils/generateConnectedDatabase.ts

@@ -51,7 +51,9 @@ export async function generateConnectedDatabase(
     },
     remote
   );
-  const powersync = factory.getInstance();
+
+  const openAnother = factory.getInstance.bind(factory);
+  const powersync = openAnother();
 
   const waitForStream = () =>
     new Promise<void>((resolve) => {
@@ -75,6 +77,7 @@ export async function generateConnectedDatabase(
     await connectedPromise;
   };
 
+
   await connect();
 
   onTestFinished(async () => {
@@ -89,6 +92,7 @@ export async function generateConnectedDatabase(
     powersync,
     remote,
     uploadSpy,
-    waitForStream
+    waitForStream,
+    openAnother,
   };
 }