feat: add throttle and delay to connect (#478)

DominicGBauer · web-flow · commit 0f28fb3c10af · 2025-01-30T12:44:07.000+02:00
diff --git a/.changeset/slow-crews-watch.md b/.changeset/slow-crews-watch.md
@@ -0,0 +1,6 @@
+---
+'@powersync/react-native': minor
+'@powersync/common': minor
+---
+
+Add `retryDelayMs` and `crudUploadThrottleMs` to `connect` so that the values can be dynamically changed upon reconnecting.
diff --git a/packages/common/src/client/AbstractPowerSyncDatabase.ts b/packages/common/src/client/AbstractPowerSyncDatabase.ts
@@ -24,9 +24,12 @@ import { CrudEntry, CrudEntryJSON } from './sync/bucket/CrudEntry.js';
 import { CrudTransaction } from './sync/bucket/CrudTransaction.js';
 import {
   DEFAULT_CRUD_UPLOAD_THROTTLE_MS,
-  PowerSyncConnectionOptions,
+  type AdditionalConnectionOptions,
+  type PowerSyncConnectionOptions,
   StreamingSyncImplementation,
-  StreamingSyncImplementationListener
+  StreamingSyncImplementationListener,
+  DEFAULT_RETRY_DELAY_MS,
+  type RequiredAdditionalConnectionOptions
 } from './sync/stream/AbstractStreamingSyncImplementation.js';
 import { runOnSchemaChange } from './runOnSchemaChange.js';
 
@@ -35,21 +38,13 @@ export interface DisconnectAndClearOptions {
   clearLocal?: boolean;
 }
 
-export interface BasePowerSyncDatabaseOptions {
+export interface BasePowerSyncDatabaseOptions extends AdditionalConnectionOptions {
   /** Schema used for the local database. */
   schema: Schema;
-
   /**
-   * Delay for retrying sync streaming operations
-   * from the PowerSync backend after an error occurs.
+   * @deprecated Use {@link retryDelayMs} instead as this will be removed in future releases.
    */
   retryDelay?: number;
-  /**
-   * Backend Connector CRUD operations are throttled
-   * to occur at most every `crudUploadThrottleMs`
-   * milliseconds.
-   */
-  crudUploadThrottleMs?: number;
   logger?: ILogger;
 }
 
@@ -129,7 +124,7 @@ export const DEFAULT_POWERSYNC_CLOSE_OPTIONS: PowerSyncCloseOptions = {
 export const DEFAULT_WATCH_THROTTLE_MS = 30;
 
 export const DEFAULT_POWERSYNC_DB_OPTIONS = {
-  retryDelay: 5000,
+  retryDelayMs: 5000,
   logger: Logger.get('PowerSyncDatabase'),
   crudUploadThrottleMs: DEFAULT_CRUD_UPLOAD_THROTTLE_MS
 };
@@ -243,7 +238,8 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver<PowerSyncDB
   protected abstract openDBAdapter(options: PowerSyncDatabaseOptionsWithSettings): DBAdapter;
 
   protected abstract generateSyncStreamImplementation(
-    connector: PowerSyncBackendConnector
+    connector: PowerSyncBackendConnector,
+    options: RequiredAdditionalConnectionOptions
   ): StreamingSyncImplementation;
 
   protected abstract generateBucketStorageAdapter(): BucketStorageAdapter;
@@ -376,6 +372,15 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver<PowerSyncDB
     return this.waitForReady();
   }
 
+  // Use the options passed in during connect, or fallback to the options set during database creation or fallback to the default options
+  resolvedConnectionOptions(options?: PowerSyncConnectionOptions): RequiredAdditionalConnectionOptions {
+    return {
+      retryDelayMs: options?.retryDelayMs ?? this.options.retryDelayMs ?? this.options.retryDelay ?? DEFAULT_RETRY_DELAY_MS,
+      crudUploadThrottleMs:
+        options?.crudUploadThrottleMs ?? this.options.crudUploadThrottleMs ?? DEFAULT_CRUD_UPLOAD_THROTTLE_MS
+    };
+  }
+
   /**
    * Connects to stream of events from the PowerSync instance.
    */
@@ -388,7 +393,12 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver<PowerSyncDB
       throw new Error('Cannot connect using a closed client');
     }
 
-    this.syncStreamImplementation = this.generateSyncStreamImplementation(connector);
+    const { retryDelayMs, crudUploadThrottleMs } = this.resolvedConnectionOptions(options);
+
+    this.syncStreamImplementation = this.generateSyncStreamImplementation(connector, {
+      retryDelayMs,
+      crudUploadThrottleMs,
+    });
     this.syncStatusListenerDisposer = this.syncStreamImplementation.registerListener({
       statusChanged: (status) => {
         this.currentStatus = new SyncStatus({
diff --git a/packages/common/src/client/sync/stream/AbstractStreamingSyncImplementation.ts b/packages/common/src/client/sync/stream/AbstractStreamingSyncImplementation.ts
@@ -37,18 +37,16 @@ export interface LockOptions<T> {
   signal?: AbortSignal;
 }
 
-export interface AbstractStreamingSyncImplementationOptions {
+export interface AbstractStreamingSyncImplementationOptions extends AdditionalConnectionOptions {
   adapter: BucketStorageAdapter;
   uploadCrud: () => Promise<void>;
-  crudUploadThrottleMs?: number;
   /**
    * An identifier for which PowerSync DB this sync implementation is
    * linked to. Most commonly DB name, but not restricted to DB name.
    */
   identifier?: string;
   logger?: ILogger;
   remote: AbstractRemote;
-  retryDelayMs?: number;
 }
 
 export interface StreamingSyncImplementationListener extends BaseListener {
@@ -67,7 +65,10 @@ export interface StreamingSyncImplementationListener extends BaseListener {
  * Configurable options to be used when connecting to the PowerSync
  * backend instance.
  */
-export interface PowerSyncConnectionOptions {
+export interface PowerSyncConnectionOptions extends BaseConnectionOptions, AdditionalConnectionOptions {}
+
+ /** @internal */
+export interface BaseConnectionOptions {
   /**
    * The connection method to use when streaming updates from
    * the PowerSync backend instance.
@@ -81,6 +82,25 @@ export interface PowerSyncConnectionOptions {
   params?: Record<string, StreamingSyncRequestParameterType>;
 }
 
+ /** @internal */
+export interface AdditionalConnectionOptions {
+  /**
+   * Delay for retrying sync streaming operations
+   * from the PowerSync backend after an error occurs.
+   */
+  retryDelayMs?: number;
+  /**
+   * Backend Connector CRUD operations are throttled
+   * to occur at most every `crudUploadThrottleMs`
+   * milliseconds.
+   */
+  crudUploadThrottleMs?: number;
+}
+
+
+/** @internal */
+export type RequiredAdditionalConnectionOptions = Required<AdditionalConnectionOptions>
+
 export interface StreamingSyncImplementation extends BaseObserver<StreamingSyncImplementationListener>, Disposable {
   /**
    * Connects to the sync service
@@ -102,14 +122,17 @@ export interface StreamingSyncImplementation extends BaseObserver<StreamingSyncI
 }
 
 export const DEFAULT_CRUD_UPLOAD_THROTTLE_MS = 1000;
+export const DEFAULT_RETRY_DELAY_MS = 5000;
 
 export const DEFAULT_STREAMING_SYNC_OPTIONS = {
-  retryDelayMs: 5000,
+  retryDelayMs: DEFAULT_RETRY_DELAY_MS,
   logger: Logger.get('PowerSyncStream'),
   crudUploadThrottleMs: DEFAULT_CRUD_UPLOAD_THROTTLE_MS
 };
 
-export const DEFAULT_STREAM_CONNECTION_OPTIONS: Required<PowerSyncConnectionOptions> = {
+export type RequiredPowerSyncConnectionOptions = Required<BaseConnectionOptions>;
+
+export const DEFAULT_STREAM_CONNECTION_OPTIONS: RequiredPowerSyncConnectionOptions = {
   connectionMethod: SyncStreamConnectionMethod.WEB_SOCKET,
   params: {}
 };
@@ -427,7 +450,7 @@ The next upload iteration will be delayed.`);
       type: LockType.SYNC,
       signal,
       callback: async () => {
-        const resolvedOptions: Required<PowerSyncConnectionOptions> = {
+        const resolvedOptions: RequiredPowerSyncConnectionOptions = {
           ...DEFAULT_STREAM_CONNECTION_OPTIONS,
           ...(options ?? {})
         };
diff --git a/packages/react-native/src/db/PowerSyncDatabase.ts b/packages/react-native/src/db/PowerSyncDatabase.ts
@@ -5,6 +5,7 @@ import {
   DBAdapter,
   PowerSyncBackendConnector,
   PowerSyncDatabaseOptionsWithSettings,
+  type RequiredAdditionalConnectionOptions,
   SqliteBucketStorage
 } from '@powersync/common';
 import { ReactNativeRemote } from '../sync/stream/ReactNativeRemote';
@@ -42,7 +43,8 @@ export class PowerSyncDatabase extends AbstractPowerSyncDatabase {
   }
 
   protected generateSyncStreamImplementation(
-    connector: PowerSyncBackendConnector
+    connector: PowerSyncBackendConnector,
+    options: RequiredAdditionalConnectionOptions
   ): AbstractStreamingSyncImplementation {
     const remote = new ReactNativeRemote(connector);
 
@@ -53,8 +55,8 @@ export class PowerSyncDatabase extends AbstractPowerSyncDatabase {
         await this.waitForReady();
         await connector.uploadData(this);
       },
-      retryDelayMs: this.options.retryDelay,
-      crudUploadThrottleMs: this.options.crudUploadThrottleMs,
+      retryDelayMs: options.retryDelayMs,
+      crudUploadThrottleMs: options.crudUploadThrottleMs,
       identifier: this.database.name
     });
   }
diff --git a/packages/web/src/db/PowerSyncDatabase.ts b/packages/web/src/db/PowerSyncDatabase.ts
@@ -3,6 +3,7 @@ import {
   type PowerSyncBackendConnector,
   type PowerSyncCloseOptions,
   type PowerSyncConnectionOptions,
+  type RequiredAdditionalConnectionOptions,
   AbstractPowerSyncDatabase,
   DBAdapter,
   DEFAULT_POWERSYNC_CLOSE_OPTIONS,
@@ -13,7 +14,7 @@ import {
   PowerSyncDatabaseOptionsWithOpenFactory,
   PowerSyncDatabaseOptionsWithSettings,
   SqliteBucketStorage,
-  StreamingSyncImplementation
+  StreamingSyncImplementation,
 } from '@powersync/common';
 import { Mutex } from 'async-mutex';
 import { getNavigatorLocks } from '../shared/navigator';
@@ -194,11 +195,15 @@ export class PowerSyncDatabase extends AbstractPowerSyncDatabase {
     return getNavigatorLocks().request(`lock-${this.database.name}`, cb);
   }
 
-  protected generateSyncStreamImplementation(connector: PowerSyncBackendConnector): StreamingSyncImplementation {
+  protected generateSyncStreamImplementation(
+    connector: PowerSyncBackendConnector,
+    options: RequiredAdditionalConnectionOptions
+  ): StreamingSyncImplementation {
     const remote = new WebRemote(connector);
-
     const syncOptions: WebStreamingSyncImplementationOptions = {
       ...(this.options as {}),
+      retryDelayMs: options.retryDelayMs,
+      crudUploadThrottleMs: options.crudUploadThrottleMs,
       flags: this.resolvedFlags,
       adapter: this.bucketStorageAdapter,
       remote,
diff --git a/packages/web/tests/src/db/AbstractPowerSyncDatabase.test.ts b/packages/web/tests/src/db/AbstractPowerSyncDatabase.test.ts
@@ -0,0 +1,151 @@
+import { describe, it, expect, vi } from 'vitest';
+import { AbstractPowerSyncDatabase, DEFAULT_RETRY_DELAY_MS, DEFAULT_CRUD_UPLOAD_THROTTLE_MS, BucketStorageAdapter, DBAdapter, PowerSyncBackendConnector, PowerSyncDatabaseOptionsWithSettings, RequiredAdditionalConnectionOptions, StreamingSyncImplementation } from '@powersync/common';
+import { testSchema } from '../../utils/testDb';
+
+class TestPowerSyncDatabase extends AbstractPowerSyncDatabase {
+  protected openDBAdapter(options: PowerSyncDatabaseOptionsWithSettings): DBAdapter {
+    return {} as any
+  }
+  protected generateSyncStreamImplementation(connector: PowerSyncBackendConnector, options: RequiredAdditionalConnectionOptions): StreamingSyncImplementation {
+    return undefined as any;
+  }
+  protected generateBucketStorageAdapter(): BucketStorageAdapter {
+    return {
+      init: vi.fn()
+    } as any
+  }
+  _initialize(): Promise<void> {
+    return Promise.resolve();
+  }
+
+  get database() {
+    return {
+      get: vi.fn().mockResolvedValue({ version: '0.3.0'}),
+      execute: vi.fn(),
+      refreshSchema: vi.fn(),
+    } as any
+  }
+  // Expose protected method for testing
+  public testResolvedConnectionOptions(options?: any) {
+    return this.resolvedConnectionOptions(options);
+  }
+}
+
+describe('AbstractPowerSyncDatabase', () => {
+  describe('resolvedConnectionOptions', () => {
+    it('should use connect options when provided', () => {
+      const db = new TestPowerSyncDatabase({
+        schema: testSchema,
+        database: { dbFilename: 'test.db' }
+      });
+
+      const result = db.testResolvedConnectionOptions({
+        retryDelayMs: 1000,
+        crudUploadThrottleMs: 2000
+      });
+
+      expect(result).toEqual({
+        retryDelayMs: 1000,
+        crudUploadThrottleMs: 2000
+      });
+    });
+
+    it('should fallback to constructor options when connect options not provided', () => {
+      const db = new TestPowerSyncDatabase({
+        schema: testSchema,
+        database: { dbFilename: 'test.db' },
+        retryDelayMs: 3000,
+        crudUploadThrottleMs: 4000
+      });
+
+      const result = db.testResolvedConnectionOptions();
+
+      expect(result).toEqual({
+        retryDelayMs: 3000,
+        crudUploadThrottleMs: 4000
+      });
+    });
+
+    it('should convert retryDelay to retryDelayMs', () => {
+      const db = new TestPowerSyncDatabase({
+        schema: testSchema,
+        database: { dbFilename: 'test.db' },
+        retryDelay: 5000
+      });
+
+      const result = db.testResolvedConnectionOptions();
+
+      expect(result).toEqual({
+        retryDelayMs: 5000,
+        crudUploadThrottleMs: DEFAULT_CRUD_UPLOAD_THROTTLE_MS
+      });
+    });
+
+    it('should prioritize retryDelayMs over retryDelay in constructor options', () => {
+      const db = new TestPowerSyncDatabase({
+        schema: testSchema,
+        database: { dbFilename: 'test.db' },
+        retryDelay: 5000,
+        retryDelayMs: 6000
+      });
+
+      const result = db.testResolvedConnectionOptions();
+
+      expect(result).toEqual({
+        retryDelayMs: 6000,
+        crudUploadThrottleMs: DEFAULT_CRUD_UPLOAD_THROTTLE_MS
+      });
+    });
+
+    it('should prioritize connect options over constructor options', () => {
+      const db = new TestPowerSyncDatabase({
+        schema: testSchema,
+        database: { dbFilename: 'test.db' },
+        retryDelayMs: 5000,
+        crudUploadThrottleMs: 6000
+      });
+
+      const result = db.testResolvedConnectionOptions({
+        retryDelayMs: 7000,
+        crudUploadThrottleMs: 8000
+      });
+
+      expect(result).toEqual({
+        retryDelayMs: 7000,
+        crudUploadThrottleMs: 8000
+      });
+    });
+
+    it('should use default values when no options provided', () => {
+      const db = new TestPowerSyncDatabase({
+        schema: testSchema,
+        database: { dbFilename: 'test.db' }
+      });
+
+      const result = db.testResolvedConnectionOptions();
+
+      expect(result).toEqual({
+        retryDelayMs: DEFAULT_RETRY_DELAY_MS,
+        crudUploadThrottleMs: DEFAULT_CRUD_UPLOAD_THROTTLE_MS
+      });
+    });
+
+    it('should handle partial connect options', () => {
+      const db = new TestPowerSyncDatabase({
+        schema: testSchema,
+        database: { dbFilename: 'test.db' },
+        retryDelayMs: 5000,
+        crudUploadThrottleMs: 6000
+      });
+
+      const result = db.testResolvedConnectionOptions({
+        retryDelayMs: 7000
+      });
+
+      expect(result).toEqual({
+        retryDelayMs: 7000,
+        crudUploadThrottleMs: 6000
+      });
+    });
+  });
+});
diff --git a/packages/web/tests/src/db/PowersyncDatabase.test.ts b/packages/web/tests/src/db/PowersyncDatabase.test.ts
