Migrate subkey format

Author: simolus3

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

@@ -65,14 +65,16 @@ export interface BucketStorageListener extends BaseListener {
 
 export interface BucketStorageAdapter extends BaseObserver<BucketStorageListener>, Disposable {
   init(): Promise<void>;
-  saveSyncData(batch: SyncDataBatch): Promise<void>;
+  saveSyncData(batch: SyncDataBatch, fixedKeyFormat: boolean): Promise<void>;
   removeBuckets(buckets: string[]): Promise<void>;
   setTargetCheckpoint(checkpoint: Checkpoint): Promise<void>;
 
   startSession(): void;
 
   getBucketStates(): Promise<BucketState[]>;
   getBucketOperationProgress(): Promise<BucketOperationProgress>;
+  hasMigratedSubkeys(): Promise<boolean>;
+  migrateToFixedSubkeys(): Promise<void>;
 
   syncLocalDatabase(
     checkpoint: Checkpoint,

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

@@ -8,7 +8,7 @@ export interface OplogEntryJSON {
   object_type?: string;
   op_id: string;
   op: OpTypeJSON;
-  subkey?: string | object;
+  subkey?: string;
 }
 
 export class OplogEntry {
@@ -17,7 +17,7 @@ export class OplogEntry {
       row.op_id,
       OpType.fromJSON(row.op),
       row.checksum,
-      typeof row.subkey == 'string' ? row.subkey : JSON.stringify(row.subkey),
+      row.subkey,
       row.object_type,
       row.object_id,
       row.data
@@ -28,21 +28,23 @@ export class OplogEntry {
     public op_id: OpId,
     public op: OpType,
     public checksum: number,
-    public subkey: string,
+    public subkey?: string,
     public object_type?: string,
     public object_id?: string,
     public data?: string
   ) {}
 
-  toJSON(): OplogEntryJSON {
+  toJSON(fixedKeyEncoding = false): OplogEntryJSON {
     return {
       op_id: this.op_id,
       op: this.op.toJSON(),
       object_type: this.object_type,
       object_id: this.object_id,
       checksum: this.checksum,
       data: this.data,
-      subkey: JSON.stringify(this.subkey)
+      // Older versions of the JS SDK used to always JSON.stringify here. That has always been wrong,
+      // but we need to migrate gradually to not break existing databases.
+      subkey: fixedKeyEncoding ? this.subkey : JSON.stringify(this.subkey)
     };
   }
 }

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

@@ -99,13 +99,13 @@ export class SqliteBucketStorage extends BaseObserver<BucketStorageListener> imp
     return Object.fromEntries(rows.map((r) => [r.name, { atLast: r.count_at_last, sinceLast: r.count_since_last }]));
   }
 
-  async saveSyncData(batch: SyncDataBatch) {
+  async saveSyncData(batch: SyncDataBatch, fixedKeyFormat: boolean) {
     await this.writeTransaction(async (tx) => {
       let count = 0;
       for (const b of batch.buckets) {
         const result = await tx.execute('INSERT INTO powersync_operations(op, data) VALUES(?, ?)', [
           'save',
-          JSON.stringify({ buckets: [b.toJSON()] })
+          JSON.stringify({ buckets: [b.toJSON(fixedKeyFormat)] })
         ]);
         this.logger.debug('saveSyncData', JSON.stringify(result));
         count += b.data.length;
@@ -420,6 +420,25 @@ export class SqliteBucketStorage extends BaseObserver<BucketStorageListener> imp
       return raw;
     });
   }
+
+  async hasMigratedSubkeys(): Promise<boolean> {
+    const { r } = await this.db.get<{ r: number }>('SELECT EXISTS(SELECT * FROM ps_kv WHERE key = ?) as r', [
+      SqliteBucketStorage._subkeyMigrationKey
+    ]);
+    return r != 0;
+  }
+
+  async migrateToFixedSubkeys(): Promise<void> {
+    await this.writeTransaction(async (tx) => {
+      await tx.execute('UPDATE ps_oplog SET key = powersync_remove_duplicate_key_encoding(key);');
+      await tx.execute('INSERT OR REPLACE INTO ps_kv (key, value) VALUES (?, ?);', [
+        SqliteBucketStorage._subkeyMigrationKey,
+        '1'
+      ]);
+    });
+  }
+
+  static _subkeyMigrationKey = 'powersync_js_migrated_subkeys';
 }
 
 function hasMatchingPriority(priority: number, bucket: BucketChecksum) {

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

@@ -37,13 +37,13 @@ export class SyncDataBucket {
     public next_after?: OpId
   ) {}
 
-  toJSON(): SyncDataBucketJSON {
+  toJSON(fixedKeyEncoding = false): SyncDataBucketJSON {
     return {
       bucket: this.bucket,
       has_more: this.has_more,
       after: this.after,
       next_after: this.next_after,
-      data: this.data.map((entry) => entry.toJSON())
+      data: this.data.map((entry) => entry.toJSON(fixedKeyEncoding))
     };
   }
 }

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

@@ -558,6 +558,31 @@ The next upload iteration will be delayed.`);
     return [req, localDescriptions];
   }
 
+  /**
+   * Older versions of the JS SDK used to encode subkeys as JSON in {@link OplogEntry.toJSON}.
+   * Because subkeys are always strings, this leads to quotes being added around them in `ps_oplog`.
+   * While this is not a problem as long as it's done consistently, it causes issues when a database
+   * created by the JS SDK is used with other SDKs, or (more likely) when the new Rust sync client
+   * is enabled.
+   *
+   * So, we add a migration from the old key format (with quotes) to the new one (no quotes). The
+   * migration is only triggered when necessary (for now). The function returns whether the new format
+   * should be used, so that the JS SDK is able to write to updated databases.
+   *
+   * @param requireFixedKeyFormat Whether we require the new format or also support the old one.
+   *        The Rust client requires the new subkey format.
+   * @returns Whether the database is now using the new, fixed subkey format.
+   */
+  private async requireKeyFormat(requireFixedKeyFormat: boolean): Promise<boolean> {
+    const hasMigrated = await this.options.adapter.hasMigratedSubkeys();
+    if (requireFixedKeyFormat && !hasMigrated) {
+      await this.options.adapter.migrateToFixedSubkeys();
+      return true;
+    } else {
+      return hasMigrated;
+    }
+  }
+
   protected async streamingSyncIteration(signal: AbortSignal, options?: PowerSyncConnectionOptions): Promise<void> {
     await this.obtainLock({
       type: LockType.SYNC,
@@ -571,6 +596,7 @@ The next upload iteration will be delayed.`);
         if (resolvedOptions.clientImplementation == SyncClientImplementation.JAVASCRIPT) {
           await this.legacyStreamingSyncIteration(signal, resolvedOptions);
         } else {
+          await this.requireKeyFormat(true);
           await this.rustSyncIteration(signal, resolvedOptions);
         }
       }
@@ -588,6 +614,7 @@ The next upload iteration will be delayed.`);
     let appliedCheckpoint: Checkpoint | null = null;
 
     const clientId = await this.options.adapter.getClientId();
+    const usingFixedKeyFormat = await this.requireKeyFormat(false);
 
     this.logger.debug('Requesting stream from server');
 
@@ -746,7 +773,7 @@ The next upload iteration will be delayed.`);
             downloadProgress: updatedProgress
           }
         });
