Configurable storage versions

.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.

docs/storage-v3.md

@@ -0,0 +1,78 @@
+# Storage version 3 - Data structure
+
+## Replication stream
+
+A replication stream represents one conceptual replication "job":
+
+1. One logical replication stream on the client in Postgres.
+2. One change stream in MongoDB.
+3. Generally, one entity creating checkpoints from a source database stream.
+
+This does not refer to concurrency - we may add concurrency in each of these streams at a later point, which may use multiple underlying database streams. Instead, it just refers to conceptually having one replication job, advancing checkpoints one at a time.
+
+Right now, each "sync config version", or `sync_rules` document, is one replication stream.
+
+For incremental reprocessing, this will change so that multiple sync config versions can be processed by the same stream.
+
+It is possible to have multiple replication streams running concurrently, for example when:
+
+1. Incremental reprocessing is not used, so nothing is shared between the sync config versions.
+2. Changing storage versions - each replication stream can only handle one storage version at a time.
+
+## source_table
+
+Scoped to a replication stream.
+
+Collection: `source_table_${stream_id}`
+
+[FUTURE CHANGE] May have multiple copies per phyisical table per stream, especially when adding definitions to a stream.
+
+[FUTURE CHANGE] We can remove a source definition from a source table, but never add one.
+
+## source_records (previously current_data)
+
+Scoped to a source_table in a replication stream.
+
+Collection: `source_records_${stream_id}_${source_table_id}`
+
+The `_id` field is now the source row id. Unlike V1 storage model, this does not include `g` (group_id) or `t` (table id), since those are already encapsulated in the collection name.
+
+When a table is dropped, we first create relevant REMOVE operations, then drop the relevant current_data collection.
+
+[FUTURE CHANGE] If a _definition_ using a source table is removed:
+
+1. We remove the bucket_data (drop the collection - see below).
+2. We _don't_ update the source-records collection - stale records will remain. (Purely because this would be a slow operation, without gaining much)
+3. When re-processing a source record, we then check for orphaned references.
+
+When all definitions for a source table is removed, we remove the drop the corresponding source_records collection.
+
+## bucket_data
+
+Scoped by replication stream and definition id.
+
+Collection: `bucket_data_${stream_id}_${definition_id}`
+
+`_id.g` is removed, since this is encapsulated in the collection name now.
+
+`definition_id` is new here - that is not tracked in storage V1.
+
+[FUTURE CHANGE] collection must be dropped when the definition is removed.
+
+## parameter_index (previously bucket_parameters)
+
+Scoped by replication stream and index definition.
+
+Collection: `parameter_index_${stream_id}_${index_id}`
+
+_Also_ indexed by compound `key`, which includes {t: source_table_id, k: source_record_key}
+
+The `lookup` array drops the first two fields compared to V1 lookups (lookupName and queryId), since those are encapsulated in `index_id` in the collection name. In-memory, we use lookupName = indexId, queryId = '' (may change in the future).
+
+## bucket_state
+
+Scoped by replication stream.
+
+Collection: `bucket_state_${stream_id}`.
+
+`_id` is now compound: `{d: <definition id>, b: <bucket name>}` (previously `{g, b}`)

libs/lib-mongodb/src/db/mongo.ts

@@ -31,11 +31,13 @@ export const MONGO_OPERATION_TIMEOUT_MS = 40_000;
 export const MONGO_CHECKSUM_TIMEOUT_MS = 50_000;
 
 /**
- * Same as above, but specifically for clear operations.
+ * Same as MONGO_OPERATION_TIMEOUT_MS, but specifically for clear operations.
  *
  * These are retried when reaching the timeout.
+ *
+ * Used to be 5s. Increased to attempt to improve efficiency (deleted documents / scanned documents).
  */
