Storage version

.changeset/short-cheetahs-unite.md

@@ -0,0 +1,14 @@
+---
+'@powersync/service-module-postgres-storage': minor
+'@powersync/service-module-mongodb-storage': minor
+'@powersync/service-core-tests': minor
+'@powersync/service-module-postgres': minor
+'@powersync/service-module-mongodb': minor
+'@powersync/service-core': minor
+'@powersync/service-module-mssql': minor
+'@powersync/service-module-mysql': minor
+'@powersync/service-sync-rules': minor
+'@powersync/service-errors': patch
+---
+
+Introduce storage versions.

modules/module-mongodb-storage/src/migrations/db/migrations/1770213298299-storage-version.ts

@@ -0,0 +1,44 @@
+import { migrations, storage as core_storage } from '@powersync/service-core';
+import * as mongo_storage from '../../../storage/storage-index.js';
+import { MongoStorageConfig } from '../../../types/types.js';
+
+export const up: migrations.PowerSyncMigrationFunction = async (context) => {
+  const {
+    service_context: { configuration }
+  } = context;
+  const db = mongo_storage.createPowerSyncMongo(configuration.storage as MongoStorageConfig);
+
+  try {
+    await db.sync_rules.updateMany(
+      { storage_version: { $exists: false } },
+      { $set: { storage_version: core_storage.LEGACY_STORAGE_VERSION } }
+    );
+  } finally {
+    await db.client.close();
+  }
+};
+
+export const down: migrations.PowerSyncMigrationFunction = async (context) => {
+  const {
+    service_context: { configuration }
+  } = context;
+
+  const db = mongo_storage.createPowerSyncMongo(configuration.storage as MongoStorageConfig);
+
+  try {
+    const newRules = await db.sync_rules
+      .find({ storage_version: { $gt: core_storage.LEGACY_STORAGE_VERSION } })
+      .toArray();
+    if (newRules.length > 0) {
+      throw new Error(
+        `Cannot revert migration due to newer storage versions in use: ${newRules.map((r) => `${r._id}: v${r.storage_version}`).join(', ')}`
+      );
+    }
+    await db.sync_rules.updateMany(
+      { storage_version: core_storage.LEGACY_STORAGE_VERSION },
+      { $unset: { storage_version: 1 } }
+    );
+  } finally {
+    await db.client.close();
+  }
+};

modules/module-mongodb-storage/src/storage/MongoBucketStorage.ts

@@ -13,6 +13,11 @@ import { SyncRuleDocument } from './implementation/models.js';
 import { MongoPersistedSyncRulesContent } from './implementation/MongoPersistedSyncRulesContent.js';
 import { MongoSyncBucketStorage, MongoSyncBucketStorageOptions } from './implementation/MongoSyncBucketStorage.js';
 import { generateSlotName } from '../utils/util.js';
+import { MongoChecksumOptions } from './implementation/MongoChecksums.js';
+
+export interface MongoBucketStorageOptions {
+  checksumOptions?: Omit<MongoChecksumOptions, 'storageConfig'>;
+}
 
 export class MongoBucketStorage
   extends BaseObserver<storage.BucketStorageFactoryListener>
@@ -32,7 +37,7 @@ export class MongoBucketStorage
     options: {
       slot_name_prefix: string;
     },
