update: added a mechanism for throttling callback executions (#190)

Co-authored-by: Christiaan Landman <christiaanlandman@Christiaans-MacBook-Pro-2.local>

Author: Chriztiaan

.changeset/forty-avocados-yell.md

@@ -0,0 +1,5 @@
+---
+'@powersync/web': major
+---
+
+Bump version for Stable release.

.changeset/itchy-gifts-dance.md

@@ -0,0 +1,5 @@
+---
+'@powersync/common': minor
+---
+
+Added a mechanism for throttling watch callback executions.

packages/common/src/client/AbstractPowerSyncDatabase.ts

@@ -28,6 +28,7 @@ import {
   StreamingSyncImplementation,
   PowerSyncConnectionOptions
 } from './sync/stream/AbstractStreamingSyncImplementation';
+import { ControlledExecutor } from '../utils/ControlledExecutor';
 
 export interface DisconnectAndClearOptions {
   /** When set to false, data in local-only tables is preserved. */
@@ -74,7 +75,7 @@ export interface WatchHandler {
 }
 
 export interface WatchOnChangeHandler {
-  onChange: (event: WatchOnChangeEvent) => void;
+  onChange: (event: WatchOnChangeEvent) => Promise<void> | void;
   onError?: (error: Error) => void;
 }
 
@@ -845,12 +846,15 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver<PowerSyncDB
     const changedTables = new Set<string>();
     const throttleMs = resolvedOptions.throttleMs ?? DEFAULT_WATCH_THROTTLE_MS;
 
+    const executor = new ControlledExecutor(async (e: WatchOnChangeEvent) => {
+      await onChange(e);
+    });
+
     const flushTableUpdates = throttle(
       () =>
         this.handleTableChanges(changedTables, watchedTables, (intersection) => {
           if (resolvedOptions?.signal?.aborted) return;
-
-          onChange({ changedTables: intersection });
+          executor.schedule({ changedTables: intersection });
         }),
       throttleMs,
       { leading: false, trailing: true }
@@ -869,6 +873,7 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver<PowerSyncDB
     });
 
     resolvedOptions.signal?.addEventListener('abort', () => {
+      executor.dispose();
       dispose();
     });
 

packages/common/src/utils/ControlledExecutor.ts

@@ -0,0 +1,72 @@
+export interface ControlledExecutorOptions {
+  /**
+   * If throttling is enabled, it ensures only one task runs at a time,
+   * and only one additional task can be scheduled to run after the current task completes. The pending task will be overwritten by the latest task.
+   * Enabled by default.
+   */
+  throttleEnabled?: boolean;
+}
+
+export class ControlledExecutor<T> {
+  private task: (param: T) => Promise<void> | void;
+
+  /**
+   * Represents the currently running task, which could be a Promise or undefined if no task is running.
+   */
+  private runningTask: undefined | (Promise<void> | void);
+
+  private pendingTaskParam: T | undefined;
+
+  /**
+   * Flag to determine if throttling is enabled, which controls whether tasks are queued or run immediately.
+   */
+  private isThrottling: boolean;
+
+  private closed: boolean;
+
+  constructor(task: (param: T) => Promise<void> | void, options?: ControlledExecutorOptions) {
+    this.task = task;
+    const { throttleEnabled = true } = options ?? {};
+    this.isThrottling = throttleEnabled;
+    this.closed = false;
+  }
+
+  schedule(param: T) {
+    if (this.closed) {
+      return;
+    }
+    if (!this.isThrottling) {
+      this.task(param);
+      return;
+    }
+
+    if (this.runningTask) {
+      // set or replace the pending task param with latest one
+      this.pendingTaskParam = param;
+      return;
+    }
+
+    this.execute(param);
+  }
+
+  dispose() {
+    this.closed = true;
+
+    if (this.runningTask) {
+      this.runningTask = undefined;
+    }
+  }
+
+  private async execute(param: T) {
+    this.runningTask = this.task(param);
+    await this.runningTask;
+    this.runningTask = undefined;
+
+    if (this.pendingTaskParam) {
+      const pendingParam = this.pendingTaskParam;
+      this.pendingTaskParam = undefined;
+
+      this.execute(pendingParam);
+    }
+  }
+}

packages/web/README.md

@@ -10,10 +10,6 @@ This package (`packages/web`) is the PowerSync SDK for JavaScript Web clients. I
 
 See a summary of features [here](https://docs.powersync.com/client-sdk-references/js-web).
 
-## Beta Release
-
-The web SDK package is currently in a Beta release.
-
 # Installation
 
 ## Install Package

packages/web/tests/watch.test.ts

@@ -146,7 +146,9 @@ describe('Watch Tests', () => {
     const updatesCount = 5;
     let receivedUpdatesCount = 0;
 
-    const onUpdate = () => receivedUpdatesCount++;
+    const onUpdate = () => {
+      receivedUpdatesCount++;
+    };
 
     powersync.watch(
       'SELECT count() AS count FROM assets INNER JOIN customers ON customers.id = assets.customer_id',
@@ -216,7 +218,9 @@ describe('Watch Tests', () => {
     const assetsAbortController = new AbortController();
 
     let receivedAssetsUpdatesCount = 0;
-    const onWatchAssets = () => receivedAssetsUpdatesCount++;
+    const onWatchAssets = () => {
+      receivedAssetsUpdatesCount++;
+    };
 
     powersync.watch(
       'SELECT count() AS count FROM assets',
@@ -230,7 +234,9 @@ describe('Watch Tests', () => {
     const customersAbortController = new AbortController();
 
     let receivedCustomersUpdatesCount = 0;
-    const onWatchCustomers = () => receivedCustomersUpdatesCount++;
+    const onWatchCustomers = () => {
+      receivedCustomersUpdatesCount++;
+    };
 
     powersync.watch(
       'SELECT count() AS count FROM customers',
@@ -280,4 +286,39 @@ describe('Watch Tests', () => {
     await receivedError;
     expect(receivedErrorCount).equals(1);
   });
+
+  it('should throttle watch callback overflow', async () => {
+    const abortController = new AbortController();
+
+    const updatesCount = 25;
+
+    let receivedWithManagedOverflowCount = 0;
+    const onResultOverflow = () => {
+      receivedWithManagedOverflowCount++;
+    };
+
+    const overflowAbortController = new AbortController();
+    powersync.watch(
+      'SELECT count() AS count FROM assets',
+      [],
+      { onResult: onResultOverflow },
+      { signal: overflowAbortController.signal, throttleMs: 1 }
+    );
+
+    // Allows us to count the number of updates received without the initial trigger
+    await new Promise<void>((resolve) => setTimeout(resolve, 1 * throttleDuration));
+
+    // Perform a large number of inserts to trigger overflow
+    for (let i = 0; i < updatesCount; i++) {
+      powersync.execute('INSERT INTO assets(id, make, customer_id) VALUES (uuid(), ?, ?)', ['test', uuid()]);
+    }
+
+    await new Promise<void>((resolve) => setTimeout(resolve, 1 * throttleDuration));
+
+    abortController.abort();
+    overflowAbortController.abort();
+
+    // Initial onResult plus two left after overflow was throttled for onChange triggers
+    expect(receivedWithManagedOverflowCount).toBe(3);
+  });
 });