Merge branch 'main' into fix/clearing-bucket-parameters

Author: rkistner

libs/lib-postgres/CHANGELOG.md

@@ -1,5 +1,12 @@
 # @powersync/lib-service-postgres
 
+## 0.4.1
+
+### Patch Changes
+
+- Updated dependencies [d1b83ce]
+  - @powersync/service-types@0.10.0
+
 ## 0.4.0
 
 ### Minor Changes

libs/lib-postgres/package.json

@@ -2,7 +2,7 @@
   "name": "@powersync/lib-service-postgres",
   "repository": "https://github.com/powersync-ja/powersync-service",
   "types": "dist/index.d.ts",
-  "version": "0.4.0",
+  "version": "0.4.1",
   "main": "dist/index.js",
   "license": "FSL-1.1-Apache-2.0",
   "type": "module",

modules/module-mongodb-storage/CHANGELOG.md

@@ -1,5 +1,17 @@
 # @powersync/service-module-mongodb-storage
 
+## 0.9.0
+
+### Minor Changes
+
+- d1b83ce: Refactored Metrics to use a MetricsEngine which is telemetry framework agnostic.
+
+### Patch Changes
+
+- Updated dependencies [d1b83ce]
+  - @powersync/service-core@1.11.0
+  - @powersync/service-types@0.10.0
+
 ## 0.8.2
 
 ### Patch Changes

modules/module-mongodb-storage/package.json

@@ -2,7 +2,7 @@
   "name": "@powersync/service-module-mongodb-storage",
   "repository": "https://github.com/powersync-ja/powersync-service",
   "types": "dist/index.d.ts",
-  "version": "0.8.2",
+  "version": "0.9.0",
   "main": "dist/index.js",
   "license": "FSL-1.1-Apache-2.0",
   "type": "module",

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

@@ -1,13 +1,12 @@
 import { container } from '@powersync/lib-services-framework';
-import { test_utils } from '@powersync/service-core-tests';
 import { beforeAll, beforeEach } from 'vitest';
+import { METRICS_HELPER } from '@powersync/service-core-tests';
 
 beforeAll(async () => {
   // Executes for every test file
   container.registerDefaults();
-  await test_utils.initMetrics();
 });
 
 beforeEach(async () => {
-  await test_utils.resetMetrics();
+  METRICS_HELPER.resetMetrics();
 });

modules/module-mongodb/CHANGELOG.md

@@ -1,5 +1,17 @@
 # @powersync/service-module-mongodb
 
+## 0.8.0
+
+### Minor Changes
+
+- d1b83ce: Refactored Metrics to use a MetricsEngine which is telemetry framework agnostic.
+
+### Patch Changes
+
+- Updated dependencies [d1b83ce]
+  - @powersync/service-core@1.11.0
+  - @powersync/service-types@0.10.0
+
 ## 0.7.5
 
 ### Patch Changes

modules/module-mongodb/package.json

@@ -2,7 +2,7 @@
   "name": "@powersync/service-module-mongodb",
   "repository": "https://github.com/powersync-ja/powersync-service",
   "types": "dist/index.d.ts",
-  "version": "0.7.5",
+  "version": "0.8.0",
   "main": "dist/index.js",
   "license": "FSL-1.1-Apache-2.0",
   "type": "module",

modules/module-mongodb/src/module/MongoModule.ts

@@ -37,11 +37,14 @@ export class MongoModule extends replication.ReplicationModule<types.MongoConnec
       id: this.getDefaultId(normalisedConfig.database ?? ''),
       syncRuleProvider: syncRuleProvider,
       storageEngine: context.storageEngine,
+      metricsEngine: context.metricsEngine,
       connectionFactory: connectionFactory,
       rateLimiter: new MongoErrorRateLimiter()
     });
   }
 
+  async onInitialized(context: system.ServiceContextContainer): Promise<void> {}
+
   /**
    * Combines base config with normalized connection settings
    */

modules/module-mongodb/src/replication/ChangeStream.ts

@@ -8,25 +8,20 @@ import {
   ReplicationAssertionError,
   ServiceError
 } from '@powersync/lib-services-framework';
-import {
-  BSON_DESERIALIZE_DATA_OPTIONS,
-  Metrics,
-  SaveOperationTag,
-  SourceEntityDescriptor,
-  SourceTable,
-  storage
-} from '@powersync/service-core';
+import { MetricsEngine, SaveOperationTag, SourceEntityDescriptor, SourceTable, storage } from '@powersync/service-core';
 import { DatabaseInputRow, SqliteRow, SqlSyncRules, TablePattern } from '@powersync/service-sync-rules';
 import { MongoLSN } from '../common/MongoLSN.js';
 import { PostImagesOption } from '../types/types.js';
 import { escapeRegExp } from '../utils.js';
 import { MongoManager } from './MongoManager.js';
 import { constructAfterRecord, createCheckpoint, getCacheIdentifier, getMongoRelation } from './MongoRelation.js';
 import { CHECKPOINTS_COLLECTION } from './replication-utils.js';
+import { ReplicationMetric } from '@powersync/service-types';
 
 export interface ChangeStreamOptions {
   connections: MongoManager;
   storage: storage.SyncRulesBucketStorage;
+  metrics: MetricsEngine;
   abort_signal: AbortSignal;
 }
 
@@ -59,13 +54,15 @@ export class ChangeStream {
   private connections: MongoManager;
   private readonly client: mongo.MongoClient;
   private readonly defaultDb: mongo.Db;
+  private readonly metrics: MetricsEngine;
 
   private abort_signal: AbortSignal;
 
   private relation_cache = new Map<string | number, storage.SourceTable>();
 
   constructor(options: ChangeStreamOptions) {
     this.storage = options.storage;
+    this.metrics = options.metrics;
     this.group_id = options.storage.group_id;
     this.connections = options.connections;
     this.client = this.connections.client;
@@ -318,7 +315,7 @@ export class ChangeStream {
       }
 
       at += docBatch.length;
-      Metrics.getInstance().rows_replicated_total.add(docBatch.length);
+      this.metrics.getCounter(ReplicationMetric.ROWS_REPLICATED).add(docBatch.length);
       const duration = performance.now() - lastBatch;
       lastBatch = performance.now();
       logger.info(
@@ -446,7 +443,7 @@ export class ChangeStream {
       return null;
     }
 
-    Metrics.getInstance().rows_replicated_total.add(1);
+    this.metrics.getCounter(ReplicationMetric.ROWS_REPLICATED).add(1);
     if (change.operationType == 'insert') {
       const baseRecord = constructAfterRecord(change.fullDocument);
       return await batch.save({

modules/module-mongodb/src/replication/ChangeStreamReplicationJob.ts

@@ -71,6 +71,7 @@ export class ChangeStreamReplicationJob extends replication.AbstractReplicationJ
       const stream = new ChangeStream({
         abort_signal: this.abortController.signal,
         storage: this.options.storage,
+        metrics: this.options.metrics,
         connections: connectionManager
       });
       await stream.replicate();