Improve Connection Locks

.changeset/empty-pants-give.md

@@ -0,0 +1,8 @@
+---
+'@powersync/react-native': minor
+'@powersync/common': minor
+'@powersync/node': minor
+'@powersync/web': minor
+---
+
+Improved behvaiour when connect is called multiple times in quick succession. Updating client parameters should now be more responsive.

packages/common/src/client/AbstractPowerSyncDatabase.ts

@@ -9,13 +9,14 @@ import {
   UpdateNotification,
   isBatchedUpdateNotification
 } from '../db/DBAdapter.js';
+import { FULL_SYNC_PRIORITY } from '../db/crud/SyncProgress.js';
 import { SyncPriorityStatus, SyncStatus } from '../db/crud/SyncStatus.js';
 import { UploadQueueStats } from '../db/crud/UploadQueueStatus.js';
 import { Schema } from '../db/schema/Schema.js';
 import { BaseObserver } from '../utils/BaseObserver.js';
 import { ControlledExecutor } from '../utils/ControlledExecutor.js';
-import { mutexRunExclusive } from '../utils/mutex.js';
 import { throttleTrailing } from '../utils/async.js';
+import { mutexRunExclusive } from '../utils/mutex.js';
 import { SQLOpenFactory, SQLOpenOptions, isDBAdapter, isSQLOpenFactory, isSQLOpenOptions } from './SQLOpenFactory.js';
 import { PowerSyncBackendConnector } from './connection/PowerSyncBackendConnector.js';
 import { runOnSchemaChange } from './runOnSchemaChange.js';
@@ -32,7 +33,6 @@ import {
   type PowerSyncConnectionOptions,
   type RequiredAdditionalConnectionOptions
 } from './sync/stream/AbstractStreamingSyncImplementation.js';
-import { FULL_SYNC_PRIORITY } from '../db/crud/SyncProgress.js';
 
 export interface DisconnectAndClearOptions {
   /** When set to false, data in local-only tables is preserved. */
@@ -112,6 +112,11 @@ export interface PowerSyncCloseOptions {
   disconnect?: boolean;
 }
 
+type StoredConnectionOptions = {
+  connector: PowerSyncBackendConnector;
+  options: PowerSyncConnectionOptions;
+};
+
 const POWERSYNC_TABLE_MATCH = /(^ps_data__|^ps_data_local__)/;
 
 const DEFAULT_DISCONNECT_CLEAR_OPTIONS: DisconnectAndClearOptions = {
@@ -176,6 +181,29 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver<PowerSyncDB
 
   private _database: DBAdapter;
 
+  /**
+   * Tracks active connection attempts
+   */
+  protected connectingPromise: Promise<void> | null;
+  /**
+   * Tracks actively instantiating a streaming sync implementation.
+   */
+  protected syncStreamInitPromise: Promise<void> | null;
+  /**
+   * Active disconnect operation. Calling disconnect multiple times
+   * will resolve to the same operation.
+   */
+  protected disconnectingPromise: Promise<void> | null;
+  /**
+   * Tracks the last parameters supplied to `connect` calls.
+   * Calling `connect` multiple times in succession will result in:
+   * - 1 pending connection operation which will be aborted.
+   * - updating the last set of parameters while waiting for the pending
+   *   attempt to be aborted
+   * - internally connecting with the last set of parameters
+   */
+  protected pendingConnectionOptions: StoredConnectionOptions | null;
+
   constructor(options: PowerSyncDatabaseOptionsWithDBAdapter);
   constructor(options: PowerSyncDatabaseOptionsWithOpenFactory);
   constructor(options: PowerSyncDatabaseOptionsWithSettings);
@@ -206,6 +234,9 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver<PowerSyncDB
     this._schema = schema;
     this.ready = false;
     this.sdkVersion = '';
+    this.connectingPromise = null;
+    this.syncStreamInitPromise = null;
+    this.pendingConnectionOptions = null;
     // Start async init
     this._isReadyPromise = this.initialize();
   }
@@ -425,34 +456,117 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver<PowerSyncDB
     };
   }
 
