Track progress for sync lines

Author: simolus3

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

@@ -30,6 +30,11 @@ export interface SyncLocalDatabaseResult {
   checkpointFailures?: string[];
 }
 
+export type BucketOperationProgress = Record<string, {
+  atLast: number;
+  sinceLast: number;
+}>;
+
 export interface BucketChecksum {
   bucket: string;
   priority?: number;
@@ -65,6 +70,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,11 @@ 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 +205,16 @@ 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]));
+          await tx.execute('UPDATE ps_buckets SET count_since_last = 0, count_at_last = ?1->name WHERE ?1->name IS NOT NULL', [JSON.stringify(bucketToCount)]);
+        }
+
+        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',
@@ -163,21 +164,23 @@ export abstract class AbstractStreamingSyncImplementation
   protected streamingSyncPromise?: 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(() => {
@@ -411,7 +414,8 @@ The next upload iteration will be delayed.`);
         connected: false,
         connecting: false,
         dataFlow: {
-          downloading: false
+          downloading: false,
+          downloadProgress: null,
         }
       });
     });
@@ -569,6 +573,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)) {
             this.logger.debug('Checkpoint complete', targetCheckpoint);
             const result = await this.options.adapter.syncLocalDatabase(targetCheckpoint!);
@@ -588,7 +593,8 @@ The next upload iteration will be delayed.`);
                 connected: true,
                 lastSyncedAt: new Date(),
                 dataFlow: {
-                  downloading: false
+                  downloading: false,
+                  downloadProgress: null,
                 }
               });
             }
@@ -645,6 +651,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) =>
@@ -662,9 +669,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)] });
@@ -707,7 +728,8 @@ The next upload iteration will be delayed.`);
                   lastSyncedAt: new Date(),
                   priorityStatusEntries: [],
                   dataFlow: {
-                    downloading: false
+                    downloading: false,
+                    downloadProgress: null,
                   }
                 });
               }
@@ -721,6 +743,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,
+      }
+    });
+  }
+
   protected updateSyncStatus(options: SyncStatusOptions) {
     const updatedStatus = new SyncStatus({
       connected: options.connected ?? this.syncStatus.connected,
@@ -734,6 +780,7 @@ The next upload iteration will be delayed.`);
     });
 
     if (!this.syncStatus.isEqual(updatedStatus)) {
+      this.syncStatusOptions = options;
       this.syncStatus = updatedStatus;
       // Only trigger this is there was a change
       this.iterateListeners((cb) => cb.statusChanged?.(updatedStatus));

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

@@ -1,6 +1,7 @@
 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
@@ -9,7 +10,7 @@ export type InternalProgressInformation = Record<string, {
   }>;
 
 /**
- * The priority used by the core extension to indicate that a full sync was completed.
+ * @internal The priority used by the core extension to indicate that a full sync was completed.
  */
 export const FULL_SYNC_PRIORITY = 2147483647;
 

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

@@ -8,7 +8,7 @@ export type SyncDataFlowStatus = Partial<{
    * 
    * Please use the {@link SyncStatus#downloadProgress} property to track sync progress.
    */
-  downloadProgress: InternalProgressInformation,
+  downloadProgress: InternalProgressInformation | null,
 }>;
 
 export interface SyncPriorityStatus {