cleanup port assignments and init flow

Author: stevensJourney

packages/web/src/db/sync/SharedWebStreamingSyncImplementation.ts

@@ -2,7 +2,6 @@ import {
   PowerSyncConnectionOptions,
   PowerSyncCredentials,
   SubscribedStream,
-  SyncStatus,
   SyncStatusOptions
 } from '@powersync/common';
 import * as Comlink from 'comlink';
@@ -187,6 +186,8 @@ export class SharedWebStreamingSyncImplementation extends WebStreamingSyncImplem
      *    - The shared worker can then request the same lock. The client has been closed if the shared worker can acquire the lock.
      *    - Once the shared worker knows the client's lock, we can guarentee that the shared worker will detect if the client has been closed.
      *    - This makes the client safe for the shared worker to use.
+     *    - The client is only added to the SharedSyncImplementation once the lock has been registered.
+     *      This ensures we don't ever keep track of dead clients (tabs that closed before the lock was registered).
      *    - The client side lock is held until the client is disposed.
      *    - We resolve the top-level promise after the lock has been registered with the shared worker.
      * - The client sends the params to the shared worker after locks have been registered.
@@ -288,12 +289,4 @@ export class SharedWebStreamingSyncImplementation extends WebStreamingSyncImplem
   updateSubscriptions(subscriptions: SubscribedStream[]): void {
     this.syncManager.updateSubscriptions(subscriptions);
   }
-
-  /**
-   * Used in tests to force a connection states
-   */
-  private async _testUpdateStatus(status: SyncStatus) {
-    await this.isInitialized;
-    return this.syncManager._testUpdateAllStatuses(status.toJSON());
-  }
 }

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

@@ -76,10 +76,6 @@ export type WrappedSyncPort = {
   db?: DBAdapter;
   currentSubscriptions: SubscribedStream[];
   closeListeners: (() => void | Promise<void>)[];
-  /**
-   * If we can use Navigator locks to detect if the client has closed.
-   */
-  isProtectedFromClose: boolean;
   isClosing: boolean;
 };
 
@@ -176,14 +172,16 @@ export class SharedSyncImplementation extends BaseObserver<SharedSyncImplementat
   /**
    * Gets the last client port which we know is safe from unexpected closes.
    */