-        await this.options.adapter.saveSyncData({ buckets: [SyncDataBucket.fromRow(data)] });
+        await this.options.adapter.saveSyncData({ buckets: [SyncDataBucket.fromRow(data)] }, usingFixedKeyFormat);
       } else if (isStreamingKeepalive(line)) {
         const remaining_seconds = line.token_expires_in;
         if (remaining_seconds == 0) {

packages/node/tests/sync.test.ts

@@ -22,6 +22,88 @@ describe('Sync', () => {
   describe('rust client', () => {
     defineSyncTests(SyncClientImplementation.RUST);
   });
+
+  mockSyncServiceTest('can migrate between sync implementations', async ({ syncService }) => {
+    function addData(id: string) {
+      syncService.pushLine({
+        data: {
+          bucket: 'a',
+          data: [
+            {
+              checksum: 0,
+              op_id: id,
+              op: 'PUT',
+              object_id: id,
+              object_type: 'lists',
+              subkey: `subkey_${id}`,
+              data: '{}'
+            }
+          ]
+        }
+      });
+    }
+    const checkpoint = {
+      checkpoint: {
+        last_op_id: '3',
+        buckets: [bucket('a', 3)]
+      }
+    };
+
+    let database = await syncService.createDatabase();
+    database.connect(new TestConnector(), {
+      clientImplementation: SyncClientImplementation.JAVASCRIPT,
+      connectionMethod: SyncStreamConnectionMethod.HTTP
+    });
+    await vi.waitFor(() => expect(syncService.connectedListeners).toHaveLength(1));
+    syncService.pushLine(checkpoint);
+    addData('1');
+
+    await vi.waitFor(async () => {
+      expect(await database.getAll('SELECT * FROM ps_oplog')).toHaveLength(1);
+    });
+    await database.disconnect();
+    // The JavaScript client encodes subkeys to JSON when it shouldn't...
+    expect(await database.getAll('SELECT * FROM ps_oplog')).toEqual([
+      expect.objectContaining({ key: 'lists/1/"subkey_1"' })
+    ]);
+
+    // Connecting again with the new client should fix the format
+    database.connect(new TestConnector(), {
+      clientImplementation: SyncClientImplementation.RUST,
+      connectionMethod: SyncStreamConnectionMethod.HTTP
+    });
+    await vi.waitFor(() => expect(syncService.connectedListeners).toHaveLength(1));
+    syncService.pushLine(checkpoint);
+    addData('2');
+    await vi.waitFor(async () => {
+      expect(await database.getAll('SELECT * FROM ps_oplog')).toHaveLength(2);
+    });
+    await database.disconnect();
+    expect(await database.getAll('SELECT * FROM ps_oplog')).toEqual([
+      // Existing entry should be fixed too!
+      expect.objectContaining({ key: 'lists/1/subkey_1' }),
+      expect.objectContaining({ key: 'lists/2/subkey_2' })
+    ]);
+
+    // Finally, connecting with JS again should keep the fixed subkey format.
+    database.connect(new TestConnector(), {
+      clientImplementation: SyncClientImplementation.RUST,
+      connectionMethod: SyncStreamConnectionMethod.HTTP
+    });
+    await vi.waitFor(() => expect(syncService.connectedListeners).toHaveLength(1));
+    syncService.pushLine(checkpoint);
+    addData('3');
+    await vi.waitFor(async () => {
+      expect(await database.getAll('SELECT * FROM ps_oplog')).toHaveLength(3);
+    });
+    await database.disconnect();
+    expect(await database.getAll('SELECT * FROM ps_oplog')).toEqual([
+      // Existing entry should be fixed too!
+      expect.objectContaining({ key: 'lists/1/subkey_1' }),
+      expect.objectContaining({ key: 'lists/2/subkey_2' }),
+      expect.objectContaining({ key: 'lists/3/subkey_3' })
+    ]);
+  });
 });
 
 function defineSyncTests(impl: SyncClientImplementation) {