Improve Connection Locks (#607)

Author: stevensJourney

.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 behaviour when connect is called multiple times in quick succession. Updating client parameters should now be more responsive.

.changeset/fuzzy-beers-occur.md

@@ -0,0 +1,7 @@
+---
+'@powersync/diagnostics-app': minor
+---
+
+- Added Sync error alert banner to all views.
+- Fix bug where clicking signOut would not disconnect from the PowerSync service.
+- Updated implementation to fetch sync errors from the SyncStatus.

.changeset/fuzzy-ties-double.md

@@ -0,0 +1,5 @@
+---
+'@powersync/common': patch
+---
+
+Fixed bug where changes in SyncStatus downloadError and uploadError might not be reported.

demos/react-native-supabase-todolist/library/powersync/system.ts

@@ -5,16 +5,16 @@ import React from 'react';
 import { SupabaseStorageAdapter } from '../storage/SupabaseStorageAdapter';
 
 import { type AttachmentRecord } from '@powersync/attachments';
+import { configureFts } from '../fts/fts_setup';
 import { KVStorage } from '../storage/KVStorage';
 import { AppConfig } from '../supabase/AppConfig';
 import { SupabaseConnector } from '../supabase/SupabaseConnector';
 import { AppSchema } from './AppSchema';
 import { PhotoAttachmentQueue } from './PhotoAttachmentQueue';
-import { configureFts } from '../fts/fts_setup';
 
 const logger = createBaseLogger();
 logger.useDefaults();
-logger.setLevel(LogLevel.INFO);
+logger.setLevel(LogLevel.DEBUG);
 
 export class System {
   kvStorage: KVStorage;
@@ -31,19 +31,22 @@ export class System {
       schema: AppSchema,
       database: {
         dbFilename: 'sqlite.db'
-      }
+      },
+      logger
     });
     /**
      * The snippet below uses OP-SQLite as the default database adapter.
      * You will have to uninstall `@journeyapps/react-native-quick-sqlite` and
      * install both `@powersync/op-sqlite` and `@op-engineering/op-sqlite` to use this.
      *
+     * ```typescript
      * import { OPSqliteOpenFactory } from '@powersync/op-sqlite'; // Add this import
      *
      * const factory = new OPSqliteOpenFactory({
-     * dbFilename: 'sqlite.db'
+     *  dbFilename: 'sqlite.db'
      * });
      * this.powersync = new PowerSyncDatabase({ database: factory, schema: AppSchema });
+     * ```
      */
 
     if (AppConfig.supabaseBucket) {

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;
   }