Merge branch 'main' into metrics-refactor

# Conflicts:
#	modules/module-mongodb/test/src/change_stream_utils.ts
#	modules/module-mysql/test/src/BinlogStreamUtils.ts
#	modules/module-postgres/test/src/wal_stream_utils.ts
#	packages/service-core-tests/src/tests/register-sync-tests.ts
#	packages/service-core/src/routes/endpoints/socket-route.ts
#	packages/service-core/src/routes/endpoints/sync-stream.ts
#	packages/service-core/src/system/ServiceContext.ts

Author: Rentacookie

libs/lib-mongodb/CHANGELOG.md

@@ -1,5 +1,11 @@
 # @powersync/lib-service-mongodb
 
+## 0.5.0
+
+### Minor Changes
+
+- ba7baeb: Make some service limits configurable.
+
 ## 0.4.3
 
 ### Patch Changes

libs/lib-mongodb/package.json

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

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

@@ -29,7 +29,11 @@ export const MONGO_OPERATION_TIMEOUT_MS = 30_000;
  */
 export const MONGO_CLEAR_OPERATION_TIMEOUT_MS = 5_000;
 
-export function createMongoClient(config: BaseMongoConfigDecoded) {
+export interface MongoConnectionOptions {
+  maxPoolSize: number;
+}
+
+export function createMongoClient(config: BaseMongoConfigDecoded, options?: MongoConnectionOptions) {
   const normalized = normalizeMongoConfig(config);
   return new mongo.MongoClient(normalized.uri, {
     auth: {
@@ -48,7 +52,7 @@ export function createMongoClient(config: BaseMongoConfigDecoded) {
     // Avoid too many connections:
     // 1. It can overwhelm the source database.
     // 2. Processing too many queries in parallel can cause the process to run out of memory.
-    maxPoolSize: 8,
+    maxPoolSize: options?.maxPoolSize ?? 8,
 
     maxConnecting: 3,
     maxIdleTimeMS: 60_000

libs/lib-postgres/CHANGELOG.md

@@ -1,5 +1,17 @@
 # @powersync/lib-service-postgres
 
+## 0.4.0
+
+### Minor Changes
+
+- 698467c: Use bigint everywhere internally for OpId.
+- ba7baeb: Make some service limits configurable.
+
+### Patch Changes
+
+- Updated dependencies [ba7baeb]
+  - @powersync/service-types@0.9.0
+
 ## 0.3.3
 
 ### Patch 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.3.3",
+  "version": "0.4.0",
   "main": "dist/index.js",
   "license": "FSL-1.1-Apache-2.0",
   "type": "module",

libs/lib-postgres/src/db/connection/DatabaseClient.ts

@@ -42,7 +42,9 @@ export class DatabaseClient extends AbstractPostgresConnection<DatabaseClientLis
   constructor(protected options: DatabaseClientOptions) {
     super();
     this.closed = false;
-    this.pool = pgwire.connectPgWirePool(options.config);
+    this.pool = pgwire.connectPgWirePool(options.config, {
+      maxSize: options.config.max_pool_size
+    });
     this.connections = Array.from({ length: TRANSACTION_CONNECTION_COUNT }, (v, index) => {
       // Only listen to notifications on a single (the first) connection
       const notificationChannels = index == 0 ? options.notificationChannels : [];

libs/lib-postgres/src/types/types.ts

@@ -4,7 +4,9 @@ import * as service_types from '@powersync/service-types';
 import * as t from 'ts-codec';
 import * as urijs from 'uri-js';
 
-export interface NormalizedBasePostgresConnectionConfig extends jpgwire.NormalizedConnectionConfig {}
+export interface NormalizedBasePostgresConnectionConfig extends jpgwire.NormalizedConnectionConfig {
+  max_pool_size: number;
+}
 
 export const POSTGRES_CONNECTION_TYPE = 'postgresql' as const;
 
@@ -42,7 +44,9 @@ export const BasePostgresConnectionConfig = t.object({
   /**
    * Prefix for the slot name. Defaults to "powersync_"
    */
-  slot_name_prefix: t.string.optional()
+  slot_name_prefix: t.string.optional(),
+
+  max_pool_size: t.number.optional()
 });
 
 export type BasePostgresConnectionConfig = t.Encoded<typeof BasePostgresConnectionConfig>;
@@ -125,7 +129,9 @@ export function normalizeConnectionConfig(options: BasePostgresConnectionConfigD
     lookup,
 
     client_certificate: options.client_certificate ?? undefined,
-    client_private_key: options.client_private_key ?? undefined
+    client_private_key: options.client_private_key ?? undefined,
+
+    max_pool_size: options.max_pool_size ?? 8
   } satisfies NormalizedBasePostgresConnectionConfig;
 }
 

modules/module-mongodb-storage/CHANGELOG.md

@@ -1,5 +1,22 @@
 # @powersync/service-module-mongodb-storage
 
+## 0.7.0
+
+### Minor Changes
+
+- 698467c: Use bigint everywhere internally for OpId.
+- ba7baeb: Make some service limits configurable.
+
+### Patch Changes
+
+- Updated dependencies [0298720]
+- Updated dependencies [698467c]
+- Updated dependencies [ba7baeb]
+  - @powersync/service-sync-rules@0.24.1
+  - @powersync/service-core@1.8.0
+  - @powersync/lib-service-mongodb@0.5.0
+  - @powersync/service-types@0.9.0
+
 ## 0.6.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.6.2",
+  "version": "0.7.0",
   "main": "dist/index.js",
   "license": "FSL-1.1-Apache-2.0",
   "type": "module",

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

@@ -11,7 +11,7 @@ import {
   ReplicationAssertionError,
   ServiceError
 } from '@powersync/lib-services-framework';
-import { deserializeBson, SaveOperationTag, storage, utils } from '@powersync/service-core';
+import { deserializeBson, InternalOpId, SaveOperationTag, storage, utils } from '@powersync/service-core';
 import * as timers from 'node:timers/promises';
 import { PowerSyncMongo } from './db.js';
 import { CurrentBucket, CurrentDataDocument, SourceKey, SyncRuleDocument } from './models.js';
@@ -39,7 +39,7 @@ export interface MongoBucketBatchOptions {
   groupId: number;
   slotName: string;
   lastCheckpointLsn: string | null;
-  keepaliveOp: string | null;
+  keepaliveOp: InternalOpId | null;
   noCheckpointBeforeLsn: string;
   storeCurrentData: boolean;
   /**
@@ -77,12 +77,12 @@ export class MongoBucketBatch
 
   private no_checkpoint_before_lsn: string;
 
-  private persisted_op: bigint | null = null;
+  private persisted_op: InternalOpId | null = null;
 
   /**
    * For tests only - not for persistence logic.
    */
-  public last_flushed_op: bigint | null = null;
+  public last_flushed_op: InternalOpId | null = null;
 
   constructor(options: MongoBucketBatchOptions) {
     super();
@@ -98,9 +98,7 @@ export class MongoBucketBatch
     this.skipExistingRows = options.skipExistingRows;
     this.batch = new OperationBatch();
 
-    if (options.keepaliveOp) {
-      this.persisted_op = BigInt(options.keepaliveOp);
-    }
+    this.persisted_op = options.keepaliveOp ?? null;
   }
 
   addCustomWriteCheckpoint(checkpoint: storage.BatchedCustomWriteCheckpointOptions): void {
@@ -135,7 +133,7 @@ export class MongoBucketBatch
       return null;
     }
 
-    let last_op: bigint | null = null;
+    let last_op: InternalOpId | null = null;
     let resumeBatch: OperationBatch | null = null;
 
     await this.withReplicationTransaction(`Flushing ${batch.length} ops`, async (session, opSeq) => {
@@ -153,7 +151,7 @@ export class MongoBucketBatch
 
     this.persisted_op = last_op;
     this.last_flushed_op = last_op;
-    return { flushed_op: String(last_op) };
+    return { flushed_op: last_op };
   }
 
   private async replicateBatch(
@@ -776,22 +774,23 @@ export class MongoBucketBatch
   async truncate(sourceTables: storage.SourceTable[]): Promise<storage.FlushedResult | null> {
     await this.flush();
 
-    let last_op: bigint | null = null;
+    let last_op: InternalOpId | null = null;
     for (let table of sourceTables) {
       last_op = await this.truncateSingle(table);
     }
 
     if (last_op) {
       this.persisted_op = last_op;
+      return {
+        flushed_op: last_op
+      };
+    } else {
+      return null;
     }
-
-    return {
-      flushed_op: String(last_op!)
-    };
   }
 
-  async truncateSingle(sourceTable: storage.SourceTable): Promise<bigint> {
-    let last_op: bigint | null = null;
+  async truncateSingle(sourceTable: storage.SourceTable): Promise<InternalOpId> {
+    let last_op: InternalOpId | null = null;
 
     // To avoid too large transactions, we limit the amount of data we delete per transaction.
     // Since we don't use the record data here, we don't have explicit size limits per batch.