Configurable storage versions (#601)

* WIP: Implement models for incremental reprocessing.

* Handle current_data v1/v3 differences.

* Split out PersistedBatch implementations.

* Split out MongoBucketBatch implementations.

* Resolve circular imports.

* Back to old bucket names for now.

* Fix type check.

* nullable CurrentDataDocumentV3.data.

* Use string ids.

* Split collections for bucket_data.

* Use clustered collections.

* Drop bucket_data collections when clearing.

* Fix type issues.

* Fixes.

* Fix tests.

* Split out checksum implementations.

* Split bucket_parameter collections.

* Initialize collections upfront.

* Workaround for MongoDB SERVER-121822.

* Update snapshots.

* Fix for parameter lookups.

* Optimize parameter query lookups.

* Minor restructuring.

* Split current_data into separate source_record_ collections.

* Refactor _id for source_records.

* Split out CurrentDataStore.

* Further split out implementations.

* Rename CurrentData -> SourceRecord.

* Split out source_tables collections.

* Don't do initializeCurrentDataCollection for v1.

* Initialize source records collection on resolveTable instead of flush.

* Refactor more collection initialization.

* Restructure v3 parameter index lookup values.

* Update tests.

* Further test fix.

* Document collection structure.

* Add some comments.

* Rename.

* Use $unionWith to find parameter index changes.

* Fix current_data / source_records structure.

* Further fixes for v1 current_data.

* Cleanup.

* Fix type issue post merge.

* Rename postCommitCleanup.

* Track pending deletes & fix other source_table issues.

* Only cleanup pending deletes for affected source tables.

* Add timeout and retry for deletes.

* Increase clear timeout.

* Retry inner clear operations instead of the entire loop.

* Split bucket_state collections.

* Split compactor implementations.

* Avoid listing collections unless we drop them.

* Split implementation versions, phase 1.

* Split MongoChecksums.

* Split MongoParameterCompactor.

* Split methods in PersistedBatch.

* Further split MongoBucketBatch.

* Move out helpers for MongoSyncBucketStorage

* Refactor MongoSyncBucketStorage.

* Split db implementations.

* Cleanup.

* Tweak types.

* More type tweaks.

* Type tweak.

* Simplify MongoChecksums.

* Remove MongoCompactorBase.

* Remove compatibility re-rexports.

* Remove MongoParameterCompactorBase.

* Rename MongoSyncBucketStorage.

* Simplify db.

* Split v1/v3 models.

* Move files back to their original location.

* Update docs.

* Add back clear for current_data.

* Add back missing metric tracking.

* Check if collection exists before getting storage stats.

* Report metrics even if there are no active replication streams yet.

* Update comment.

* Some collection listing cleanup.

* Optimize bucket collection lookup for v3 compacting.

* Remove unused functions.

* Organize imports.

* Cleanup types.

* Reduce type casting in tests.

* Further improve test types.

* More type cast improvements and comments.

* Reduce instance state on MongoCompactor.

* Introduce SingleBucketStore.

* Minor cleanup.

* Explicitly list fields when persisting.

* Merge implementations for saveBucketData.

* Fix regression.

* Avoid double-parsing sync rules in many cases when updating.

* Add storage version parsing.

* Use configured storage version.

* Changeset.

* Re-use persisted storage version in asUpdateOptions().

* Further tweaks and comments.

* Fix tests.

Author: rkistner

.changeset/nice-eels-dance.md

@@ -0,0 +1,11 @@
+---
+'@powersync/service-module-postgres-storage': patch
+'@powersync/service-module-mongodb-storage': patch
+'@powersync/service-core-tests': patch
+'@powersync/service-module-postgres': patch
+'@powersync/service-core': patch
+'@powersync/service-sync-rules': patch
+'@powersync/lib-service-mongodb': patch
+---
+
+Add `config.storage_version` configuration option.

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

@@ -156,7 +156,8 @@ export class MongoBucketStorage extends storage.BucketStorageFactory {
   }
 
   async updateSyncRules(options: storage.UpdateSyncRulesOptions): Promise<MongoPersistedSyncRulesContent> {
-    const storageVersion = options.storageVersion ?? storage.CURRENT_STORAGE_VERSION;
+    const storageVersion =
+      options.storageVersion ?? options.config.parsed.config.storageVersion ?? storage.CURRENT_STORAGE_VERSION;
     const storageConfig = getMongoStorageConfig(storageVersion);
 
     let rules: MongoPersistedSyncRulesContent | undefined = undefined;

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

@@ -154,7 +154,8 @@ export class PostgresBucketStorageFactory extends storage.BucketStorageFactory {
   }
 
   async updateSyncRules(options: storage.UpdateSyncRulesOptions): Promise<PostgresPersistedSyncRulesContent> {
-    const storageVersion = options.storageVersion ?? storage.CURRENT_STORAGE_VERSION;
+    const storageVersion =
+      options.storageVersion ?? options.config.parsed.config.storageVersion ?? storage.CURRENT_STORAGE_VERSION;
     const storageConfig = storage.STORAGE_VERSION_CONFIG[storageVersion];
     if (storageConfig == null) {
       throw new framework.ServiceError(

packages/service-core/src/routes/endpoints/admin.ts

@@ -131,6 +131,10 @@ export const reprocess = routeDefinition({
       });
     }
 
+    // There are some differences between this and using asUpdateOptions():
+    // 1. This always re-parses the source YAML. If there are changes to the sync stream compiler, that can affect the sync plan.
+    // 2. If the source does not set the storage version, this will update it do the current version.
+    // We can consider tweaking this behavior in the future.
     const new_rules = await activeBucketStorage.updateSyncRules(
       storage.updateSyncRulesFromYaml(active.sync_rules.config.content, {
         // These sync rules already passed validation. But if the rules are not valid anymore due

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

@@ -103,6 +103,10 @@ export abstract class PersistedSyncRulesContent implements PersistedSyncRulesCon
         sourceText: this.sync_rules_content
       });
 
+      // Note: If the original content did not define a storage version, this will still set the storage version.
+      // This means asUpdateOptions will not change the storage version, even if the default changes.
+      precompiled.storageVersion = this.storageVersion;
+
       const errors: YamlError[] = [];
       if (this.compiled_plan.errors) {
         for (const error of this.compiled_plan.errors) {

packages/sync-rules/src/StorageVersion.ts

@@ -0,0 +1,37 @@
+/**
+ * This is only for the purpose of validating the version in the sync config, for reporting configurable versions
+ * upfront.
+ *
+ * The service itself may:
+ * 1. Support additional storage versions, such as legacy storage versions.
+ * 2. Attach specific behavior to storage versions.
+ *
+ * See: service-core/src/storage/StorageVersionConfig.js
+ */
+export interface ValidatedStorageVersion {
+  version: number;
+
+  /**
+   * If false, this version may be dropped or fundamentally changed in any future service version.
+   */
+  stable: boolean;
+}
+
+export const STORAGE_VERSIONS = new Map<number, ValidatedStorageVersion>([
+  // version 1 is supported by the storage modules, but cannot be used in sync config
+  [2, { version: 2, stable: true }],
+  [3, { version: 3, stable: false }]
+]);
+
+export const DEFAULT_STORAGE_VERSION = STORAGE_VERSIONS.get(2)!;
+
+/**
+ * Parse a storage version.
+ *
+ * If the version number is unknown or not supported, returns undefined.
+ *
+ * Generally, even storage versions are stable, and odd storage versions unstable.
+ */
+export function validateStorageVersion(version: number): ValidatedStorageVersion | undefined {
+  return STORAGE_VERSIONS.get(version);
+}

packages/sync-rules/src/SyncConfig.ts

@@ -17,6 +17,12 @@ export abstract class SyncConfig {
   bucketParameterLookupSources: ParameterIndexLookupCreator[] = [];
   bucketSources: BucketSource[] = [];
   compatibility: CompatibilityContext = CompatibilityContext.FULL_BACKWARDS_COMPATIBILITY;
+  /**
+   * If not defined, the storage module picks the latest stable version.
+   *
+   * Only supported storage versions can be set here when parsing from yaml.
+   */
+  storageVersion: number | undefined;
   eventDescriptors: SqlEventDescriptor[] = [];
 
   /**

packages/sync-rules/src/from_yaml.ts

@@ -13,6 +13,7 @@ import { SqlEventDescriptor } from './events/SqlEventDescriptor.js';
 import { validateSyncRulesSchema } from './json_schema.js';
 import { QueryParseResult, SqlBucketDescriptor } from './SqlBucketDescriptor.js';
 import { SqlSyncRules } from './SqlSyncRules.js';
+import { validateStorageVersion } from './StorageVersion.js';
 import { syncStreamFromSql } from './streams/from_sql.js';
 import { javaScriptExpressionEngine } from './sync_plan/engine/javascript.js';
 import { PrecompiledSyncConfig } from './sync_plan/evaluator/index.js';
@@ -69,9 +70,11 @@ export class SyncConfigFromYaml {
     }
 
     let compatibility: CompatibilityContext;
+    let storageVersion: number | undefined;
     if (parsed.has('config')) {
       const declaredOptions = parsed.get('config') as YAMLMap;
       compatibility = this.#parseCompatibilityOptions(declaredOptions);
+      storageVersion = this.#validateStorageVersion(declaredOptions);
     } else {
       compatibility = CompatibilityContext.FULL_BACKWARDS_COMPATIBILITY;
     }
@@ -98,6 +101,8 @@ export class SyncConfigFromYaml {
       result = this.#legacyParseBucketDefinitionsAndStreams(bucketMap, streamMap, compatibility);
     }
 
+    result.storageVersion = storageVersion;
+
     const eventDefinitions = this.#parseEventDefinitions(parsed, compatibility);
     result.eventDescriptors.push(...eventDefinitions);
 
@@ -411,6 +416,31 @@ export class SyncConfigFromYaml {
     return rules;
   }
 
+  #validateStorageVersion(config: YAMLMap): number | undefined {
+    const storageScalar = config.get('storage_version', true);
+    if (storageScalar != null) {
+      if (typeof storageScalar.value == 'number') {
+        const rawVersion = storageScalar.value;
+        const version = validateStorageVersion(storageScalar.value);
+        if (version == null) {
+          this.#errors.push(this.#yamlError(storageScalar, `Storage version ${storageScalar.value} is not supported`));
+        } else if (!version.stable) {
+          const error = this.#yamlError(
+            storageScalar,
+            `Storage version ${version.version} is unstable, and may cause unexpected behavior or stop functioning in any release`
+          );
+          error.type = 'warning';
+          this.#errors.push(error);
+        }
+        return version?.version;
+      } else {
+        this.#errors.push(this.#yamlError(storageScalar, 'Storage version must be numeric'));
+        return undefined;
+      }
+    }
+    return undefined;
+  }
+
   #parseEventDefinitions(parsed: Document, compatibility: CompatibilityContext) {
     const eventMap = parsed.get('event_definitions') as YAMLMap;
     const eventDescriptors: SqlEventDescriptor[] = [];

packages/sync-rules/src/index.ts

@@ -22,6 +22,7 @@ export * from './SqlDataQuery.js';
 export * from './SqlParameterQuery.js';
 export * from './SqlSyncRules.js';
 export * from './StaticSchema.js';
+export * from './StorageVersion.js';
 export { syncStreamFromSql } from './streams/from_sql.js';
 export { STREAM_FUNCTIONS } from './streams/functions.js';
 export { SyncStream } from './streams/stream.js';

packages/sync-rules/src/json_schema.ts

@@ -1,5 +1,6 @@
 import ajvModule from 'ajv';
 import { CompatibilityEdition, CompatibilityOption, TimeValuePrecision } from './compatibility.js';
+import { DEFAULT_STORAGE_VERSION, STORAGE_VERSIONS } from './StorageVersion.js';
 // Hack to make this work both in NodeJS and a browser
 const Ajv = ajvModule.default ?? ajvModule;
 const ajv = new Ajv({ allErrors: true, verbose: true });
@@ -141,6 +142,12 @@ export const syncRulesSchema: ajvModule.Schema = {
           minimum: CompatibilityEdition.LEGACY,
           exclusiveMaximum: CompatibilityEdition.COMPILED_STREAMS + 1
         },
+        storage_version: {
+          type: 'integer',
+          description: 'Storage version used by the storage database. By default, the latest stable version is used.',
+          default: DEFAULT_STORAGE_VERSION.version,
+          enum: [...STORAGE_VERSIONS.keys()]
+        },
         timestamp_max_precision: {
           type: 'string',
           enum: Object.values(TimeValuePrecision.byName).map((e) => e.name)