Add test for partial syncs

Author: simolus3

packages/common/src/client/AbstractPowerSyncDatabase.ts

@@ -260,16 +260,30 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver<PowerSyncDB
   }
 
   /**
+   * Wait for the first sync operation to complete.
+   *
+   * @argument request Either an abort signal (after which the promise will complete regardless of
+   * whether a full sync was completed) or an object providing an abort signal and a priority target.
+   * When a priority target is set, the promise may complete when all buckets with the given (or higher)
+   * priorities have been synchronized. This can be earlier than a complete sync.
    * @returns A promise which will resolve once the first full sync has completed.
    */
-  async waitForFirstSync(signal?: AbortSignal): Promise<void> {
-    if (this.currentStatus.hasSynced) {
+  async waitForFirstSync(request?: AbortSignal | { signal?: AbortSignal; priority?: number }): Promise<void> {
+    const signal = request instanceof AbortSignal ? request : request?.signal;
+    const priority = request && 'priority' in request ? request.priority : undefined;
+
+    const statusMatches =
+      priority === undefined
+        ? (status: SyncStatus) => status.hasSynced
+        : (status: SyncStatus) => status.statusForPriority(priority).hasSynced;
+
+    if (statusMatches(this.currentStatus)) {
       return;
     }
     return new Promise((resolve) => {
       const dispose = this.registerListener({
         statusChanged: (status) => {
-          if (status.hasSynced) {
+          if (statusMatches(status)) {
             dispose();
             resolve();
           }
@@ -379,7 +393,8 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver<PowerSyncDB
   // Use the options passed in during connect, or fallback to the options set during database creation or fallback to the default options
   resolvedConnectionOptions(options?: PowerSyncConnectionOptions): RequiredAdditionalConnectionOptions {
     return {
-      retryDelayMs: options?.retryDelayMs ?? this.options.retryDelayMs ?? this.options.retryDelay ?? DEFAULT_RETRY_DELAY_MS,
+      retryDelayMs:
+        options?.retryDelayMs ?? this.options.retryDelayMs ?? this.options.retryDelay ?? DEFAULT_RETRY_DELAY_MS,
       crudUploadThrottleMs:
         options?.crudUploadThrottleMs ?? this.options.crudUploadThrottleMs ?? DEFAULT_CRUD_UPLOAD_THROTTLE_MS
     };
@@ -401,7 +416,7 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver<PowerSyncDB
 
     this.syncStreamImplementation = this.generateSyncStreamImplementation(connector, {
       retryDelayMs,
-      crudUploadThrottleMs,
+      crudUploadThrottleMs
     });
     this.syncStatusListenerDisposer = this.syncStreamImplementation.registerListener({
       statusChanged: (status) => {

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

@@ -66,7 +66,10 @@ export interface BucketStorageAdapter extends BaseObserver<BucketStorageListener
 
   getBucketStates(): Promise<BucketState[]>;
 
-  syncLocalDatabase(checkpoint: Checkpoint, priority?: number): Promise<{ checkpointValid: boolean; ready: boolean; failures?: any[] }>;
+  syncLocalDatabase(
+    checkpoint: Checkpoint,
+    priority?: number
+  ): Promise<{ checkpointValid: boolean; ready: boolean; failures?: any[] }>;
 
   nextCrudItem(): Promise<CrudEntry | undefined>;
   hasCrud(): Promise<boolean>;

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

@@ -186,20 +186,22 @@ export class SqliteBucketStorage extends BaseObserver<BucketStorageListener> imp
         }
       }
 
-      arg = JSON.stringify({priority, buckets: affectedBuckets});
+      arg = JSON.stringify({ priority, buckets: affectedBuckets });
     }
 
     return this.writeTransaction(async (tx) => {
       const { insertId: result } = await tx.execute('INSERT INTO powersync_operations(op, data) VALUES(?, ?)', [
         'sync_local',
-        arg,
+        arg
       ]);
       return result == 1;
     });
   }
 
   async validateChecksums(checkpoint: Checkpoint, priority: number | undefined): Promise<SyncLocalDatabaseResult> {
-    const rs = await this.db.execute('SELECT powersync_validate_checkpoint(?) as result', [JSON.stringify({...checkpoint, priority})]);
+    const rs = await this.db.execute('SELECT powersync_validate_checkpoint(?) as result', [
+      JSON.stringify({ ...checkpoint, priority })
+    ]);
 
     const resultItem = rs.rows?.item(0);
     this.logger.debug('validateChecksums result item', resultItem);

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

@@ -126,6 +126,7 @@ export interface StreamingSyncImplementation extends BaseObserver<StreamingSyncI
   triggerCrudUpload: () => void;
   waitForReady(): Promise<void>;
   waitForStatus(status: SyncStatusOptions): Promise<void>;
+  waitUntilStatusMatches(predicate: (status: SyncStatus) => boolean): Promise<void>;
 }
 
 export const DEFAULT_CRUD_UPLOAD_THROTTLE_MS = 1000;
@@ -184,24 +185,35 @@ export abstract class AbstractStreamingSyncImplementation
   async waitForReady() {}
 
   waitForStatus(status: SyncStatusOptions): Promise<void> {
+    return this.waitUntilStatusMatches((currentStatus) => {
+      /**
+       * Match only the partial status options provided in the
+       * matching status
+       */
+      const matchPartialObject = (compA: object, compB: object) => {
+        return Object.entries(compA).every(([key, value]) => {
+          const comparisonBValue = compB[key];
+          if (typeof value == 'object' && typeof comparisonBValue == 'object') {
+            return matchPartialObject(value, comparisonBValue);
+          }
+          return value == comparisonBValue;
+        });
+      };
+
+      return matchPartialObject(status, currentStatus);
+    });
+  }
+
+  waitUntilStatusMatches(predicate: (status: SyncStatus) => boolean): Promise<void> {
     return new Promise((resolve) => {
+      if (predicate(this.syncStatus)) {
+        resolve();
+        return;
+      }
+
       const l = this.registerListener({
         statusChanged: (updatedStatus) => {
-          /**
-           * Match only the partial status options provided in the
-           * matching status
-           */
-          const matchPartialObject = (compA: object, compB: object) => {
-            return Object.entries(compA).every(([key, value]) => {
-              const comparisonBValue = compB[key];
-              if (typeof value == 'object' && typeof comparisonBValue == 'object') {
-                return matchPartialObject(value, comparisonBValue);
-              }
-              return value == comparisonBValue;
-            });
-          };
-
-          if (matchPartialObject(status, updatedStatus.toJSON())) {
+          if (predicate(updatedStatus)) {
             resolve();
             l?.();
           }
@@ -524,6 +536,7 @@ The next upload iteration will be delayed.`);
             // The stream has closed while waiting
             return { retry: true };
           }
+
           // A connection is active and messages are being received
           if (!this.syncStatus.connected) {
             // There is a connection now
@@ -538,7 +551,7 @@ The next upload iteration will be delayed.`);
             const bucketsToDelete = new Set<string>(bucketMap.keys());
             const newBuckets = new Map<string, BucketDescription>();
             for (const checksum of line.checkpoint.buckets) {
-              newBuckets.set(checksum.bucket, { name: checksum.bucket, priority: checksum.priority })
+              newBuckets.set(checksum.bucket, { name: checksum.bucket, priority: checksum.priority });
               bucketsToDelete.delete(checksum.bucket);
             }
             if (bucketsToDelete.size > 0) {
@@ -567,14 +580,14 @@ The next upload iteration will be delayed.`);
                 lastSyncedAt: new Date(),
                 dataFlow: {
                   downloading: false
-                },
+                }
               });
             }
 
             validatedCheckpoint = targetCheckpoint;
           } else if (isStreamingSyncCheckpointPartiallyComplete(line)) {
-            this.logger.debug('Partial checkpoint complete', targetCheckpoint);
             const priority = line.partial_checkpoint_complete.priority;
+            this.logger.debug('Partial checkpoint complete', priority);
             const result = await this.options.adapter.syncLocalDatabase(targetCheckpoint!, priority);
             if (!result.checkpointValid) {
               // This means checksums failed. Start again with a new checkpoint.
@@ -592,16 +605,12 @@ The next upload iteration will be delayed.`);
               priorityStates.push({
                 priority,
                 lastSyncedAt: new Date(),
-                hasSynced: true,
+                hasSynced: true
               });
 
               this.updateSyncStatus({
                 connected: true,
-                lastSyncedAt: new Date(),
-                statusInPriority: priorityStates,
-                dataFlow: {
-                  downloading: false
-                },
+                statusInPriority: priorityStates
               });
             }
           } else if (isStreamingSyncCheckpointDiff(line)) {
@@ -628,10 +637,13 @@ The next upload iteration will be delayed.`);
             };
             targetCheckpoint = newCheckpoint;
 
-            bucketMap = new Map(newBuckets.entries().map(([name, checksum]) => [name, {
-              name: checksum.bucket,
-              priority: checksum.priority,
-            }]));
+            bucketMap = new Map();
+            newBuckets.forEach((checksum, name) =>
+              bucketMap.set(name, {
+                name: checksum.bucket,
+                priority: checksum.priority
+              })
+            );
 
             const bucketsToDelete = diff.removed_buckets;
             if (bucketsToDelete.length > 0) {
@@ -667,7 +679,7 @@ The next upload iteration will be delayed.`);
               this.updateSyncStatus({
                 connected: true,
                 lastSyncedAt: new Date(),
-                statusInPriority: [],
+                statusInPriority: []
               });
             } else if (validatedCheckpoint === targetCheckpoint) {
               const result = await this.options.adapter.syncLocalDatabase(targetCheckpoint!);
@@ -709,7 +721,7 @@ The next upload iteration will be delayed.`);
         ...this.syncStatus.dataFlowStatus,
         ...options.dataFlow
       },
-      statusInPriority: options.statusInPriority,
+      statusInPriority: options.statusInPriority ?? this.syncStatus.statusInPriority
     });
 
     if (!this.syncStatus.isEqual(updatedStatus)) {

packages/common/src/client/sync/stream/streaming-sync-types.ts

@@ -161,7 +161,9 @@ export function isStreamingSyncCheckpointComplete(line: StreamingSyncLine): line
   return (line as StreamingSyncCheckpointComplete).checkpoint_complete != null;
 }
 
-export function isStreamingSyncCheckpointPartiallyComplete(line: StreamingSyncLine): line is StreamingSyncCheckpointPartiallyComplete {
+export function isStreamingSyncCheckpointPartiallyComplete(
+  line: StreamingSyncLine
+): line is StreamingSyncCheckpointPartiallyComplete {
   return (line as StreamingSyncCheckpointPartiallyComplete).partial_checkpoint_complete != null;
 }
 

packages/common/src/db/DBAdapter.ts

@@ -103,7 +103,7 @@ export interface DBAdapter extends BaseObserverInterface<DBAdapterListener>, DBG
   writeTransaction: <T>(fn: (tx: Transaction) => Promise<T>, options?: DBLockOptions) => Promise<T>;
   /**
    * This method refreshes the schema information across all connections. This is for advanced use cases, and should generally not be needed.
-   */ 
+   */
   refreshSchema: () => Promise<void>;
 }
 

packages/common/src/db/crud/SyncStatus.ts

@@ -5,17 +5,17 @@ export type SyncDataFlowStatus = Partial<{
 
 export interface SyncPriorityStatus {
   priority: number;
-  lastSyncedAt?: Date,
-  hasSynced?: boolean,
-};
+  lastSyncedAt?: Date;
+  hasSynced?: boolean;
+}
 
 export type SyncStatusOptions = {
   connected?: boolean;
   connecting?: boolean;
   dataFlow?: SyncDataFlowStatus;
   lastSyncedAt?: Date;
   hasSynced?: boolean;
-  statusInPriority?: SyncPriorityStatus[],
+  statusInPriority?: SyncPriorityStatus[];
 };
 
 export class SyncStatus {
@@ -77,7 +77,7 @@ export class SyncStatus {
   /**
    * Reports a pair of {@link SyncStatus#hasSynced} and {@link SyncStatus#lastSyncedAt} fields that apply
    * to a specific bucket priority instead of the entire sync operation.
-   * 
+   *
    * When buckets with different priorities are declared, PowerSync may choose to synchronize higher-priority
    * buckets first. When a consistent view over all buckets for all priorities up until the given priority is
    * reached, PowerSync makes data from those buckets available before lower-priority buckets have finished
@@ -86,7 +86,7 @@ export class SyncStatus {
    * be consistent with that checkpoint. For this reason, this method may also return the status for lower priorities.
    * In a state where the PowerSync just finished synchronizing buckets in priority level 3, calling this method
    * with a priority of 1 may return information for priority level 3.
-   * 
+   *
    * @param priority The bucket priority for which the status should be reported.
    */
   statusForPriority(priority: number): SyncPriorityStatus {
@@ -102,7 +102,7 @@ export class SyncStatus {
     return {
       priority,
       lastSyncedAt: this.lastSyncedAt,
-      hasSynced: this.hasSynced,
+      hasSynced: this.hasSynced
     };
   }
 
@@ -122,7 +122,7 @@ export class SyncStatus {
       dataFlow: this.dataFlowStatus,
       lastSyncedAt: this.lastSyncedAt,
       hasSynced: this.hasSynced,
-      statusInPriority: this.options.statusInPriority,
+      statusInPriority: this.statusInPriority
     };
   }
 

packages/common/src/utils/DataStream.ts

@@ -99,10 +99,6 @@ export class DataStream<Data extends any = any> extends BaseObserver<DataStreamL
    * @returns a Data payload or Null if the stream closed.
    */
   async read(): Promise<Data | null> {
-    if (this.dataQueue.length <= this.lowWatermark) {
-      await this.iterateAsyncErrored(async (l) => l.lowWater?.());
-    }
-
     if (this.closed) {
       return null;
     }
@@ -181,14 +177,15 @@ export class DataStream<Data extends any = any> extends BaseObserver<DataStreamL
   }
 
   protected async _processQueue() {
-    if (!this.dataQueue.length || this.isClosed || !this.hasDataReader()) {
+    if (this.isClosed || !this.hasDataReader()) {
       Promise.resolve().then(() => (this.processingPromise = null));
       return;
     }
 
-    const data = this.dataQueue.shift()!;
-
-    await this.iterateAsyncErrored(async (l) => l.data?.(data));
+    if (this.dataQueue.length) {
+      const data = this.dataQueue.shift()!;
+      await this.iterateAsyncErrored(async (l) => l.data?.(data));
+    }
 
     if (this.dataQueue.length <= this.lowWatermark) {
       await this.iterateAsyncErrored(async (l) => l.lowWater?.());

packages/common/tests/db/schema/Schema.test.ts

@@ -35,7 +35,7 @@ describe('Schema', () => {
       }),
       posts: new Table({
         title: column.text,
-        content: column.text,
+        content: column.text
       })
     };
     const schema = new Schema(schemaDefinition);
@@ -62,7 +62,7 @@ describe('Schema', () => {
 
     const invalidSchema = new Schema({
       invalidTable: new Table({
-        'invalid name': column.text,
+        'invalid name': column.text
       })
     });
 
@@ -77,7 +77,7 @@ describe('Schema', () => {
       }),
       posts: new Table({
         title: column.text,
-        content: column.text,
+        content: column.text
       })
     });
 

packages/web/src/db/sync/SSRWebStreamingSyncImplementation.ts

@@ -52,7 +52,14 @@ export class SSRStreamingSyncImplementation extends BaseObserver implements Stre
    * This will never resolve in SSR Mode.
    */
   async waitForStatus(status: SyncStatusOptions) {
-    return new Promise<void>((r) => {});
+    return this.waitUntilStatusMatches(() => false);
+  }
+
+  /**
+   * This will never resolve in SSR Mode.
+   */
+  waitUntilStatusMatches(_predicate: (status: SyncStatus) => boolean): Promise<void> {
+    return new Promise<void>(() => {});
   }
 
   /**