Use versioned bucket ids

Author: simolus3

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

@@ -45,6 +45,7 @@ const replicationMutex = new utils.Mutex();
 export interface MongoBucketBatchOptions {
   db: PowerSyncMongo;
   syncRules: SqlSyncRules;
+  syncRulesId: string;
   groupId: number;
   slotName: string;
   lastCheckpointLsn: string | null;
@@ -72,6 +73,7 @@ export class MongoBucketBatch
   public readonly db: PowerSyncMongo;
   public readonly session: mongo.ClientSession;
   private readonly sync_rules: SqlSyncRules;
+  private readonly syncRulesId: string;
 
   private readonly group_id: number;
 
@@ -127,6 +129,7 @@ export class MongoBucketBatch
     this.session = this.client.startSession();
     this.slot_name = options.slotName;
     this.sync_rules = options.syncRules;
+    this.syncRulesId = options.syncRulesId;
     this.storeCurrentData = options.storeCurrentData;
     this.skipExistingRows = options.skipExistingRows;
     this.markRecordUnavailable = options.markRecordUnavailable;
@@ -461,7 +464,8 @@ export class MongoBucketBatch
       if (sourceTable.syncData) {
         const { results: evaluated, errors: syncErrors } = this.sync_rules.evaluateRowWithErrors({
           record: after,
-          sourceTable
+          sourceTable,
+          bucketIdTransformer: SqlSyncRules.versionedBucketIdTransformer(this.syncRulesId)
         });
 
         for (let error of syncErrors) {

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

@@ -154,6 +154,7 @@ export class MongoSyncBucketStorage
       logger: options.logger,
       db: this.db,
       syncRules: this.sync_rules.parsed(options).sync_rules,
+      syncRulesId: `${this.sync_rules.id}`,
       groupId: this.group_id,
       slotName: this.slot_name,
       lastCheckpointLsn: checkpoint_lsn,

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

@@ -330,6 +330,7 @@ export class PostgresSyncRulesStorage
       logger: options.logger ?? framework.logger,
       db: this.db,
       sync_rules: this.sync_rules.parsed(options).sync_rules,
+      syncRulesId: `${this.sync_rules.id}`,
       group_id: this.group_id,
       slot_name: this.slot_name,
       last_checkpoint_lsn: checkpoint_lsn,

modules/module-postgres-storage/src/storage/batch/PostgresBucketBatch.ts

@@ -25,6 +25,7 @@ export interface PostgresBucketBatchOptions {
   logger: Logger;
   db: lib_postgres.DatabaseClient;
   sync_rules: sync_rules.SqlSyncRules;
+  syncRulesId: string;
   group_id: number;
   slot_name: string;
   last_checkpoint_lsn: string | null;
@@ -73,6 +74,7 @@ export class PostgresBucketBatch
 
   protected write_checkpoint_batch: storage.CustomWriteCheckpointOptions[];
   protected readonly sync_rules: sync_rules.SqlSyncRules;
+  private readonly syncRulesId: string;
   protected batch: OperationBatch | null;
   private lastWaitingLogThrottled = 0;
   private markRecordUnavailable: BucketStorageMarkRecordUnavailable | undefined;
@@ -88,6 +90,7 @@ export class PostgresBucketBatch
     this.resumeFromLsn = options.resumeFromLsn;
     this.write_checkpoint_batch = [];
     this.sync_rules = options.sync_rules;
+    this.syncRulesId = options.syncRulesId;
     this.markRecordUnavailable = options.markRecordUnavailable;
     this.batch = null;
     this.persisted_op = null;
@@ -825,7 +828,8 @@ export class PostgresBucketBatch
       if (sourceTable.syncData) {
         const { results: evaluated, errors: syncErrors } = this.sync_rules.evaluateRowWithErrors({
           record: after,
-          sourceTable
+          sourceTable,
+          bucketIdTransformer: sync_rules.SqlSyncRules.versionedBucketIdTransformer(this.syncRulesId)
         });
 
         for (const error of syncErrors) {

packages/sync-rules/src/BaseSqlDataQuery.ts

@@ -5,6 +5,7 @@ import { SourceTableInterface } from './SourceTableInterface.js';
 import { SqlTools } from './sql_filters.js';
 import { TablePattern } from './TablePattern.js';
 import {
+  BucketIdTransformer,
   EvaluationResult,
   QueryParameters,
   QuerySchema,
@@ -25,6 +26,7 @@ export interface EvaluateRowOptions {
   table: SourceTableInterface;
   row: SqliteRow;
   bucketIds: (params: QueryParameters) => string[];
+  bucketIdTransformer: BucketIdTransformer | null;
 }
 
 export interface BaseSqlDataQueryOptions {
@@ -177,7 +179,7 @@ export class BaseSqlDataQuery {
 
   evaluateRowWithOptions(options: EvaluateRowOptions): EvaluationResult[] {
     try {
-      const { table, row, bucketIds } = options;
+      const { table, row, bucketIds, bucketIdTransformer } = options;
 
       const tables = { [this.table]: this.addSpecialParameters(table, row) };
       const resolvedBucketIds = bucketIds(tables);
@@ -197,7 +199,7 @@ export class BaseSqlDataQuery {
 
       return resolvedBucketIds.map((bucketId) => {
         return {
-          bucket: bucketId,
+          bucket: bucketIdTransformer ? bucketIdTransformer(bucketId) : bucketId,
           table: outputTable,
           id: id,
           data

packages/sync-rules/src/SqlBucketDescriptor.ts

@@ -104,7 +104,17 @@ export class SqlBucketDescriptor implements BucketSource {
         continue;
       }
 
-      results.push(...query.evaluateRow(options.sourceTable, applyRowContext(options.record, this.compatibility)));
+      const bucketIdTransformer = this.compatibility.isFixed(Quirk.versionedBucketIds)
+        ? options.bucketIdTransformer
+        : null;
+
+      results.push(
+        ...query.evaluateRow(
+          options.sourceTable,
+          applyRowContext(options.record, this.compatibility),
+          bucketIdTransformer
+        )
+      );
     }
     return results;
   }

packages/sync-rules/src/SqlDataQuery.ts

@@ -10,7 +10,7 @@ import { checkUnsupportedFeatures, isClauseError } from './sql_support.js';
 import { SyncRulesOptions } from './SqlSyncRules.js';
 import { TablePattern } from './TablePattern.js';
 import { TableQuerySchema } from './TableQuerySchema.js';
-import { EvaluationResult, ParameterMatchClause, QuerySchema, SqliteRow } from './types.js';
+import { BucketIdTransformer, EvaluationResult, ParameterMatchClause, QuerySchema, SqliteRow } from './types.js';
 import { getBucketId, isSelectStatement } from './utils.js';
 
 export interface SqlDataQueryOptions extends BaseSqlDataQueryOptions {
@@ -185,10 +185,15 @@ export class SqlDataQuery extends BaseSqlDataQuery {
     this.filter = options.filter;
   }
 
-  evaluateRow(table: SourceTableInterface, row: SqliteRow): EvaluationResult[] {
+  evaluateRow(
+    table: SourceTableInterface,
+    row: SqliteRow,
+    bucketIdTransformer: BucketIdTransformer | null
+  ): EvaluationResult[] {
     return this.evaluateRowWithOptions({
       table,
       row,
+      bucketIdTransformer,
       bucketIds: (tables) => {
         const bucketParameters = this.filter.filterRow(tables);
         return bucketParameters.map((params) => getBucketId(this.descriptorName, this.bucketParameters, params));

packages/sync-rules/src/SqlSyncRules.ts

@@ -494,4 +494,8 @@ export class SqlSyncRules implements SyncRules {
       }
     }
   }
+
+  static versionedBucketIdTransformer(version: string) {
+    return (bucketId: string) => `${version}#${bucketId}`;
+  }
 }

packages/sync-rules/src/quirks.ts

@@ -26,9 +26,15 @@ export class Quirk {
     CompatibilityLevel.SYNC_STREAMS
   );
 
+  static versionedBucketIds = new Quirk(
+    'versioned_bucket_ids',
+    'Old versions of the sync service did not update bucket ids when sync rules change. This can lead to a resync on clients taking longer than necessary, and break client-side progress reports.',
+    CompatibilityLevel.SYNC_STREAMS
+  );
+
   static byName: Record<string, Quirk> = (() => {
     const byName: Record<string, Quirk> = {};
-    for (const entry of [this.nonIso8601Timestamps]) {
+    for (const entry of [this.nonIso8601Timestamps, this.versionedBucketIds]) {
       byName[entry.name] = entry;
     }
 

packages/sync-rules/src/types.ts

@@ -283,7 +283,24 @@ export interface InputParameter {
   parametersToLookupValue(parameters: ParameterValueSet): SqliteValue;
 }
 
-export interface EvaluateRowOptions extends TableRow<SqliteInputRow> {}
+/**
+ * Transforms bucket ids generated when evaluating the row by e.g. encoding version information.
+ *
+ * Because buckets are recreated on a sync rule redeploy, it makes sense to use different bucket ids (otherwise, clients
+ * may run into checksum errors causing a sync to take longer than necessary or breaking progress).
+ *
+ * So, this transformer receives the original bucket id as generated by defined sync rules, and can prepend a version
+ * identifier.
+ *
+ * Note that this transformation has not been present in older versions of the sync service. To preserve backwards
+ * compatibility, data queries will not invoke this function by default when creating bucket ids. Using this function
+ * requires an opt-in.
+ */
+export type BucketIdTransformer = (regularId: string) => string;
+
+export interface EvaluateRowOptions extends TableRow<SqliteInputRow> {
+  bucketIdTransformer: BucketIdTransformer;
+}
 
 /**
  * A row associated with the table it's coming from.