Improve Connection Locks

.changeset/cyan-penguins-smell.md

@@ -0,0 +1,5 @@
+---
+'@powersync/react-native': minor
+---
+
+Fixed issue where iOS WebSockets could fail to reconnect after a connection issue.

.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/rollup.config.mjs

@@ -4,8 +4,11 @@ import json from '@rollup/plugin-json';
 import nodeResolve from '@rollup/plugin-node-resolve';
 import terser from '@rollup/plugin-terser';
 
+/**
+ * @returns {import('rollup').RollupOptions}
+ */
 export default (commandLineArgs) => {
-  const sourcemap = (commandLineArgs.sourceMap || 'true') == 'true';
+  const sourceMap = (commandLineArgs.sourceMap || 'true') == 'true';
 
   // Clears rollup CLI warning https://github.com/rollup/rollup/issues/2694
   delete commandLineArgs.sourceMap;
@@ -15,7 +18,7 @@ export default (commandLineArgs) => {
     output: {
       file: 'dist/bundle.mjs',
       format: 'esm',
-      sourcemap: sourcemap
+      sourcemap: sourceMap
     },
     plugins: [
       json(),
@@ -27,7 +30,7 @@ export default (commandLineArgs) => {
         // Used by can-ndjson-stream
         TextDecoder: ['text-encoding', 'TextDecoder']
       }),
-      terser()
+      terser({ sourceMap })
     ],
     // This makes life easier
     external: [

packages/common/src/client/AbstractPowerSyncDatabase.ts

@@ -9,13 +9,15 @@ 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 { ConnectionManager } from './ConnectionManager.js';
 import { SQLOpenFactory, SQLOpenOptions, isDBAdapter, isSQLOpenFactory, isSQLOpenOptions } from './SQLOpenFactory.js';
 import { PowerSyncBackendConnector } from './connection/PowerSyncBackendConnector.js';
 import { runOnSchemaChange } from './runOnSchemaChange.js';
@@ -32,7 +34,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. */
@@ -165,17 +166,22 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver<PowerSyncDB
    */
   currentStatus: SyncStatus;
 
-  syncStreamImplementation?: StreamingSyncImplementation;
   sdkVersion: string;
 
   protected bucketStorageAdapter: BucketStorageAdapter;
-  private syncStatusListenerDisposer?: () => void;
   protected _isReadyPromise: Promise<void>;
+  protected connectionManager: ConnectionManager;
+
+  get syncStreamImplementation() {
+    return this.connectionManager.syncStreamImplementation;
+  }
 
   protected _schema: Schema;
 
   private _database: DBAdapter;
 
+  protected runExclusiveMutex: Mutex;
+
   constructor(options: PowerSyncDatabaseOptionsWithDBAdapter);
   constructor(options: PowerSyncDatabaseOptionsWithOpenFactory);
   constructor(options: PowerSyncDatabaseOptionsWithSettings);
@@ -206,7 +212,33 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver<PowerSyncDB
     this._schema = schema;
     this.ready = false;
     this.sdkVersion = '';
+    this.runExclusiveMutex = new Mutex();
     // Start async init
+    this.connectionManager = new ConnectionManager({
+      createSyncImplementation: async (connector, options) => {
+        await this.waitForReady();
+
+        return this.runExclusive(async () => {
+          const sync = this.generateSyncStreamImplementation(connector, this.resolvedConnectionOptions(options));
+          const onDispose = sync.registerListener({
+            statusChanged: (status) => {
+              this.currentStatus = new SyncStatus({
+                ...status.toJSON(),
+                hasSynced: this.currentStatus?.hasSynced || !!status.lastSyncedAt
+              });
+              this.iterateListeners((cb) => cb.statusChanged?.(this.currentStatus));
+            }
+          });
+          await sync.waitForReady();
+
+          return {
+            sync,
+            onDispose
+          };
+        });
+      },
+      logger: this.logger
+    });
     this._isReadyPromise = this.initialize();
   }
 
@@ -425,34 +457,19 @@ 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 runExclusive<T>(callback: () => Promise<T>): Promise<T> {
+    return this.runExclusiveMutex.runExclusive(callback);
+  }
+
   /**
    * Connects to stream of events from the PowerSync instance.
    */
   async connect(connector: PowerSyncBackendConnector, options?: PowerSyncConnectionOptions) {
-    await this.waitForReady();
-
-    // close connection if one is open
-    await this.disconnect();
-    if (this.closed) {
-      throw new Error('Cannot connect using a closed client');
-    }
-
-    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));
-      }
-    });
-
-    await this.syncStreamImplementation.waitForReady();
-    this.syncStreamImplementation.triggerCrudUpload();
-    await this.syncStreamImplementation.connect(options);
+    return this.connectionManager.connect(connector, options);
   }
 
   /**
@@ -461,11 +478,7 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver<PowerSyncDB
    * Use {@link connect} to connect again.
    */
   async disconnect() {
-    await this.waitForReady();
-    await this.syncStreamImplementation?.disconnect();
-    this.syncStatusListenerDisposer?.();
-    await this.syncStreamImplementation?.dispose();
-    this.syncStreamImplementation = undefined;
+    return this.connectionManager.disconnect();
   }
 
   /**
@@ -512,7 +525,7 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver<PowerSyncDB
       await this.disconnect();
     }
 
-    await this.syncStreamImplementation?.dispose();
+    await this.connectionManager.close();
     await this.database.close();
     this.closed = true;
   }