Sync progress

.changeset/sharp-singers-search.md

@@ -0,0 +1,5 @@
+---
+'@powersync/common': minor
+---
+
+Report progress information about downloaded rows.

packages/common/src/client/AbstractPowerSyncDatabase.ts

@@ -32,6 +32,7 @@ import {
   type PowerSyncConnectionOptions,
   type RequiredAdditionalConnectionOptions
 } from './sync/stream/AbstractStreamingSyncImplementation.js';
+import { FULL_SYNC_PRIORITY } from '../db/crud/SyncProgress.js';
 
 export interface DisconnectAndClearOptions {
   /** When set to false, data in local-only tables is preserved. */
@@ -146,11 +147,6 @@ export const isPowerSyncDatabaseOptionsWithSettings = (test: any): test is Power
   return typeof test == 'object' && isSQLOpenOptions(test.database);
 };
 
-/**
- * The priority used by the core extension to indicate that a full sync was completed.
- */
-const FULL_SYNC_PRIORITY = 2147483647;
-
 export abstract class AbstractPowerSyncDatabase extends BaseObserver<PowerSyncDBListener> {
   /**
    * Transactions should be queued in the DBAdapter, but we also want to prevent

packages/common/src/client/SQLOpenFactory.ts

@@ -7,7 +7,7 @@ export interface SQLOpenOptions {
   dbFilename: string;
   /**
    * Directory where the database file is located.
-   * 
+   *
    * When set, the directory must exist when the database is opened, it will
    * not be created automatically.
    */

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

@@ -30,6 +30,14 @@ export interface SyncLocalDatabaseResult {
   checkpointFailures?: string[];
 }
 
+export type BucketOperationProgress = Record<
+  string,
+  {
+    atLast: number;
+    sinceLast: number;
+  }
+>;
+
 export interface BucketChecksum {
   bucket: string;
   priority?: number;
@@ -65,6 +73,7 @@ export interface BucketStorageAdapter extends BaseObserver<BucketStorageListener
   startSession(): void;
 
   getBucketStates(): Promise<BucketState[]>;
+  getBucketOperationProgress(): Promise<BucketOperationProgress>;
 
   syncLocalDatabase(
     checkpoint: Checkpoint,

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

@@ -5,6 +5,7 @@ import { BaseObserver } from '../../../utils/BaseObserver.js';
 import { MAX_OP_ID } from '../../constants.js';
 import {
   BucketChecksum,
+  BucketOperationProgress,
   BucketState,
   BucketStorageAdapter,
   BucketStorageListener,
@@ -91,6 +92,13 @@ export class SqliteBucketStorage extends BaseObserver<BucketStorageListener> imp
     return result;
   }
 
+  async getBucketOperationProgress(): Promise<BucketOperationProgress> {
+    const rows = await this.db.getAll<{ name: string; count_at_last: number; count_since_last: number }>(
+      'SELECT name, count_at_last, count_since_last FROM ps_buckets'
+    );
+    return Object.fromEntries(rows.map((r) => [r.name, { atLast: r.count_at_last, sinceLast: r.count_since_last }]));
+  }
+
   async saveSyncData(batch: SyncDataBatch) {
     await this.writeTransaction(async (tx) => {
       let count = 0;
@@ -199,7 +207,21 @@ export class SqliteBucketStorage extends BaseObserver<BucketStorageListener> imp
         'sync_local',
         arg
       ]);
-      return result == 1;
+      if (result == 1) {
+        if (priority == null) {
+          const bucketToCount = Object.fromEntries(checkpoint.buckets.map((b) => [b.bucket, b.count]));
+          // The two parameters could be replaced with one, but: https://github.com/powersync-ja/better-sqlite3/pull/6
+          const jsonBucketCount = JSON.stringify(bucketToCount);
+          await tx.execute(
+            "UPDATE ps_buckets SET count_since_last = 0, count_at_last = ?->name WHERE name != '$local' AND ?->name IS NOT NULL",
+            [jsonBucketCount, jsonBucketCount]
+          );
+        }
+
+        return true;
+      } else {
+        return false;
+      }
     });
   }
 

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

@@ -20,6 +20,7 @@ import {
   isStreamingSyncData
 } from './streaming-sync-types.js';
 import { DataStream } from 'src/utils/DataStream.js';
+import { InternalProgressInformation } from 'src/db/crud/SyncProgress.js';
 
 export enum LockType {
   CRUD = 'crud',
@@ -164,21 +165,23 @@ export abstract class AbstractStreamingSyncImplementation
   private pendingCrudUpload?: Promise<void>;
 
   syncStatus: SyncStatus;
+  private syncStatusOptions: SyncStatusOptions;
   triggerCrudUpload: () => void;
 
   constructor(options: AbstractStreamingSyncImplementationOptions) {
     super();
     this.options = { ...DEFAULT_STREAMING_SYNC_OPTIONS, ...options };
 
-    this.syncStatus = new SyncStatus({
+    this.syncStatusOptions = {
       connected: false,
       connecting: false,
       lastSyncedAt: undefined,
       dataFlow: {
         uploading: false,
         downloading: false
       }
-    });
+    };
+    this.syncStatus = new SyncStatus(this.syncStatusOptions);
     this.abortController = null;
 
     this.triggerCrudUpload = throttleLeadingTrailing(() => {
@@ -424,7 +427,8 @@ The next upload iteration will be delayed.`);
         connected: false,
         connecting: false,
         dataFlow: {
-          downloading: false
+          downloading: false,
+          downloadProgress: null
         }
       });
     });
@@ -497,10 +501,7 @@ The next upload iteration will be delayed.`);
     return [req, localDescriptions];
   }
 
-  protected async streamingSyncIteration(
-    signal: AbortSignal,
-    options?: PowerSyncConnectionOptions
-  ): Promise<void> {
+  protected async streamingSyncIteration(signal: AbortSignal, options?: PowerSyncConnectionOptions): Promise<void> {
     await this.obtainLock({
       type: LockType.SYNC,
       signal,
@@ -580,6 +581,7 @@ The next upload iteration will be delayed.`);
             bucketMap = newBuckets;
             await this.options.adapter.removeBuckets([...bucketsToDelete]);
             await this.options.adapter.setTargetCheckpoint(targetCheckpoint);
+            await this.updateSyncStatusForStartingCheckpoint(targetCheckpoint);
           } else if (isStreamingSyncCheckpointComplete(line)) {
             const result = await this.applyCheckpoint(targetCheckpoint!, signal);
             if (result.endIteration) {
@@ -640,6 +642,7 @@ The next upload iteration will be delayed.`);
               write_checkpoint: diff.write_checkpoint
             };
             targetCheckpoint = newCheckpoint;
+            await this.updateSyncStatusForStartingCheckpoint(targetCheckpoint);
 
             bucketMap = new Map();
             newBuckets.forEach((checksum, name) =>
@@ -657,9 +660,23 @@ The next upload iteration will be delayed.`);
             await this.options.adapter.setTargetCheckpoint(targetCheckpoint);
           } else if (isStreamingSyncData(line)) {
             const { data } = line;
+            const previousProgress = this.syncStatusOptions.dataFlow?.downloadProgress;
+            let updatedProgress: InternalProgressInformation | null = null;
+            if (previousProgress) {
+              updatedProgress = { ...previousProgress };
+              const progressForBucket = updatedProgress[data.bucket];
+              if (progressForBucket) {
+                updatedProgress[data.bucket] = {
+                  ...progressForBucket,
+                  sinceLast: progressForBucket.sinceLast + data.data.length
+                };
+              }
+            }
+
             this.updateSyncStatus({
               dataFlow: {
-                downloading: true
+                downloading: true,
+                downloadProgress: updatedProgress
               }
             });
             await this.options.adapter.saveSyncData({ buckets: [SyncDataBucket.fromRow(data)] });
@@ -673,7 +690,7 @@ The next upload iteration will be delayed.`);
                * (uses the same one), this should have some delay.
                */
               await this.delayRetry();
-              return ;
+              return;
             }
             this.triggerCrudUpload();
           } else {
@@ -705,6 +722,30 @@ The next upload iteration will be delayed.`);
     });
   }
 
+  private async updateSyncStatusForStartingCheckpoint(checkpoint: Checkpoint) {
+    const localProgress = await this.options.adapter.getBucketOperationProgress();
+    const progress: InternalProgressInformation = {};
+
+    for (const bucket of checkpoint.buckets) {
+      const savedProgress = localProgress[bucket.bucket];
+      progress[bucket.bucket] = {
+        // The fallback priority doesn't matter here, but 3 is the one newer versions of the sync service
+        // will use by default.
+        priority: bucket.priority ?? 3,
+        atLast: savedProgress?.atLast ?? 0,
+        sinceLast: savedProgress?.sinceLast ?? 0,
+        targetCount: bucket.count ?? 0
+      };
+    }
+
+    this.updateSyncStatus({
+      dataFlow: {
+        downloading: true,
+        downloadProgress: progress
+      }
+    });
+  }
+
   private async applyCheckpoint(checkpoint: Checkpoint, abort: AbortSignal) {
     let result = await this.options.adapter.syncLocalDatabase(checkpoint);
     const pending = this.pendingCrudUpload;
@@ -739,6 +780,7 @@ The next upload iteration will be delayed.`);
         lastSyncedAt: new Date(),
         dataFlow: {
           downloading: false,
+          downloadProgress: null,
           downloadError: undefined
         }
       });
@@ -763,6 +805,7 @@ The next upload iteration will be delayed.`);
     });
 
     if (!this.syncStatus.isEqual(updatedStatus)) {
+      Object.assign(this.syncStatusOptions, options);
       this.syncStatus = updatedStatus;
       // Only trigger this is there was a change
       this.iterateListeners((cb) => cb.statusChanged?.(updatedStatus));

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

@@ -102,7 +102,7 @@ export interface StreamingSyncCheckpointDiff {
     last_op_id: OpId;
     updated_buckets: BucketChecksum[];
     removed_buckets: string[];
-    write_checkpoint: string;
+    write_checkpoint?: string;
   };
 }
 

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

@@ -0,0 +1,107 @@
+import type { SyncStatus } from './SyncStatus.js';
+
+// (bucket, progress) pairs
+/** @internal */
+export type InternalProgressInformation = Record<
+  string,
+  {
+    priority: number; // Priority of the associated buckets
+    atLast: number; // Total ops at last completed sync, or 0
+    sinceLast: number; // Total ops _since_ the last completed sync.
+    targetCount: number; // Total opcount for next checkpoint as indicated by service.
+  }
+>;
+
+/**
+ * @internal The priority used by the core extension to indicate that a full sync was completed.
+ */
+export const FULL_SYNC_PRIORITY = 2147483647;
+
+/**
+ * Information about a progressing download made by the PowerSync SDK.
+ *
+ * To obtain these values, use {@link SyncProgress}, available through
+ * {@link SyncStatus#downloadProgress}.
+ */
+export interface ProgressWithOperations {
+  /**
+   * The total amount of operations to download for the current sync iteration
+   * to complete.
+   */
+  totalOperations: number;
+  /**
+   * The amount of operations that have already been downloaded.
+   */
+  downloadedOperations: number;
+
+  /**
+   * Relative progress, as {@link downloadedOperations} of {@link totalOperations}.
+   *
+   * This will be a number between `0.0` and `1.0` (inclusive).
+   * 
+   * When this number reaches `1.0`, all changes have been received from the sync service.
+   * Actually applying these changes happens before the `downloadProgress` field is cleared from
+   * {@link SyncStatus}, so progress can stay at `1.0` for a short while before completing.
+   */
+  downloadedFraction: number;
+}
+
+/**
+ * Provides realtime progress on how PowerSync is downloading rows.
+ *
+ * The progress until the next complete sync is available through the fields on {@link ProgressWithOperations},
+ * which this class implements.
+ * Additionally, the {@link SyncProgress.untilPriority} method can be used to otbain progress towards
+ * a specific priority (instead of the progress for the entire download).
+ *
+ * The reported progress always reflects the status towards th end of a sync iteration (after
+ * which a consistent snapshot of all buckets is available locally).
+ *
+ * In rare cases (in particular, when a [compacting](https://docs.powersync.com/usage/lifecycle-maintenance/compacting-buckets)
+ * operation takes place between syncs), it's possible for the returned numbers to be slightly
+ * inaccurate. For this reason, {@link SyncProgress} should be seen as an approximation of progress.
+ * The information returned is good enough to build progress bars, but not exact enough to track
+ * individual download counts.
+ *
+ * Also note that data is downloaded in bulk, which means that individual counters are unlikely
+ * to be updated one-by-one.
+ */
+export class SyncProgress implements ProgressWithOperations {
+  totalOperations: number;
+  downloadedOperations: number;
+  downloadedFraction: number;
+
+  constructor(protected internal: InternalProgressInformation) {
+    const untilCompletion = this.untilPriority(FULL_SYNC_PRIORITY);
+
+    this.totalOperations = untilCompletion.totalOperations;
+    this.downloadedOperations = untilCompletion.downloadedOperations;
+    this.downloadedFraction = untilCompletion.downloadedFraction;
+  }
+
+  /**
+   * Returns download progress towards all data up until the specified priority being received.
+   *
+   * The returned {@link ProgressWithOperations} tracks the target amount of operations that need
+   * to be downloaded in total and how many of them have already been received.
+   */
+  untilPriority(priority: number): ProgressWithOperations {
+    let total = 0;
+    let downloaded = 0;
+
+    for (const progress of Object.values(this.internal)) {
+      // Include higher-priority buckets, which are represented by lower numbers.
+      if (progress.priority <= priority) {
+        downloaded += progress.sinceLast;
+        total += progress.targetCount - progress.atLast;
+      }
+    }
+
+    let progress = total == 0 ? 0.0 : downloaded / total;
+    return {
+      totalOperations: total,
+      downloadedOperations: downloaded,
+      downloadedFraction: progress
+    };
+  }
+}