Fix reconnect bugs

Author: stevensJourney

packages/common/src/client/ConnectionManager.ts

@@ -11,6 +11,10 @@ import {
  */
 export interface ConnectionManagerSyncImplementationResult {
   sync: StreamingSyncImplementation;
+  /**
+   * Additional cleanup function which is called after the sync stream implementation
+   * is disposed.
+   */
   onDispose: () => Promise<void> | void;
 }
 
@@ -65,7 +69,12 @@ export class ConnectionManager extends BaseObserver<ConnectionManagerListener> {
   protected pendingConnectionOptions: StoredConnectionOptions | null;
 
   syncStreamImplementation: StreamingSyncImplementation | null;
-  syncDisposer: (() => Promise<void> | void) | null;
+
+  /**
+   * Additional cleanup function which is called after the sync stream implementation
+   * is disposed.
+   */
+  protected syncDisposer: (() => Promise<void> | void) | null;
 
   constructor(protected options: ConnectionManagerOptions) {
     super();
@@ -82,6 +91,7 @@ export class ConnectionManager extends BaseObserver<ConnectionManagerListener> {
   }
 
   async close() {
+    await this.syncStreamImplementation?.dispose();
     await this.syncDisposer?.();
   }
 
@@ -101,7 +111,7 @@ export class ConnectionManager extends BaseObserver<ConnectionManagerListener> {
     // 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.
     // We should also disconnect if we have already completed a connection attempt.
-    if (!hadPendingOptions) {
+    if (!hadPendingOptions || this.syncStreamImplementation) {
       await this.disconnectInternal();
     }
 
@@ -145,11 +155,14 @@ export class ConnectionManager extends BaseObserver<ConnectionManagerListener> {
           return;
         }
 
+        if (this.disconnectingPromise) {
+          return;
+        }
+
         const { connector, options } = this.pendingConnectionOptions;
         appliedOptions = options;
 
         this.pendingConnectionOptions = null;
-
         const { sync, onDispose } = await this.options.createSyncImplementation(connector, options);
         this.iterateListeners((l) => l.syncStreamCreated?.(sync));
         this.syncStreamImplementation = sync;
@@ -195,17 +208,17 @@ export class ConnectionManager extends BaseObserver<ConnectionManagerListener> {
       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() {
+    // 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;
+
     // Keep reference to the sync stream implementation and disposer
     // The class members will be cleared before we trigger the disconnect
     // to prevent any further calls to the sync stream implementation.

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

@@ -449,6 +449,7 @@ The next upload iteration will be delayed.`);
         await this.streamingSyncIteration(nestedAbortController.signal, options);
         // Continue immediately, streamingSyncIteration will wait before completing if necessary.
       } catch (ex) {
+        let delay = true;
         /**
          * Either:
          *  - A network request failed with a failed connection or not OKAY response code.
@@ -459,6 +460,8 @@ The next upload iteration will be delayed.`);
          */
         if (ex instanceof AbortOperation) {
           this.logger.warn(ex);
+          delay = false;
+          // A disconnect was requested, we should not delay since there is no explicit retry
         } else {
           this.logger.error(ex);
         }
@@ -470,7 +473,9 @@ The next upload iteration will be delayed.`);
         });
 
         // On error, wait a little before retrying
-        await this.delayRetry();
+        if (delay) {
+          await this.delayRetry();
+        }
       } finally {
         if (!signal.aborted) {
           nestedAbortController.abort(new AbortOperation('Closing sync stream network requests before retry.'));

packages/web/src/worker/sync/SharedSyncImplementation.ts

@@ -30,6 +30,7 @@ import { AbstractSharedSyncClientProvider } from './AbstractSharedSyncClientProv
 import { BroadcastLogger } from './BroadcastLogger';
 
 /**
+ * @internal
  * Manual message events for shared sync clients
  */
 export enum SharedSyncClientEvent {
@@ -42,6 +43,9 @@ export enum SharedSyncClientEvent {
   CLOSE_ACK = 'close-ack'
 }
 
+/**
+ * @internal
+ */
 export type ManualSharedSyncPayload = {
   event: SharedSyncClientEvent;
   data: any; // TODO update in future
@@ -85,6 +89,7 @@ export type RemoteOperationAbortController = {
  * We provide this unused placeholder when connecting with the ConnectionManager.
  */
 const CONNECTOR_PLACEHOLDER = {} as PowerSyncBackendConnector;
+
 /**
  * @internal
  * Shared sync implementation which runs inside a shared webworker
@@ -133,23 +138,25 @@ export class SharedSyncImplementation
     this.broadCastLogger = new BroadcastLogger(this.ports);
 
     this.connectionManager = new ConnectionManager({
-      createSyncImplementation: async (connector, options) => {
-        await this.waitForReady();
-        if (!this.dbAdapter) {
-          await this.openInternalDB();
-        }
-
-        const sync = this.generateStreamingImplementation();
-        const onDispose = sync.registerListener({
-          statusChanged: (status) => {
-            this.updateAllStatuses(status.toJSON());
+      createSyncImplementation: async () => {
+        return this.portMutex.runExclusive(async () => {
+          await this.waitForReady();
+          if (!this.dbAdapter) {
+            await this.openInternalDB();
           }
-        });
 
-        return {
-          sync,
-          onDispose
-        };
+          const sync = await this.generateStreamingImplementation();
+          const onDispose = sync.registerListener({
+            statusChanged: (status) => {
+              this.updateAllStatuses(status.toJSON());
+            }
+          });
+
+          return {
+            sync,
+            onDispose
+          };
+        });
       },
       logger: this.logger
     });
@@ -190,15 +197,17 @@ export class SharedSyncImplementation
   async setParams(params: SharedSyncInitOptions) {
     await this.portMutex.runExclusive(async () => {
       if (this.syncParams) {
+        // Cannot modify already existing sync implementation params
+        // But we can ask for a DB adapter, if required, at this point.
+
         if (!this.dbAdapter) {
           await this.openInternalDB();
         }
-        // Cannot modify already existing sync implementation
         return;
       }
 
+      // First time setting params
       this.syncParams = params;
-
       if (params.streamOptions?.flags?.broadcastLogs) {
         this.logger = this.broadCastLogger;
       }
@@ -229,17 +238,12 @@ export class SharedSyncImplementation
    * connects.
    */
   async connect(options?: PowerSyncConnectionOptions) {
-    await this.portMutex.runExclusive(async () => {
-      // Keep track of the last connect options if we need to reconnect due to a lost client
-      this.lastConnectOptions = options;
-      return this.connectionManager.connect(CONNECTOR_PLACEHOLDER, options);
-    });
+    this.lastConnectOptions = options;
+    return this.connectionManager.connect(CONNECTOR_PLACEHOLDER, options);
   }
 
   async disconnect() {
-    await this.portMutex.runExclusive(async () => {
-      await this.connectionManager.disconnect();
-    });
+    return this.connectionManager.disconnect();
   }
 
   /**
@@ -266,11 +270,11 @@ export class SharedSyncImplementation
    * clients.
    */
   async removePort(port: MessagePort) {
-    return await this.portMutex.runExclusive(async () => {
+    const { trackedPort, shouldReconnect } = await this.portMutex.runExclusive(async () => {
       const index = this.ports.findIndex((p) => p.port == port);
       if (index < 0) {
         this.logger.warn(`Could not remove port ${port} since it is not present in active ports.`);
-        return;
+        return {};
       }
 
       const trackedPort = this.ports[index];
@@ -291,26 +295,35 @@ export class SharedSyncImplementation
 
       const shouldReconnect = !!this.connectionManager.syncStreamImplementation && this.ports.length > 0;
 
-      if (this.dbAdapter && this.dbAdapter == trackedPort.db) {
-        if (shouldReconnect) {
-          await this.connectionManager.disconnect();
-        }
+      return {
+        shouldReconnect,
+        trackedPort
+      };
+    });
 
-        // Clearing the adapter will result in a new one being opened in connect
-        this.dbAdapter = null;
+    if (!trackedPort) {
+      // We could not find the port to remove
+      return () => {};
+    }
 
-        if (shouldReconnect) {
-          await this.connectionManager.connect(CONNECTOR_PLACEHOLDER, this.lastConnectOptions);
-        }
+    if (this.dbAdapter && this.dbAdapter == trackedPort.db) {
+      if (shouldReconnect) {
+        await this.connectionManager.disconnect();
       }
 
-      if (trackedPort.db) {
-        await trackedPort.db.close();
+      // Clearing the adapter will result in a new one being opened in connect
+      this.dbAdapter = null;
+
+      if (shouldReconnect) {
+        await this.connectionManager.connect(CONNECTOR_PLACEHOLDER, this.lastConnectOptions);
       }
-      this.logger.debug(`Port ${port} removed from shared sync implementation.`);
-      // Release proxy
-      return () => trackedPort.clientProvider[Comlink.releaseProxy]();
-    });
+    }
+
+    if (trackedPort.db) {
+      await trackedPort.db.close();
+    }
+    // Release proxy
+    return () => trackedPort.clientProvider[Comlink.releaseProxy]();
   }
 
   triggerCrudUpload() {
@@ -351,7 +364,6 @@ export class SharedSyncImplementation
   protected generateStreamingImplementation() {
     // This should only be called after initialization has completed
     const syncParams = this.syncParams!;
-
     // Create a new StreamingSyncImplementation for each connect call. This is usually done is all SDKs.
     return new WebStreamingSyncImplementation({
       adapter: new SqliteBucketStorage(this.dbAdapter!, new Mutex(), this.logger),
@@ -454,7 +466,7 @@ export class SharedSyncImplementation
    * A function only used for unit tests which updates the internal
    * sync stream client and all tab client's sync status
    */
-  private _testUpdateAllStatuses(status: SyncStatusOptions) {
+  private async _testUpdateAllStatuses(status: SyncStatusOptions) {
     if (!this.connectionManager.syncStreamImplementation) {
       // This is just for testing purposes
       this.connectionManager.syncStreamImplementation = this.generateStreamingImplementation();