fix potential race conditions

Author: stevensJourney

.changeset/chilly-penguins-learn.md

@@ -0,0 +1,5 @@
+---
+'@powersync/common': patch
+---
+
+Fixed potential race conditions in WatchedQueries when updateSettings is called frequently.

packages/common/src/client/watched/processors/AbstractQueryProcessor.ts

@@ -83,21 +83,31 @@ export abstract class AbstractQueryProcessor<
    * Updates the underlying query.
    */
   async updateSettings(settings: Settings) {
+    // Abort any previous requests
     this.abortController.abort();
+
+    this.options.watchOptions = settings;
+    // Keep track of this controller's abort status
+    const abortController = new AbortController();
+    // Allow this to be aborted externally
+    this.abortController = abortController;
+
     await this.initialized;
 
+    // This may have been aborted while awaiting or if multiple calls to `updateSettings` were made
+    if (abortController.signal.aborted) {
+      return;
+    }
+
     if (!this.state.isFetching && this.reportFetching) {
       await this.updateState({
         isFetching: true
       });
     }
 
-    this.options.watchOptions = settings;
-
-    this.abortController = new AbortController();
     await this.runWithReporting(() =>
       this.linkQuery({
-        abortSignal: this.abortController.signal,
+        abortSignal: abortController.signal,
         settings
       })
     );

packages/web/tests/watch.test.ts

@@ -733,6 +733,47 @@ describe('Watch Tests', { sequential: true }, () => {
     expect(watch.state.data[0].make).equals('nottest');
   });
 
+  it('should allow updating queries', async () => {
+    const watch = powersync
+      .query<{ make: string }>({
+        sql: 'SELECT ? as result',
+        parameters: [0]
+      })
+      .watch({
+        reportFetching: false
+      });
+
+    let states: WatchedQueryState<any>[] = [];
+
+    // Keep track of all the states which have been updated
+    const dispose = watch.registerListener({
+      onStateChange: (state) => {
+        states.push(state);
+      }
+    });
+
+    // Spam the updateSettings
+    let updatePromises = Array.from({ length: 100 }).map(async (_, index) =>
+      watch.updateSettings({
+        query: new GetAllQuery({
+          sql: 'SELECT ? as result',
+          parameters: [index + 1]
+        })
+      })
+    );
+
+    await Promise.all(updatePromises);
+
+    await vi.waitFor(
+      () => {
+        expect(states[states.length - 1].isFetching).false;
+        expect(states[states.length - 1].data[0].result).eq(100);
+      },
+      { timeout: 3000 }
+    );
+    dispose();
+  });
+
   it('should report differential query results', async () => {
     const watch = powersync
       .query({