[Fix] - Possible race condition/handling uncaught exception in watch() (#120)

Co-authored-by: Christiaan Landman <[email protected]>

Author: Chriztiaan

.changeset/fresh-fishes-own.md

@@ -0,0 +1,5 @@
+---
+'@journeyapps/powersync-sdk-common': patch
+---
+
+Resolving tables for watch() before handling any results, eliminating a potential race condition between initial result and changes. Also handling a potential uncaught exception.

packages/powersync-sdk-common/src/client/AbstractPowerSyncDatabase.ts

@@ -476,8 +476,8 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver<PowerSyncDB
   }
 
   /**
-   * Execute a write query (INSERT/UPDATE/DELETE) multiple times with each parameter set 
-   * and optionally return results. 
+   * Execute a write query (INSERT/UPDATE/DELETE) multiple times with each parameter set
+   * and optionally return results.
    * This is faster than executing separately with each parameter set.
    */
   async executeBatch(sql: string, parameters?: any[][]) {
@@ -637,13 +637,24 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver<PowerSyncDB
 
     (async () => {
       try {
-        // Fetch initial data
-        onResult(await this.executeReadOnly(sql, parameters));
-
         const resolvedTables = await this.resolveTables(sql, parameters, options);
 
+        // Fetch initial data
+        const result = await this.executeReadOnly(sql, parameters);
+        onResult(result);
+
         this.onChangeWithCallback(
-          { onChange: async () => onResult(await this.executeReadOnly(sql, parameters)), onError },
+          {
+            onChange: async () => {
+              try {
+                const result = await this.executeReadOnly(sql, parameters);
+                onResult(result);
+              } catch (error) {
+                onError?.(error);
+              }
+            },
+            onError
+          },
           {
             ...(options ?? {}),
             tables: resolvedTables
@@ -663,11 +674,11 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver<PowerSyncDB
   watchWithAsyncGenerator(sql: string, parameters?: any[], options?: SQLWatchOptions): AsyncIterable<QueryResult> {
     return new EventIterator<QueryResult>((eventOptions) => {
       (async () => {
+        const resolvedTables = await this.resolveTables(sql, parameters, options);
+
         // Fetch initial data
         eventOptions.push(await this.executeReadOnly(sql, parameters));
 
-        const resolvedTables = await this.resolveTables(sql, parameters, options);
-
         for await (const event of this.onChangeWithAsyncGenerator({
           ...(options ?? {}),
           tables: resolvedTables

packages/powersync-sdk-web/tests/watch.test.ts

@@ -195,6 +195,9 @@ describe('Watch Tests', () => {
       }
     })();
 
+    // Ensures insert doesn't form part of initial result
+    await new Promise<void>((resolve) => setTimeout(resolve, throttleDuration));
+
     // Create the inserts as fast as possible
     await powersync.execute('INSERT INTO assets(id, make, customer_id) VALUES (uuid(), ?, ?)', ['test', uuid()]);
 
@@ -238,7 +241,9 @@ describe('Watch Tests', () => {
       }
     );
 
-    // Create the inserts as fast as possible
+    // Ensures insert doesn't form part of initial result
+    await new Promise<void>((resolve) => setTimeout(resolve, throttleDuration));
+
     await powersync.execute('INSERT INTO assets(id, make, customer_id) VALUES (uuid(), ?, ?)', ['test', uuid()]);
 
     await new Promise<void>((resolve) => setTimeout(resolve, throttleDuration * 2));