Add sync tests

Author: simolus3

packages/common/src/client/sync/bucket/SqliteBucketStorage.ts

@@ -208,7 +208,9 @@ export class SqliteBucketStorage extends BaseObserver<BucketStorageListener> imp
       if (result == 1) {
         if (priority == null) {
           const bucketToCount = Object.fromEntries(checkpoint.buckets.map((b) => [b.bucket, b.count]));
-          await tx.execute('UPDATE ps_buckets SET count_since_last = 0, count_at_last = ?1->name WHERE ?1->name IS NOT NULL', [JSON.stringify(bucketToCount)]);
+          // The two parameters could be replaced with one, but: https://github.com/powersync-ja/better-sqlite3/pull/6
+          const jsonBucketCount = JSON.stringify(bucketToCount);
+          await tx.execute('UPDATE ps_buckets SET count_since_last = 0, count_at_last = ?->name WHERE ?->name IS NOT NULL', [jsonBucketCount, jsonBucketCount]);
         }
 
         return true;

packages/common/src/client/sync/stream/AbstractStreamingSyncImplementation.ts

@@ -754,7 +754,7 @@ The next upload iteration will be delayed.`);
         // will use by default.
         priority: bucket.priority ?? 3,
         atLast: savedProgress?.atLast ?? 0,
-        sinceLast: savedProgress.sinceLast ?? 0,
+        sinceLast: savedProgress?.sinceLast ?? 0,
         targetCount: bucket.count ?? 0,
       };
     }
@@ -780,7 +780,7 @@ The next upload iteration will be delayed.`);
     });
 
     if (!this.syncStatus.isEqual(updatedStatus)) {
-      this.syncStatusOptions = options;
+      Object.assign(this.syncStatusOptions, options);
       this.syncStatus = updatedStatus;
       // Only trigger this is there was a change
       this.iterateListeners((cb) => cb.statusChanged?.(updatedStatus));

packages/common/src/client/sync/stream/streaming-sync-types.ts

@@ -102,7 +102,7 @@ export interface StreamingSyncCheckpointDiff {
     last_op_id: OpId;
     updated_buckets: BucketChecksum[];
     removed_buckets: string[];
-    write_checkpoint: string;
+    write_checkpoint?: string;
   };
 }
 

packages/node/package.json

@@ -57,7 +57,6 @@
     "@powersync/drizzle-driver": "workspace:*",
     "@types/async-lock": "^1.4.0",
     "drizzle-orm": "^0.35.2",
-    "fetch-mock": "^12.5.2",
     "rollup": "4.14.3",
     "typescript": "^5.5.3",
     "vitest": "^3.0.5"

packages/node/src/db/BetterSQLite3DBAdapter.ts

@@ -64,7 +64,9 @@ export class BetterSQLite3DBAdapter extends BaseObserver<DBAdapterListener> impl
       }
 
       if (!directoryExists) {
-        throw new Error(`The dbLocation directory at "${this.options.dbLocation}" does not exist. Please create it before opening the PowerSync database!`);
+        throw new Error(
+          `The dbLocation directory at "${this.options.dbLocation}" does not exist. Please create it before opening the PowerSync database!`
+        );
       }
 
       dbFilePath = path.join(this.options.dbLocation, dbFilePath);

packages/node/src/db/PowerSyncDatabase.ts

@@ -22,10 +22,10 @@ export type NodePowerSyncDatabaseOptions = PowerSyncDatabaseOptions & {
   database: DBAdapter | SQLOpenFactory | NodeSQLOpenOptions;
   /**
    * Options to override how the SDK will connect to the sync service.
-   * 
+   *
    * This option is intended to be used for internal tests.
    */
-  remoteOptions?: Partial<AbstractRemoteOptions>,
+  remoteOptions?: Partial<AbstractRemoteOptions>;
 };
 
 /**
@@ -68,7 +68,7 @@ export class PowerSyncDatabase extends AbstractPowerSyncDatabase {
     const remote = new NodeRemote(
       connector,
       DEFAULT_REMOTE_LOGGER,
-      (this.options as NodePowerSyncDatabaseOptions).remoteOptions,
+      (this.options as NodePowerSyncDatabaseOptions).remoteOptions
     );
 
     return new NodeStreamingSyncImplementation({

packages/node/src/db/RemoteConnection.ts

@@ -19,18 +19,43 @@ export class RemoteConnection implements LockContext {
     this.database = database;
   }
 
-  async executeBatch(query: string, params: any[][] = []): Promise<QueryResult> {
-    const result = await this.database.executeBatch(query, params ?? []);
-    return RemoteConnection.wrapQueryResult(result);
+  /**
+   * Runs the inner function, but appends the stack trace where this function was called. This is useful for workers
+   * because stack traces from worker errors are otherwise unrelated to the application issue that has caused them.
+   */
+  private async recoverTrace<T>(inner: () => Promise<T>): Promise<T> {
+    const trace = {};
+    Error.captureStackTrace(trace);
+
+    try {
+      return await inner();
+    } catch (e) {
+      if (e instanceof Error && e.stack) {
+        e.stack += (trace as any).stack;
+      }
+
+      throw e;
+    }
+  }
+
+  executeBatch(query: string, params: any[][] = []): Promise<QueryResult> {
+    return this.recoverTrace(async () => {
+      const result = await this.database.executeBatch(query, params ?? []);
+      return RemoteConnection.wrapQueryResult(result);
+    });
   }
 
-  async execute(query: string, params?: any[] | undefined): Promise<QueryResult> {
-    const result = await this.database.execute(query, params ?? []);
-    return RemoteConnection.wrapQueryResult(result);
+  execute(query: string, params?: any[] | undefined): Promise<QueryResult> {
+    return this.recoverTrace(async () => {
+      const result = await this.database.execute(query, params ?? []);
+      return RemoteConnection.wrapQueryResult(result);
+    });
   }
 
-  async executeRaw(query: string, params?: any[] | undefined): Promise<any[][]> {
-    return await this.database.executeRaw(query, params ?? []);
+  executeRaw(query: string, params?: any[] | undefined): Promise<any[][]> {
+    return this.recoverTrace(async () => {
+      return await this.database.executeRaw(query, params ?? []);
+    });
   }
 
   async getAll<T>(sql: string, parameters?: any[]): Promise<T[]> {

packages/node/src/sync/stream/NodeRemote.ts

@@ -29,7 +29,7 @@ export class NodeRemote extends AbstractRemote {
   ) {
     super(connector, logger, {
       fetchImplementation: options?.fetchImplementation ?? new NodeFetchProvider(),
-      ...(options ?? {}),
+      ...(options ?? {})
     });
   }
 

packages/node/tests/PowerSyncDatabase.test.ts

@@ -95,7 +95,7 @@ databaseTest('can watch tables', async ({ database }) => {
 tempDirectoryTest('throws error if target directory does not exist', async ({ tmpdir }) => {
   const directory = path.join(tmpdir, 'some', 'nested', 'location', 'that', 'does', 'not', 'exist');
 
-  expect(async () => {
+  await expect(async () => {
     const database = new PowerSyncDatabase({
       schema: AppSchema,
       database: {