-export const MONGO_CLEAR_OPERATION_TIMEOUT_MS = 5_000;
+export const MONGO_CLEAR_OPERATION_TIMEOUT_MS = MONGO_OPERATION_TIMEOUT_MS;
 
 export interface MongoConnectionOptions {
   maxPoolSize?: number;

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

@@ -7,14 +7,16 @@ import * as lib_mongo from '@powersync/lib-service-mongodb';
 import { mongo } from '@powersync/lib-service-mongodb';
 
 import { generateSlotName } from '../utils/util.js';
+import { BucketDefinitionMapping } from './implementation/BucketDefinitionMapping.js';
+import type { MongoSyncBucketStorage } from './implementation/createMongoSyncBucketStorage.js';
+import { createMongoSyncBucketStorage } from './implementation/createMongoSyncBucketStorage.js';
 import { PowerSyncMongo } from './implementation/db.js';
 import { getMongoStorageConfig, SyncRuleDocument } from './implementation/models.js';
 import { MongoChecksumOptions } from './implementation/MongoChecksums.js';
 import { MongoPersistedSyncRulesContent } from './implementation/MongoPersistedSyncRulesContent.js';
-import { MongoSyncBucketStorage } from './implementation/MongoSyncBucketStorage.js';
 
 export interface MongoBucketStorageOptions {
-  checksumOptions?: Omit<MongoChecksumOptions, 'storageConfig'>;
+  checksumOptions?: Omit<MongoChecksumOptions, 'storageConfig' | 'mapping'>;
 }
 
 export class MongoBucketStorage extends storage.BucketStorageFactory {
@@ -53,7 +55,7 @@ export class MongoBucketStorage extends storage.BucketStorageFactory {
       id = Number(id);
     }
     const storageConfig = (syncRules as MongoPersistedSyncRulesContent).getStorageConfig();
-    const storage = new MongoSyncBucketStorage(
+    const storage = createMongoSyncBucketStorage(
       this,
       id,
       syncRules as MongoPersistedSyncRulesContent,
@@ -154,9 +156,9 @@ 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?.version ?? storage.CURRENT_STORAGE_VERSION;
     const storageConfig = getMongoStorageConfig(storageVersion);
-    await this.db.initializeStorageVersion(storageConfig);
 
     let rules: MongoPersistedSyncRulesContent | undefined = undefined;
 
@@ -205,6 +207,10 @@ export class MongoBucketStorage extends storage.BucketStorageFactory {
         last_fatal_error_ts: null,
         last_keepalive_ts: null
       };
+      if (storageConfig.incrementalReprocessing) {
+        const parsed = options.config.parsed;
+        doc.rule_mapping = BucketDefinitionMapping.fromParsedSyncRules(parsed).serialize();
+      }
       await this.db.sync_rules.insertOne(doc);
       await this.db.notifyCheckpoint();
       rules = new MongoPersistedSyncRulesContent(this.db, doc);
@@ -296,63 +302,90 @@ export class MongoBucketStorage extends storage.BucketStorageFactory {
       }
     };
 
-    const active_sync_rules = await this.getActiveSyncRules({ defaultSchema: 'public' });
-    if (active_sync_rules == null) {
-      return {
-        operations_size_bytes: 0,
-        parameters_size_bytes: 0,
-        replication_size_bytes: 0
-      };
-    }
-    const operations_aggregate = await this.db.bucket_data
+    // For now, we get storage metrics over all v1 and v3 collections.
+    // In the future, we may split these metrics to report separately for active replication streams versus processing streams.
 
-      .aggregate([
-        {
-          $collStats: {
-            storageStats: {}
-          }
-        }
-      ])
-      .toArray()
-      .catch(ignoreNotExisting);
+    const aggregateStaticCollection = async <T extends mongo.Document>(collection: mongo.Collection<T>) => {
+      // We check whether the collection exists before getting the statistics. This avoids repeated
+      // errors in the MongoDB logs if the collection hasn't been created yet.
+      const exists =
+        (await this.db.db.listCollections({ name: collection.collectionName }, { nameOnly: true }).toArray()).length >
+        0;
+      if (!exists) {
+        return [{ storageStats: { size: 0 } }];
+      }
 
-    const parameters_aggregate = await this.db.bucket_parameters
-      .aggregate([
-        {
-          $collStats: {
-            storageStats: {}
+      return collection
+        .aggregate([
+          {
+            $collStats: {
+              storageStats: {}
+            }
           }
-        }
-      ])
-      .toArray()
-      .catch(ignoreNotExisting);
+        ])
+        .toArray()
+        .catch(ignoreNotExisting);
+    };
 
-    const v1_replication_aggregate = await this.db.current_data
-      .aggregate([
-        {
-          $collStats: {
-            storageStats: {}
-          }
-        }
-      ])
-      .toArray()
-      .catch(ignoreNotExisting);
+    const operations_aggregate = await aggregateStaticCollection(this.db.bucket_data);
+    const v3_operation_aggregates = await Promise.all(
+      (await this.db.listBucketDataCollectionsV3()).map((collection) =>
+        collection
+          .aggregate([
+            {
+              $collStats: {
+                storageStats: {}
+              }
+            }
+          ])
+          .toArray()
+          .catch(ignoreNotExisting)
+      )
+    );
 
-    const v3_replication_aggregate = await this.db.v3_current_data
-      .aggregate([
-        {
-          $collStats: {
-            storageStats: {}
-          }
-        }
-      ])
-      .toArray()
-      .catch(ignoreNotExisting);
+    const parameters_aggregate = await aggregateStaticCollection(this.db.bucket_parameters);
+
+    const v3_parameter_aggregates = await Promise.all(
+      (await this.db.listAllParameterIndexCollectionsV3()).map((collection) =>
+        collection
+          .aggregate([
+            {
+              $collStats: {
+                storageStats: {}
+              }
+            }
+          ])
+          .toArray()
+          .catch(ignoreNotExisting)
+      )
+    );
+
+    const v1_source_record_aggregate = await aggregateStaticCollection(this.db.current_data);
+
+    const source_record_aggregates = await Promise.all(
+      (await this.db.listAllSourceRecordCollectionsV3()).map((collection) =>
+        collection
+          .aggregate([
+            {
+              $collStats: {
+                storageStats: {}
+              }
+            }
+          ])
+          .toArray()
+          .catch(ignoreNotExisting)
+      )
+    );
     return {
-      operations_size_bytes: Number(operations_aggregate[0].storageStats.size),
-      parameters_size_bytes: Number(parameters_aggregate[0].storageStats.size),
+      operations_size_bytes:
+        Number(operations_aggregate[0].storageStats.size) +
+        v3_operation_aggregates.reduce((total, aggregate) => total + Number(aggregate[0].storageStats.size), 0),
+      parameters_size_bytes:
+        Number(parameters_aggregate[0].storageStats.size) +
+        v3_parameter_aggregates.reduce((total, aggregate) => total + Number(aggregate[0].storageStats.size), 0),
       replication_size_bytes:
-        Number(v1_replication_aggregate[0].storageStats.size) + Number(v3_replication_aggregate[0].storageStats.size)
+        Number(v1_source_record_aggregate[0]?.storageStats?.size ?? 0) +
+        source_record_aggregates.reduce((total, aggregate) => total + Number(aggregate[0]?.storageStats?.size ?? 0), 0)
     };
   }
 

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

@@ -0,0 +1,72 @@
+import { ServiceAssertionError } from '@powersync/lib-services-framework';
+import { BucketDataSource, ParameterIndexLookupCreator, SyncConfigWithErrors } from '@powersync/service-sync-rules';
+import { SyncRuleDocument } from './models.js';
+
+export type BucketDefinitionId = string;
+export type ParameterIndexId = string;
+
+export class BucketDefinitionMapping {
+  static fromSyncRules(doc: Pick<SyncRuleDocument, 'rule_mapping'>): BucketDefinitionMapping {
+    return new BucketDefinitionMapping(doc.rule_mapping?.definitions ?? {}, doc.rule_mapping?.parameter_indexes ?? {});
+  }
+
+  static fromParsedSyncRules(syncRules: SyncConfigWithErrors): BucketDefinitionMapping {
+    const definitionNames = syncRules.config.bucketDataSources.map((source) => source.uniqueName).sort();
+    const parameterKeys = syncRules.config.bucketParameterLookupSources
+      .map((source) => `${source.defaultLookupScope.lookupName}#${source.defaultLookupScope.queryId}`)
+      .sort();
+
+    const definitions: Record<string, BucketDefinitionId> = {};
+    const parameterLookups: Record<string, ParameterIndexId> = {};
+
+    for (const [index, uniqueName] of definitionNames.entries()) {
+      definitions[uniqueName] = (index + 1).toString(16);
+    }
+    for (const [index, key] of parameterKeys.entries()) {
+      parameterLookups[key] = (index + 1).toString(16);
+    }
+
+    return new BucketDefinitionMapping(definitions, parameterLookups);
+  }
+
+  constructor(
+    private definitions: Record<string, BucketDefinitionId> = {},
+    private parameterLookupMapping: Record<string, ParameterIndexId> = {}
+  ) {}
+
+  bucketSourceId(source: BucketDataSource): BucketDefinitionId {
+    const defId = this.definitions[source.uniqueName];
+    if (defId == null) {
+      throw new ServiceAssertionError(`No mapping found for bucket source ${source.uniqueName}`);
+    }
+    return defId;
+  }
+
+  allBucketDefinitionIds(): BucketDefinitionId[] {
+    return Object.values(this.definitions);
+  }
+
+  allParameterIndexIds(): ParameterIndexId[] {
+    return Object.values(this.parameterLookupMapping);
+  }
+
+  parameterLookupId(source: ParameterIndexLookupCreator): ParameterIndexId {
+    const key = this.parameterLookupKey(source.defaultLookupScope.lookupName, source.defaultLookupScope.queryId);
+    const defId = this.parameterLookupMapping[key];
+    if (defId == null) {
+      throw new ServiceAssertionError(`No mapping found for parameter lookup source ${key}`);
+    }
+    return defId;
+  }
+
+  private parameterLookupKey(lookupName: string, queryId: string) {
+    return `${lookupName}#${queryId}`;
+  }
+
+  serialize(): NonNullable<SyncRuleDocument['rule_mapping']> {
+    return {
+      definitions: { ...this.definitions },
+      parameter_indexes: { ...this.parameterLookupMapping }
+    };
+  }
+}