+  /**
+   * Locking mechanism for exclusively running critical portions of connect/disconnect operations.
+   * Locking here is mostly only important on web for multiple tab scenarios.
+   */
+  protected abstract runExclusive<T>(callback: () => Promise<T>): Promise<T>;
+
+  protected async connectInternal() {
+    let appliedOptions: PowerSyncConnectionOptions | null = null;
+
+    // This method ensures a disconnect before any connection attempt
+    await this.disconnectInternal();
+
+    /**
+     * This portion creates a sync implementation which can be racy when disconnecting or
+     * if multiple tabs on web are in use.
+     * This is protected in an exclusive lock.
+     * The promise tracks the creation which is used to synchronize disconnect attempts.
+     */
+    this.syncStreamInitPromise = this.runExclusive(async () => {
+      if (this.closed) {
+        throw new Error('Cannot connect using a closed client');
+      }
+
+      // Always await this if present since we will be populating a new sync implementation shortly
+      await this.disconnectingPromise;
+
+      if (!this.pendingConnectionOptions) {
+        // A disconnect could have cleared this.
+        return;
+      }
+      // get pending options and clear it in order for other connect attempts to queue other options
+      const { connector, options } = this.pendingConnectionOptions;
+      appliedOptions = options;
+      this.pendingConnectionOptions = null;
+
+      this.syncStreamImplementation = this.generateSyncStreamImplementation(
+        connector,
+        this.resolvedConnectionOptions(options)
+      );
+      this.syncStatusListenerDisposer = this.syncStreamImplementation.registerListener({
+        statusChanged: (status) => {
+          this.currentStatus = new SyncStatus({
+            ...status.toJSON(),
+            hasSynced: this.currentStatus?.hasSynced || !!status.lastSyncedAt
+          });
+          this.iterateListeners((cb) => cb.statusChanged?.(this.currentStatus));
+        }
+      });
+
+      await this.syncStreamImplementation.waitForReady();
+    });
+
+    await this.syncStreamInitPromise;
+    this.syncStreamInitPromise = null;
+
+    if (!appliedOptions) {
+      // A disconnect could have cleared the options which did not create a syncStreamImplementation
+      return;
+    }
+
+    // It might be possible that a disconnect triggered between the last check
+    // and this point. Awaiting here allows the sync stream to be cleared if disconnected.
+    await this.disconnectingPromise;
+
+    this.syncStreamImplementation?.triggerCrudUpload();
+    this.options.logger?.debug('Attempting to connect to PowerSync instance');
+    await this.syncStreamImplementation?.connect(appliedOptions!);
+  }
+
   /**
    * Connects to stream of events from the PowerSync instance.
    */
   async connect(connector: PowerSyncBackendConnector, options?: PowerSyncConnectionOptions) {
+    // Keep track if there were pending operations before this call
+    const hadPendingOptions = !!this.pendingConnectionOptions;
+
+    // Update pending options to the latest values
+    this.pendingConnectionOptions = {
+      connector,
+      options: options ?? {}
+    };
+
     await this.waitForReady();
 
-    // close connection if one is open
-    await this.disconnect();
-    if (this.closed) {
-      throw new Error('Cannot connect using a closed client');
+    // Disconnecting here provides aborting in progress connection attempts.
+    // The connectInternal method will clear pending options once it starts connecting (with the options).
+    // We only need to trigger a disconnect here if we have already reached the point of connecting.
+    // If we do already have pending options, a disconnect has already been performed.
+    // The connectInternal method also does a sanity disconnect to prevent straggler connections.
+    if (!hadPendingOptions) {
+      await this.disconnectInternal();
     }
 
-    const resolvedConnectOptions = this.resolvedConnectionOptions(options);
-
-    this.syncStreamImplementation = this.generateSyncStreamImplementation(connector, resolvedConnectOptions);
-    this.syncStatusListenerDisposer = this.syncStreamImplementation.registerListener({
-      statusChanged: (status) => {
-        this.currentStatus = new SyncStatus({
-          ...status.toJSON(),
-          hasSynced: this.currentStatus?.hasSynced || !!status.lastSyncedAt
-        });
-        this.iterateListeners((cb) => cb.statusChanged?.(this.currentStatus));
+    // Triggers a connect which checks if pending options are available after the connect completes.
+    // The completion can be for a successful, unsuccessful or aborted connection attempt.
+    // If pending options are available another connection will be triggered.
+    const checkConnection = async (): Promise<void> => {
+      if (this.pendingConnectionOptions) {
+        // Pending options have been placed while connecting.
+        // Need to reconnect.
+        this.connectingPromise = this.connectInternal().finally(checkConnection);
+        return this.connectingPromise;
+      } else {
+        // Clear the connecting promise, done.
+        this.connectingPromise = null;
+        return;
       }
-    });
+    };
 
-    await this.syncStreamImplementation.waitForReady();
-    this.syncStreamImplementation.triggerCrudUpload();
-    await this.syncStreamImplementation.connect(options);
+    this.connectingPromise ??= this.connectInternal().finally(checkConnection);
+    return this.connectingPromise;
   }
 
   /**
@@ -462,6 +576,28 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver<PowerSyncDB
    */
   async disconnect() {
     await this.waitForReady();
+    // This will help abort pending connects
+    this.pendingConnectionOptions = null;
+    await this.disconnectInternal();
+  }
+
+  protected async disconnectInternal() {
+    if (this.disconnectingPromise) {
+      // A disconnect is already in progress
+      return this.disconnectingPromise;
+    }
+
+    // Wait if a sync stream implementation is being created before closing it
+    // (syncStreamImplementation must be assigned before we can properly dispose it)
+    await this.syncStreamInitPromise;
+
+    this.disconnectingPromise = this.performDisconnect();
+
+    await this.disconnectingPromise;
+    this.disconnectingPromise = null;
+  }
+
+  protected async performDisconnect() {
     await this.syncStreamImplementation?.disconnect();
     this.syncStatusListenerDisposer?.();
     await this.syncStreamImplementation?.dispose();

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

@@ -1,13 +1,15 @@
 import Logger, { ILogger } from 'js-logger';
 
+import { InternalProgressInformation } from 'src/db/crud/SyncProgress.js';
+import { DataStream } from 'src/utils/DataStream.js';
 import { SyncStatus, SyncStatusOptions } from '../../../db/crud/SyncStatus.js';
 import { AbortOperation } from '../../../utils/AbortOperation.js';
 import { BaseListener, BaseObserver, Disposable } from '../../../utils/BaseObserver.js';
 import { onAbortPromise, throttleLeadingTrailing } from '../../../utils/async.js';
 import { BucketChecksum, BucketDescription, BucketStorageAdapter, Checkpoint } from '../bucket/BucketStorageAdapter.js';
 import { CrudEntry } from '../bucket/CrudEntry.js';
 import { SyncDataBucket } from '../bucket/SyncDataBucket.js';
-import { AbstractRemote, SyncStreamOptions, FetchStrategy } from './AbstractRemote.js';
+import { AbstractRemote, FetchStrategy, SyncStreamOptions } from './AbstractRemote.js';
 import {
   BucketRequest,
   StreamingSyncLine,
@@ -19,8 +21,6 @@ import {
   isStreamingSyncCheckpointPartiallyComplete,
   isStreamingSyncData
 } from './streaming-sync-types.js';
-import { DataStream } from 'src/utils/DataStream.js';
-import { InternalProgressInformation } from 'src/db/crud/SyncProgress.js';
 
 export enum LockType {
   CRUD = 'crud',
@@ -341,12 +341,13 @@ The next upload iteration will be delayed.`);
       await this.disconnect();
     }
 
-    this.abortController = new AbortController();
+    const controller = new AbortController();
+    this.abortController = controller;
     this.streamingSyncPromise = this.streamingSync(this.abortController.signal, options);
 
     // Return a promise that resolves when the connection status is updated
     return new Promise<void>((resolve) => {
-      const l = this.registerListener({
+      const disposer = this.registerListener({
         statusUpdated: (update) => {
           // This is triggered as soon as a connection is read from
           if (typeof update.connected == 'undefined') {
@@ -356,13 +357,15 @@ The next upload iteration will be delayed.`);
 
           if (update.connected == false) {
             /**
-             * This function does not reject if initial connect attempt failed
+             * This function does not reject if initial connect attempt failed.
+             * Connected can be false if the connection attempt was aborted or if the initial connection
+             * attempt failed.
              */
             this.logger.warn('Initial connect attempt did not successfully connect to server');
           }
 
+          disposer();
           resolve();
-          l();
         }
       });
     });
