Redo checkpoint change tracking.

Author: rkistner

modules/module-mongodb-storage/src/storage/implementation/MongoSyncBucketStorage.ts

@@ -9,6 +9,8 @@ import {
 } from '@powersync/lib-services-framework';
 import {
   BroadcastIterable,
+  CheckpointChanges,
+  GetCheckpointChangesOptions,
   getLookupBucketDefinitionName,
   ReplicationCheckpoint,
   storage,
@@ -758,37 +760,13 @@ export class MongoSyncBucketStorage
   /**
    * User-specific watch on the latest checkpoint and/or write checkpoint.
    */
-  async *watchWriteCheckpoint(options: WatchWriteCheckpointOptions): AsyncIterable<storage.WriteCheckpoint> {
-    const { user_id, signal, filter } = options;
+  async *watchWriteCheckpoint(options: WatchWriteCheckpointOptions): AsyncIterable<storage.StorageCheckpointUpdate> {
+    const { user_id, signal } = options;
     let lastCheckpoint: utils.OpId | null = null;
     let lastWriteCheckpoint: bigint | null = null;
 
     const iter = wrapWithAbort(this.sharedIter, signal);
-    let shouldUpdate = false;
     for await (const event of iter) {
-      if (filter) {
-        if (event.invalidate) {
-          shouldUpdate =
-            filter({
-              invalidate: true
-            }) || shouldUpdate;
-        }
-
-        if (event.dataBucket) {
-          shouldUpdate =
-            filter({
-              changedDataBucket: event.dataBucket
-            }) || shouldUpdate;
-        }
-
-        if (event.parameterBucketDefinition) {
-          shouldUpdate =
-            filter({
-              changedParameterBucketDefinition: event.parameterBucketDefinition
-            }) || shouldUpdate;
-        }
-      }
-
       if (event.checkpoint) {
         const { checkpoint, lsn } = event.checkpoint;
 
@@ -814,16 +792,27 @@ export class MongoSyncBucketStorage
           continue;
         }
 
+        const updates: CheckpointChanges =
+          lastCheckpoint == null
+            ? {
+                invalidateDataBuckets: true,
+                invalidateParameterBuckets: true,
+                updatedDataBuckets: [],
+                updatedParameterBucketDefinitions: []
+              }
+            : await this.getCheckpointChanges({
+                lastCheckpoint: lastCheckpoint,
+                nextCheckpoint: checkpoint
+              });
+
         lastWriteCheckpoint = currentWriteCheckpoint;
         lastCheckpoint = checkpoint;
 
-        if (shouldUpdate || filter == null) {
-          yield {
-            base: event.checkpoint!,
-            writeCheckpoint: currentWriteCheckpoint
-          };
-          shouldUpdate = false;
-        }
+        yield {
+          base: event.checkpoint!,
+          writeCheckpoint: currentWriteCheckpoint,
+          update: updates
+        };
       }
     }
   }
@@ -936,6 +925,54 @@ export class MongoSyncBucketStorage
     ];
     return pipeline;
   }
+
+  async getCheckpointChanges(options: GetCheckpointChangesOptions): Promise<CheckpointChanges> {
+    const dataBuckets = await this.db.bucket_data
+      .find(
+        {
+          '_id.g': this.group_id,
+          '_id.o': { $gt: BigInt(options.lastCheckpoint), $lte: BigInt(options.nextCheckpoint) }
+        },
+        {
+          projection: {
+            '_id.b': 1
+          },
+          limit: 1001,
+          batchSize: 1001,
+          singleBatch: true
+        }
+      )
+      .toArray();
+    const invalidateDataBuckets = dataBuckets.length > 1000;
+
+    const parameterUpdates = await this.db.bucket_parameters
+      .find(
+        {
+          _id: { $gt: BigInt(options.lastCheckpoint), $lt: BigInt(options.nextCheckpoint) },
+          'key.g': this.group_id
+        },
+        {
+          projection: {
+            lookup: 1
+          },
+          limit: 1001,
+          batchSize: 1001,
+          singleBatch: true
+        }
+      )
+      .toArray();
+    const invalidateParameterUpdates = parameterUpdates.length > 1000;
+
+    return {
+      invalidateDataBuckets,
+      updatedDataBuckets: invalidateDataBuckets ? [] : dataBuckets.map((b) => b._id.b),
+
+      invalidateParameterBuckets: invalidateParameterUpdates,
+      updatedParameterBucketDefinitions: invalidateParameterUpdates
+        ? []
+        : [...new Set<string>(parameterUpdates.map((p) => getLookupBucketDefinitionName(p.lookup)))]
+    };
+  }
 }
 
 interface BucketCheckpointEvent {

modules/module-postgres-storage/src/storage/PostgresSyncRulesStorage.ts

@@ -1,6 +1,14 @@
 import * as lib_postgres from '@powersync/lib-service-postgres';
 import { DisposableObserver, ReplicationAssertionError } from '@powersync/lib-services-framework';
-import { BroadcastIterable, LastValueSink, storage, utils, WatchWriteCheckpointOptions } from '@powersync/service-core';
+import {
+  BroadcastIterable,
+  CheckpointChanges,
+  GetCheckpointChangesOptions,
+  LastValueSink,
+  storage,
+  utils,
+  WatchWriteCheckpointOptions
+} from '@powersync/service-core';
 import { JSONBig } from '@powersync/service-jsonbig';
 import * as sync_rules from '@powersync/service-sync-rules';
 import * as uuid from 'uuid';
@@ -690,11 +698,11 @@ export class PostgresSyncRulesStorage
     return this.makeActiveCheckpoint(activeCheckpoint);
   }
 
