Use BigInt for OpId internally (#209)

* Use bigint internally for op ids.

* Fix some tests.

* Fix checksum query issue.

* Fix some MySQL tests, and a race condition for aborting streaming.

* Add changeset.

* Fix merge conflict.

Author: rkistner

.changeset/ninety-dancers-agree.md

@@ -0,0 +1,12 @@
+---
+'@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-mysql': minor
+'@powersync/lib-service-postgres': minor
+---
+
+Use bigint everywhere internally for OpId.

.changeset/rotten-pianos-film.md

@@ -0,0 +1,5 @@
+---
+'@powersync/service-module-mysql': patch
+---
+
+Fix race condition when stopping replication immediately after starting it.

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.

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

@@ -1,6 +1,6 @@
 import { mongo } from '@powersync/lib-service-mongodb';
 import { logger, ReplicationAssertionError } from '@powersync/lib-services-framework';
-import { storage, utils } from '@powersync/service-core';
+import { InternalOpId, storage, utils } from '@powersync/service-core';
 
 import { PowerSyncMongo } from './db.js';
 import { BucketDataDocument, BucketDataKey } from './models.js';
@@ -12,7 +12,7 @@ interface CurrentBucketState {
   /**
    * Rows seen in the bucket, with the last op_id of each.
    */
-  seen: Map<string, bigint>;
+  seen: Map<string, InternalOpId>;
   /**
    * Estimated memory usage of the seen Map.
    */
@@ -21,7 +21,7 @@ interface CurrentBucketState {
   /**
    * Last (lowest) seen op_id that is not a PUT.
    */
-  lastNotPut: bigint | null;
+  lastNotPut: InternalOpId | null;
 
   /**
    * Number of REMOVE/MOVE operations seen since lastNotPut.
@@ -274,7 +274,7 @@ export class MongoCompactor {
    * @param bucket bucket name
    * @param op op_id of the last non-PUT operation, which will be converted to CLEAR.
    */
-  private async clearBucket(bucket: string, op: bigint) {
+  private async clearBucket(bucket: string, op: InternalOpId) {
     const opFilter = {
       _id: {
         $gte: {

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

@@ -12,6 +12,9 @@ import {
   CHECKPOINT_INVALIDATE_ALL,
   CheckpointChanges,
   GetCheckpointChangesOptions,
+  InternalOpId,
+  internalToExternalOpId,
+  ProtocolOpId,
   ReplicationCheckpoint,
   SourceTable,
   storage,
@@ -119,7 +122,7 @@ export class MongoSyncBucketStorage
       }
     );
     return {
-      checkpoint: utils.timestampToOpId(doc?.last_checkpoint ?? 0n),
+      checkpoint: doc?.last_checkpoint ?? 0n,
       lsn: doc?.last_checkpoint_lsn ?? null
     };
   }
@@ -143,7 +146,7 @@ export class MongoSyncBucketStorage
       slotName: this.slot_name,
       lastCheckpointLsn: checkpoint_lsn,
       noCheckpointBeforeLsn: doc?.no_checkpoint_before ?? options.zeroLSN,
-      keepaliveOp: doc?.keepalive_op ?? null,
+      keepaliveOp: doc?.keepalive_op ? BigInt(doc.keepalive_op) : null,
       storeCurrentData: options.storeCurrentData,
       skipExistingRows: options.skipExistingRows ?? false
     });
@@ -152,7 +155,7 @@ export class MongoSyncBucketStorage
     await callback(batch);
     await batch.flush();
     if (batch.last_flushed_op) {
-      return { flushed_op: String(batch.last_flushed_op) };
+      return { flushed_op: batch.last_flushed_op };
     } else {
       return null;
     }
@@ -249,7 +252,7 @@ export class MongoSyncBucketStorage
     return result!;
   }
 
-  async getParameterSets(checkpoint: utils.OpId, lookups: SqliteJsonValue[][]): Promise<SqliteJsonRow[]> {
+  async getParameterSets(checkpoint: utils.InternalOpId, lookups: SqliteJsonValue[][]): Promise<SqliteJsonRow[]> {
     const lookupFilter = lookups.map((lookup) => {
       return storage.serializeLookup(lookup);
     });
@@ -259,7 +262,7 @@ export class MongoSyncBucketStorage
           $match: {
             'key.g': this.group_id,
             lookup: { $in: lookupFilter },
-            _id: { $lte: BigInt(checkpoint) }
+            _id: { $lte: checkpoint }
           }
         },
         {
@@ -284,23 +287,26 @@ export class MongoSyncBucketStorage
   }
 
   async *getBucketDataBatch(
-    checkpoint: utils.OpId,
-    dataBuckets: Map<string, string>,
+    checkpoint: utils.InternalOpId,
+    dataBuckets: Map<string, InternalOpId>,
     options?: storage.BucketDataBatchOptions
   ): AsyncIterable<storage.SyncBucketDataBatch> {
     if (dataBuckets.size == 0) {
       return;
     }
     let filters: mongo.Filter<BucketDataDocument>[] = [];
 
-    const end = checkpoint ? BigInt(checkpoint) : new bson.MaxKey();
+    if (checkpoint == null) {
+      throw new ServiceAssertionError('checkpoint is null');
+    }
+    const end = checkpoint;
     for (let [name, start] of dataBuckets.entries()) {
       filters.push({
         _id: {
           $gt: {
             g: this.group_id,
             b: name,
-            o: BigInt(start)
+            o: start
           },
           $lte: {
             g: this.group_id,
@@ -347,15 +353,15 @@ export class MongoSyncBucketStorage
 
     let batchSize = 0;
     let currentBatch: utils.SyncBucketData | null = null;
-    let targetOp: bigint | null = null;
+    let targetOp: InternalOpId | null = null;
 
     // Ordered by _id, meaning buckets are grouped together
     for (let rawData of data) {
       const row = bson.deserialize(rawData, storage.BSON_DESERIALIZE_INTERNAL_OPTIONS) as BucketDataDocument;
       const bucket = row._id.b;
 
       if (currentBatch == null || currentBatch.bucket != bucket || batchSize >= sizeLimit) {
-        let start: string | undefined = undefined;
+        let start: ProtocolOpId | undefined = undefined;
         if (currentBatch != null) {
           if (currentBatch.bucket == bucket) {
             currentBatch.has_more = true;
@@ -369,9 +375,12 @@ export class MongoSyncBucketStorage
           targetOp = null;
         }
 
-        start ??= dataBuckets.get(bucket);
         if (start == null) {
-          throw new ServiceAssertionError(`data for unexpected bucket: ${bucket}`);
+          const startOpId = dataBuckets.get(bucket);
+          if (startOpId == null) {
+            throw new ServiceAssertionError(`data for unexpected bucket: ${bucket}`);
+          }
+          start = internalToExternalOpId(startOpId);
         }
         currentBatch = {
           bucket,
@@ -406,7 +415,7 @@ export class MongoSyncBucketStorage
     }
   }
 
-  async getChecksums(checkpoint: utils.OpId, buckets: string[]): Promise<utils.ChecksumMap> {
+  async getChecksums(checkpoint: utils.InternalOpId, buckets: string[]): Promise<utils.ChecksumMap> {
     return this.checksumCache.getChecksumMap(checkpoint, buckets);
   }
 
@@ -638,7 +647,7 @@ export class MongoSyncBucketStorage
 
   private makeActiveCheckpoint(doc: SyncRuleCheckpointState | null) {
     return {
-      checkpoint: utils.timestampToOpId(doc?.last_checkpoint ?? 0n),
+      checkpoint: doc?.last_checkpoint ?? 0n,
       lsn: doc?.last_checkpoint_lsn ?? null
     };
   }
@@ -755,7 +764,7 @@ export class MongoSyncBucketStorage
    */
   async *watchWriteCheckpoint(options: WatchWriteCheckpointOptions): AsyncIterable<storage.StorageCheckpointUpdate> {
     const { user_id, signal } = options;
-    let lastCheckpoint: utils.OpId | null = null;
+    let lastCheckpoint: utils.InternalOpId | null = null;
     let lastWriteCheckpoint: bigint | null = null;
 
     const iter = wrapWithAbort(this.sharedIter, signal);

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

@@ -4,7 +4,7 @@ import { EvaluatedParameters, EvaluatedRow } from '@powersync/service-sync-rules
 import * as bson from 'bson';
 
 import { logger } from '@powersync/lib-services-framework';
-import { storage, utils } from '@powersync/service-core';
+import { InternalOpId, storage, utils } from '@powersync/service-core';
 import { currentBucketKey } from './MongoBucketBatch.js';
 import { MongoIdSequence } from './MongoIdSequence.js';
 import { PowerSyncMongo } from './db.js';
@@ -52,7 +52,7 @@ export class PersistedBatch {
   /**
    * For debug logging only.
    */
-  debugLastOpId: bigint | null = null;
+  debugLastOpId: InternalOpId | null = null;
 
   /**
    * Very rough estimate of transaction size.

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

@@ -74,7 +74,7 @@ export async function readSingleBatch<T>(cursor: mongo.FindCursor<T>): Promise<{
 export function mapOpEntry(row: BucketDataDocument): utils.OplogEntry {
   if (row.op == 'PUT' || row.op == 'REMOVE') {
     return {
-      op_id: utils.timestampToOpId(row._id.o),
+      op_id: utils.internalToExternalOpId(row._id.o),
       op: row.op,
       object_type: row.table,
       object_id: row.row_id,
@@ -86,7 +86,7 @@ export function mapOpEntry(row: BucketDataDocument): utils.OplogEntry {
     // MOVE, CLEAR
 
     return {
-      op_id: utils.timestampToOpId(row._id.o),
+      op_id: utils.internalToExternalOpId(row._id.o),
       op: row.op,
       checksum: Number(row.checksum)
     };

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

@@ -74,7 +74,7 @@ describe('sync - mongodb', () => {
     const options: storage.BucketDataBatchOptions = {};
 
     const batch1 = await test_utils.fromAsync(
-      bucketStorage.getBucketDataBatch(checkpoint, new Map([['global[]', '0']]), options)
+      bucketStorage.getBucketDataBatch(checkpoint, new Map([['global[]', 0n]]), options)
     );
     expect(test_utils.getBatchData(batch1)).toEqual([
       { op_id: '1', op: 'PUT', object_id: 'test1', checksum: 2871785649 },
@@ -87,7 +87,7 @@ describe('sync - mongodb', () => {
     });
 
     const batch2 = await test_utils.fromAsync(
-      bucketStorage.getBucketDataBatch(checkpoint, new Map([['global[]', batch1[0].batch.next_after]]), options)
+      bucketStorage.getBucketDataBatch(checkpoint, new Map([['global[]', BigInt(batch1[0].batch.next_after)]]), options)
     );
     expect(test_utils.getBatchData(batch2)).toEqual([
       { op_id: '3', op: 'PUT', object_id: 'large2', checksum: 1607205872 }
@@ -99,7 +99,7 @@ describe('sync - mongodb', () => {
     });
 
     const batch3 = await test_utils.fromAsync(
-      bucketStorage.getBucketDataBatch(checkpoint, new Map([['global[]', batch2[0].batch.next_after]]), options)
+      bucketStorage.getBucketDataBatch(checkpoint, new Map([['global[]', BigInt(batch2[0].batch.next_after)]]), options)
     );
     expect(test_utils.getBatchData(batch3)).toEqual([
       { op_id: '4', op: 'PUT', object_id: 'test3', checksum: 1359888332 }