@@ -520,6 +523,10 @@ The next upload iteration will be delayed.`);
 
         const clientId = await this.options.adapter.getClientId();
 
+        if (signal.aborted) {
+          return;
+        }
+
         this.logger.debug('Requesting stream from server');
 
         const syncOptions: SyncStreamOptions = {

packages/node/src/db/PowerSyncDatabase.ts

@@ -5,22 +5,21 @@ import {
   AdditionalConnectionOptions,
   BucketStorageAdapter,
   DBAdapter,
-  DEFAULT_REMOTE_LOGGER,
   PowerSyncBackendConnector,
   PowerSyncConnectionOptions,
   PowerSyncDatabaseOptions,
   PowerSyncDatabaseOptionsWithSettings,
-  RequiredAdditionalConnectionOptions,
   SqliteBucketStorage,
   SQLOpenFactory
 } from '@powersync/common';
 
 import { NodeRemote } from '../sync/stream/NodeRemote.js';
 import { NodeStreamingSyncImplementation } from '../sync/stream/NodeStreamingSyncImplementation.js';
 
+import Lock from 'async-lock';
+import { Dispatcher } from 'undici';
 import { BetterSQLite3DBAdapter } from './BetterSQLite3DBAdapter.js';
 import { NodeSQLOpenOptions } from './options.js';
-import { Dispatcher } from 'undici';
 
 export type NodePowerSyncDatabaseOptions = PowerSyncDatabaseOptions & {
   database: DBAdapter | SQLOpenFactory | NodeSQLOpenOptions;
@@ -57,8 +56,11 @@ export type NodePowerSyncConnectionOptions = PowerSyncConnectionOptions & NodeAd
  * ```
  */
 export class PowerSyncDatabase extends AbstractPowerSyncDatabase {
+  protected lock: Lock;
+
   constructor(options: NodePowerSyncDatabaseOptions) {
     super(options);
+    this.lock = new Lock();
   }
 
   async _initialize(): Promise<void> {
@@ -83,6 +85,10 @@ export class PowerSyncDatabase extends AbstractPowerSyncDatabase {
     return super.connect(connector, options);
   }
 
+  protected async runExclusive<T>(callback: () => Promise<T>): Promise<T> {
+    return await this.lock.acquire(`lock-${this.database.name}`, callback);
+  }
+
   protected generateSyncStreamImplementation(
     connector: PowerSyncBackendConnector,
     options: NodeAdditionalConnectionOptions

packages/react-native/src/db/PowerSyncDatabase.ts

@@ -8,6 +8,7 @@ import {
   type RequiredAdditionalConnectionOptions,
   SqliteBucketStorage
 } from '@powersync/common';
+import Lock from 'async-lock';
 import { ReactNativeRemote } from '../sync/stream/ReactNativeRemote';
 import { ReactNativeStreamingSyncImplementation } from '../sync/stream/ReactNativeStreamingSyncImplementation';
 import { ReactNativeQuickSqliteOpenFactory } from './adapters/react-native-quick-sqlite/ReactNativeQuickSQLiteOpenFactory';
@@ -27,6 +28,8 @@ import { ReactNativeQuickSqliteOpenFactory } from './adapters/react-native-quick
  * ```
  */
 export class PowerSyncDatabase extends AbstractPowerSyncDatabase {
+  protected lock = new Lock();
+
   async _initialize(): Promise<void> {}
 
   /**
@@ -42,6 +45,10 @@ export class PowerSyncDatabase extends AbstractPowerSyncDatabase {
     return new SqliteBucketStorage(this.database, AbstractPowerSyncDatabase.transactionMutex);
   }
 
+  protected async runExclusive<T>(callback: () => Promise<T>): Promise<T> {
+    return await this.lock.acquire(`lock-${this.database.name}`, callback);
+  }
+
   protected generateSyncStreamImplementation(
     connector: PowerSyncBackendConnector,
     options: RequiredAdditionalConnectionOptions