-  protected get lastWrappedPort(): WrappedSyncPort | undefined {
-    // Find the last port which is protected from close
-    for (let i = this.ports.length - 1; i >= 0; i--) {
-      if (this.ports[i].isProtectedFromClose && !this.ports[i].isClosing) {
-        return this.ports[i];
+  protected async getLastWrappedPort(): Promise<WrappedSyncPort | undefined> {
+    // Find the last port which is not closing
+    return await this.portMutex.runExclusive(() => {
+      for (let i = this.ports.length - 1; i >= 0; i--) {
+        if (!this.ports[i].isClosing) {
+          return this.ports[i];
+        }
       }
-    }
-    return;
+      return;
+    });
   }
 
   async waitForStatus(status: SyncStatusOptions): Promise<void> {
@@ -232,44 +230,45 @@ export class SharedSyncImplementation extends BaseObserver<SharedSyncImplementat
   async setParams(params: SharedSyncInitOptions) {
     await this.portMutex.runExclusive(async () => {
       this.collectActiveSubscriptions();
-      if (this.syncParams) {
-        // Cannot modify already existing sync implementation params
-        return;
-      }
-
-      // First time setting params
-      this.syncParams = params;
-      if (params.streamOptions?.flags?.broadcastLogs) {
-        this.logger = this.broadCastLogger;
-      }
+    });
 
-      const lockedAdapter = new LockedAsyncDatabaseAdapter({
-        name: params.dbParams.dbFilename,
-        openConnection: async () => {
-          // Gets a connection from the clients when a new connection is requested.
-          return await this.openInternalDB();
-        },
-        logger: this.logger,
-        reOpenOnConnectionClosed: true
-      });
-      this.distributedDB = lockedAdapter;
-      await lockedAdapter.init();
-
-      lockedAdapter.registerListener({
-        databaseReOpened: () => {
-          // We may have missed some table updates while the database was closed.
-          // We can poke the crud in case we missed any updates.
-          this.connectionManager.syncStreamImplementation?.triggerCrudUpload();
-        }
-      });
+    if (this.syncParams) {
+      // Cannot modify already existing sync implementation params
+      return;
+    }
 
-      self.onerror = (event) => {
-        // Share any uncaught events on the broadcast logger
-        this.logger.error('Uncaught exception in PowerSync shared sync worker', event);
-      };
+    // First time setting params
+    this.syncParams = params;
+    if (params.streamOptions?.flags?.broadcastLogs) {
+      this.logger = this.broadCastLogger;
+    }
 
-      this.iterateListeners((l) => l.initialized?.());
+    const lockedAdapter = new LockedAsyncDatabaseAdapter({
+      name: params.dbParams.dbFilename,
+      openConnection: async () => {
+        // Gets a connection from the clients when a new connection is requested.
+        return await this.openInternalDB();
+      },
+      logger: this.logger,
+      reOpenOnConnectionClosed: true
+    });
+    this.distributedDB = lockedAdapter;
+    await lockedAdapter.init();
+
+    lockedAdapter.registerListener({
+      databaseReOpened: () => {
+        // We may have missed some table updates while the database was closed.
+        // We can poke the crud in case we missed any updates.
+        this.connectionManager.syncStreamImplementation?.triggerCrudUpload();
+      }
     });
+
+    self.onerror = (event) => {
+      // Share any uncaught events on the broadcast logger
+      this.logger.error('Uncaught exception in PowerSync shared sync worker', event);
+    };
+
+    this.iterateListeners((l) => l.initialized?.());
   }
 
   async dispose() {
@@ -303,7 +302,6 @@ export class SharedSyncImplementation extends BaseObserver<SharedSyncImplementat
         clientProvider: Comlink.wrap<AbstractSharedSyncClientProvider>(port),
         currentSubscriptions: [],
         closeListeners: [],
-        isProtectedFromClose: false,
         isClosing: false
       } satisfies WrappedSyncPort;
       this.ports.push(portProvider);
@@ -409,7 +407,7 @@ export class SharedSyncImplementation extends BaseObserver<SharedSyncImplementat
       remote: new WebRemote(
         {
           invalidateCredentials: async () => {
-            const lastPort = this.lastWrappedPort;
+            const lastPort = await this.getLastWrappedPort();
             if (!lastPort) {
               throw new Error('No client port found to invalidate credentials');
             }
@@ -421,7 +419,7 @@ export class SharedSyncImplementation extends BaseObserver<SharedSyncImplementat
             }
           },
           fetchCredentials: async () => {
-            const lastPort = this.lastWrappedPort;
+            const lastPort = await this.getLastWrappedPort();
             if (!lastPort) {
               throw new Error('No client port found to fetch credentials');
             }
@@ -447,7 +445,7 @@ export class SharedSyncImplementation extends BaseObserver<SharedSyncImplementat
         this.logger
       ),
       uploadCrud: async () => {
-        const lastPort = this.lastWrappedPort;
+        const lastPort = await this.getLastWrappedPort();
         if (!lastPort) {
           throw new Error('No client port found to upload crud');
         }
@@ -483,7 +481,7 @@ export class SharedSyncImplementation extends BaseObserver<SharedSyncImplementat
   protected async openInternalDB() {
     while (true) {
       try {
-        const lastClient = this.lastWrappedPort;
+        const lastClient = await this.getLastWrappedPort();
         if (!lastClient) {
           // Should not really happen in practice
           throw new Error(`Could not open DB connection since no client is connected.`);
@@ -563,19 +561,6 @@ export class SharedSyncImplementation extends BaseObserver<SharedSyncImplementat
     this.syncStatus = new SyncStatus(status);
     this.ports.forEach((p) => p.clientProvider.statusChanged(status));
   }
-
-  /**
-   * A function only used for unit tests which updates the internal
-   * sync stream client and all tab client's sync status
-   */
-  async _testUpdateAllStatuses(status: SyncStatusOptions) {
-    if (!this.connectionManager.syncStreamImplementation) {
-      throw new Error('Cannot update status without a sync stream implementation');
-    }
-    // Only assigning, don't call listeners for this test
-    this.connectionManager.syncStreamImplementation!.syncStatus = new SyncStatus(status);
-    this.updateAllStatuses(status);
-  }
 }
 
 /**

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

@@ -10,5 +10,5 @@ const sharedSyncImplementation = new SharedSyncImplementation();
 
 _self.onconnect = async function (event: MessageEvent<string>) {
   const port = event.ports[0];
-  await new WorkerClient(sharedSyncImplementation, port).initialize();
+  new WorkerClient(sharedSyncImplementation, port);
 };

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

@@ -1,4 +1,4 @@
-import { ILogLevel, PowerSyncConnectionOptions, SubscribedStream, SyncStatusOptions } from '@powersync/common';
+import { ILogLevel, PowerSyncConnectionOptions, SubscribedStream } from '@powersync/common';
 import * as Comlink from 'comlink';
 import { getNavigatorLocks } from '../../shared/navigator';
 import {
@@ -24,9 +24,6 @@ export class WorkerClient {
     private readonly port: MessagePort
   ) {
     Comlink.expose(this, this.port);
-  }
-
-  async initialize() {
     /**
      * Adds an extra listener which can remove this port
      * from the list of monitored ports.
@@ -37,16 +34,6 @@ export class WorkerClient {
         await this.removePort();
       }
     });
-
-    /**
-     * Keep a reference to the resolved port promise.
-     * The init timing is difficult to predict due to the async message passing.
-     * We only want to use a port if we are know it's been protected from being closed.
-     * The lock based close signal will be added asynchronously. We need to use the
-     * added port once the lock is configured.
-     */
-    this.resolvedPortPromise = this.sync.addPort(this.port);
-    this.resolvedPort = await this.resolvedPortPromise;
   }
 
   private async removePort() {
@@ -70,18 +57,9 @@ export class WorkerClient {
    * it can consider the connection to be closed.
    */
   async addLockBasedCloseSignal(name: string) {
-    if (!this.resolvedPortPromise) {
-      // The init logic above is actually synchronous, so this should not happen.
-      this.sync.broadCastLogger.warn('addLockBasedCloseSignal called before port promise registered');
-    } else {
-      const wrappedPort = await this.resolvedPortPromise;
-      /**
-       * The client registered a navigator lock. We now can guarantee detecting if the client has closed.
-       * E.g. before this point: It's possible some ports might have been created and closed before the
-       * lock based close signal is added. We should not trust those ports.
-       */
-      wrappedPort.isProtectedFromClose = true;
-    }
+    // Only add the port once the lock has been obtained on the client.
+    this.resolvedPort = await this.sync.addPort(this.port);
+    // Don't await this lock request
     getNavigatorLocks().request(name, async () => {
       await this.removePort();
     });
@@ -121,8 +99,4 @@ export class WorkerClient {
   disconnect() {
     return this.sync.disconnect();
   }
-
-  async _testUpdateAllStatuses(status: SyncStatusOptions) {
-    return this.sync._testUpdateAllStatuses(status);
-  }
 }