-  async *watchWriteCheckpoint(options: WatchWriteCheckpointOptions): AsyncIterable<storage.WriteCheckpoint> {
+  async *watchWriteCheckpoint(options: WatchWriteCheckpointOptions): AsyncIterable<storage.StorageCheckpointUpdate> {
     let lastCheckpoint: utils.OpId | null = null;
     let lastWriteCheckpoint: bigint | null = null;
 
-    const { signal, user_id, filter } = options;
+    const { signal, user_id } = options;
 
     const iter = wrapWithAbort(this.sharedIterator, signal);
     for await (const cp of iter) {
@@ -724,12 +732,16 @@ export class PostgresSyncRulesStorage
       lastWriteCheckpoint = currentWriteCheckpoint;
       lastCheckpoint = checkpoint;
 
-      // We do not track individual bucket updates yet, always send an invalidation event.
-      filter?.({
-        invalidate: true
-      });
-
-      yield { base: cp, writeCheckpoint: currentWriteCheckpoint };
+      yield {
+        base: cp,
+        writeCheckpoint: currentWriteCheckpoint,
+        update: {
+          invalidateDataBuckets: true,
+          invalidateParameterBuckets: true,
+          updatedDataBuckets: [],
+          updatedParameterBucketDefinitions: []
+        }
+      };
     }
   }
 
@@ -791,6 +803,16 @@ export class PostgresSyncRulesStorage
     }
   }
 
+  async getCheckpointChanges(options: GetCheckpointChangesOptions): Promise<CheckpointChanges> {
+    // We do not track individual changes yet
+    return {
+      invalidateDataBuckets: true,
+      invalidateParameterBuckets: true,
+      updatedDataBuckets: [],
+      updatedParameterBucketDefinitions: []
+    };
+  }
+
   private makeActiveCheckpoint(row: models.ActiveCheckpointDecoded | null) {
     return {
       checkpoint: utils.timestampToOpId(row?.last_checkpoint ?? 0n),

packages/service-core/src/storage/SyncRulesBucketStorage.ts

@@ -73,12 +73,14 @@ export interface SyncRulesBucketStorage
    */
   getParameterSets(checkpoint: util.OpId, lookups: SqliteJsonValue[][]): Promise<SqliteJsonRow[]>;
 
+  getCheckpointChanges(options: GetCheckpointChangesOptions): Promise<CheckpointChanges>;
+
   /**
    * Yields the latest user write checkpoint whenever the sync checkpoint updates.
    *
    * The stream stops or errors if this is not the active sync rules (anymore).
    */
-  watchWriteCheckpoint(options: WatchWriteCheckpointOptions): AsyncIterable<WriteCheckpoint>;
+  watchWriteCheckpoint(options: WatchWriteCheckpointOptions): AsyncIterable<StorageCheckpointUpdate>;
 
   /**
    * Get a "batch" of data for a checkpoint.
@@ -217,8 +219,6 @@ export interface WatchWriteCheckpointOptions {
   user_id: string;
 
   signal: AbortSignal;
-
-  filter?: (event: WatchFilterEvent) => boolean;
 }
 
 export interface WatchFilterEvent {
@@ -231,3 +231,19 @@ export interface WriteCheckpoint {
   base: ReplicationCheckpoint;
   writeCheckpoint: bigint | null;
 }
+
+export interface StorageCheckpointUpdate extends WriteCheckpoint {
+  update: CheckpointChanges;
+}
+
+export interface GetCheckpointChangesOptions {
+  lastCheckpoint: util.OpId;
+  nextCheckpoint: util.OpId;
+}
+
+export interface CheckpointChanges {
+  updatedDataBuckets: string[];
+  invalidateDataBuckets: boolean;
+  updatedParameterBucketDefinitions: string[];
+  invalidateParameterBuckets: boolean;
+}