-    private internalOptions?: MongoSyncBucketStorageOptions
+    private internalOptions?: MongoBucketStorageOptions
   ) {
     super();
     this.client = db.client;
@@ -50,10 +55,15 @@ export class MongoBucketStorage
     if ((typeof id as any) == 'bigint') {
       id = Number(id);
     }
-    const storage = new MongoSyncBucketStorage(this, id, syncRules, slot_name, undefined, this.internalOptions);
+    const storageConfig = (syncRules as MongoPersistedSyncRulesContent).getStorageConfig();
+    const storage = new MongoSyncBucketStorage(this, id, syncRules, slot_name, undefined, {
+      ...this.internalOptions,
+      storageConfig
+    });
     if (!options?.skipLifecycleHooks) {
       this.iterateListeners((cb) => cb.syncStorageCreated?.(storage));
     }
+
     storage.registerListener({
       batchStarted: (batch) => {
         batch.registerListener({
@@ -205,8 +215,10 @@ export class MongoBucketStorage
       const id = Number(id_doc!.op_id);
       const slot_name = generateSlotName(this.slot_name_prefix, id);
 
+      const storageVersion = options.storageVersion ?? storage.CURRENT_STORAGE_VERSION;
       const doc: SyncRuleDocument = {
         _id: id,
+        storage_version: storageVersion,
         content: options.content,
         last_checkpoint: null,
         last_checkpoint_lsn: null,

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

@@ -13,6 +13,7 @@ import {
   PartialOrFullChecksum
 } from '@powersync/service-core';
 import { PowerSyncMongo } from './db.js';
+import { StorageConfig } from './models.js';
 
 /**
  * Checksum calculation options, primarily for tests.
@@ -27,6 +28,8 @@ export interface MongoChecksumOptions {
    * Limit on the number of documents to calculate a checksum on at a time.
    */
   operationBatchLimit?: number;
+
+  storageConfig: StorageConfig;
 }
 
 const DEFAULT_BUCKET_BATCH_LIMIT = 200;
@@ -43,12 +46,15 @@ const DEFAULT_OPERATION_BATCH_LIMIT = 50_000;
  */
 export class MongoChecksums {
   private _cache: ChecksumCache | undefined;
+  private readonly storageConfig: StorageConfig;
 
   constructor(
     private db: PowerSyncMongo,
     private group_id: number,
-    private options?: MongoChecksumOptions
-  ) {}
+    private options: MongoChecksumOptions
+  ) {
+    this.storageConfig = options.storageConfig;
+  }
 
   /**
    * Lazy-instantiated cache.
@@ -222,6 +228,11 @@ export class MongoChecksums {
         });
       }
 
+      // Historically, checksum may be stored as 'int' or 'double'.
+      // More recently, this should be a 'long'.
+      // $toLong ensures that we always sum it as a long, avoiding inaccuracies in the calculations.
+      const checksumLong = this.storageConfig.longChecksums ? '$checksum' : { $toLong: '$checksum' };
+
       // Aggregate over a max of `batchLimit` operations at a time.
       // Let's say we have 3 buckets (A, B, C), each with 10 operations, and our batch limit is 12.
       // Then we'll do three batches:
@@ -245,10 +256,7 @@ export class MongoChecksums {
             {
               $group: {
                 _id: '$_id.b',
-                // Historically, checksum may be stored as 'int' or 'double'.
-                // More recently, this should be a 'long'.
-                // $toLong ensures that we always sum it as a long, avoiding inaccuracies in the calculations.
-                checksum_total: { $sum: { $toLong: '$checksum' } },
+                checksum_total: { $sum: checksumLong },
                 count: { $sum: 1 },
                 has_clear_op: {
                   $max: {

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

@@ -1,20 +1,42 @@
-import { SyncConfigWithErrors, HydratedSyncRules, versionedHydrationState } from '@powersync/service-sync-rules';
+import {
+  CompatibilityOption,
+  DEFAULT_HYDRATION_STATE,
+  HydratedSyncRules,
+  HydrationState,
+  SyncConfigWithErrors,
+  versionedHydrationState
+} from '@powersync/service-sync-rules';
 
 import { storage } from '@powersync/service-core';
 
+import { StorageConfig } from './models.js';
+
 export class MongoPersistedSyncRules implements storage.PersistedSyncRules {
   public readonly slot_name: string;
+  public readonly hydrationState: HydrationState;
 
   constructor(
     public readonly id: number,
     public readonly sync_rules: SyncConfigWithErrors,
     public readonly checkpoint_lsn: string | null,
-    slot_name: string | null
+    slot_name: string | null,
+    public readonly storageConfig: StorageConfig
   ) {
     this.slot_name = slot_name ?? `powersync_${id}`;
+
+    if (
+      storageConfig.versionedBuckets ||
+      this.sync_rules.config.compatibility.isEnabled(CompatibilityOption.versionedBucketIds)
+    ) {
+      // For new sync config versions (using the new storage version), we always enable versioned bucket names.
+      // For older versions, this depends on the compatibility option.
+      this.hydrationState = versionedHydrationState(this.id);
+    } else {
+      this.hydrationState = DEFAULT_HYDRATION_STATE;
+    }
   }
 
   hydratedSyncRules(): HydratedSyncRules {
-    return this.sync_rules.config.hydrate({ hydrationState: versionedHydrationState(this.id) });
+    return this.sync_rules.config.hydrate({ hydrationState: this.hydrationState });
   }
 }

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

@@ -4,7 +4,8 @@ import { SqlSyncRules } from '@powersync/service-sync-rules';
 import { MongoPersistedSyncRules } from './MongoPersistedSyncRules.js';
 import { MongoSyncRulesLock } from './MongoSyncRulesLock.js';
 import { PowerSyncMongo } from './db.js';
-import { SyncRuleDocument } from './models.js';
+import { getMongoStorageConfig, SyncRuleDocument } from './models.js';
+import { ErrorCode, ServiceError } from '@powersync/lib-services-framework';
 
 export class MongoPersistedSyncRulesContent implements storage.PersistedSyncRulesContent {
   public readonly slot_name: string;
@@ -17,6 +18,7 @@ export class MongoPersistedSyncRulesContent implements storage.PersistedSyncRule
   public readonly last_keepalive_ts: Date | null;
   public readonly last_checkpoint_ts: Date | null;
   public readonly active: boolean;
+  public readonly storageVersion: number;
 
   public current_lock: MongoSyncRulesLock | null = null;
 
@@ -34,14 +36,32 @@ export class MongoPersistedSyncRulesContent implements storage.PersistedSyncRule
     this.last_checkpoint_ts = doc.last_checkpoint_ts;
     this.last_keepalive_ts = doc.last_keepalive_ts;
     this.active = doc.state == 'ACTIVE';
+    this.storageVersion = doc.storage_version ?? storage.LEGACY_STORAGE_VERSION;
+  }
+
+  /**
+   * Load the storage config.
+   *
+   * This may throw if the persisted storage version is not supported.
+   */
+  getStorageConfig() {
+    const storageConfig = getMongoStorageConfig(this.storageVersion);
+    if (storageConfig == null) {
+      throw new ServiceError(
+        ErrorCode.PSYNC_S1005,
+        `Unsupported storage version ${this.storageVersion} for sync rules ${this.id}`
+      );
+    }
+    return storageConfig;
   }
 
   parsed(options: storage.ParseSyncRulesOptions) {
     return new MongoPersistedSyncRules(
       this.id,
       SqlSyncRules.fromYaml(this.sync_rules_content, options),
       this.last_checkpoint_lsn,
-      this.slot_name
+      this.slot_name,
+      this.getStorageConfig()
     );
   }
 

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

@@ -32,15 +32,23 @@ import * as timers from 'timers/promises';
 import { idPrefixFilter, mapOpEntry, readSingleBatch, setSessionSnapshotTime } from '../../utils/util.js';
 import { MongoBucketStorage } from '../MongoBucketStorage.js';
 import { PowerSyncMongo } from './db.js';
-import { BucketDataDocument, BucketDataKey, BucketStateDocument, SourceKey, SourceTableDocument } from './models.js';
+import {
+  BucketDataDocument,
+  BucketDataKey,
+  BucketStateDocument,
+  SourceKey,
+  SourceTableDocument,
+  StorageConfig
+} from './models.js';
 import { MongoBucketBatch } from './MongoBucketBatch.js';
 import { MongoChecksumOptions, MongoChecksums } from './MongoChecksums.js';
 import { MongoCompactor } from './MongoCompactor.js';
 import { MongoParameterCompactor } from './MongoParameterCompactor.js';
 import { MongoWriteCheckpointAPI } from './MongoWriteCheckpointAPI.js';
 
 export interface MongoSyncBucketStorageOptions {
-  checksumOptions?: MongoChecksumOptions;
+  checksumOptions?: Omit<MongoChecksumOptions, 'storageConfig'>;
+  storageConfig: StorageConfig;
 }
 
 /**
@@ -69,12 +77,15 @@ export class MongoSyncBucketStorage
     public readonly group_id: number,
     private readonly sync_rules: storage.PersistedSyncRulesContent,
     public readonly slot_name: string,
-    writeCheckpointMode?: storage.WriteCheckpointMode,
-    options?: MongoSyncBucketStorageOptions
+    writeCheckpointMode: storage.WriteCheckpointMode | undefined,
+    options: MongoSyncBucketStorageOptions
   ) {
     super();
     this.db = factory.db;
-    this.checksums = new MongoChecksums(this.db, this.group_id, options?.checksumOptions);
+    this.checksums = new MongoChecksums(this.db, this.group_id, {
+      ...options.checksumOptions,
+      storageConfig: options?.storageConfig
+    });
     this.writeCheckpointAPI = new MongoWriteCheckpointAPI({
       db: this.db,
       mode: writeCheckpointMode ?? storage.WriteCheckpointMode.MANAGED,

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

@@ -204,6 +204,29 @@ export interface SyncRuleDocument {
     id: string;
     expires_at: Date;
   } | null;
+
+  storage_version?: number;
+}
+
+export interface StorageConfig extends storage.StorageVersionConfig {
+  /**
+   * When true, bucket_data.checksum is guaranteed to be persisted as a Long.
+   *
+   * When false, it could also have been persisted as an Int32 or Double, in which case it must be converted to
+   * a Long before summing.
+   */
+  longChecksums: boolean;
+}
+
+const LONG_CHECKSUMS_STORAGE_VERSION = 2;
+
+export function getMongoStorageConfig(storageVersion: number): StorageConfig | undefined {
+  const baseConfig = storage.STORAGE_VERSION_CONFIG[storageVersion];
+  if (baseConfig == null) {
+    return undefined;
+  }
+
+  return { ...baseConfig, longChecksums: storageVersion >= LONG_CHECKSUMS_STORAGE_VERSION };
 }
 
 export interface CheckpointEventDocument {

modules/module-mongodb-storage/src/utils/test-utils.ts

@@ -1,14 +1,13 @@
 import { mongo } from '@powersync/lib-service-mongodb';
-import { PowerSyncMongo } from '../storage/implementation/db.js';
 import { TestStorageOptions } from '@powersync/service-core';
+import { MongoBucketStorage, MongoBucketStorageOptions } from '../storage/MongoBucketStorage.js';
 import { MongoReportStorage } from '../storage/MongoReportStorage.js';
-import { MongoBucketStorage } from '../storage/MongoBucketStorage.js';
-import { MongoSyncBucketStorageOptions } from '../storage/implementation/MongoSyncBucketStorage.js';
+import { PowerSyncMongo } from '../storage/implementation/db.js';
 
 export type MongoTestStorageOptions = {
   url: string;
   isCI: boolean;
-  internalOptions?: MongoSyncBucketStorageOptions;
+  internalOptions?: MongoBucketStorageOptions;
 };
 
 export function mongoTestStorageFactoryGenerator(factoryOptions: MongoTestStorageOptions) {