Add default lock timeout for shared sync workers. Add flagging for unprotected clients. Perform reconnect operations inside shared port mutex.

Author: stevensJourney

demos/react-supabase-todolist/src/components/providers/SystemProvider.tsx

@@ -3,7 +3,14 @@ import { AppSchema, ListRecord, LISTS_TABLE, TODOS_TABLE } from '@/library/power
 import { SupabaseConnector } from '@/library/powersync/SupabaseConnector';
 import { CircularProgress } from '@mui/material';
 import { PowerSyncContext } from '@powersync/react';
-import { createBaseLogger, DifferentialWatchedQuery, LogLevel, PowerSyncDatabase } from '@powersync/web';
+import {
+  createBaseLogger,
+  DifferentialWatchedQuery,
+  LogLevel,
+  PowerSyncDatabase,
+  WASQLiteOpenFactory,
+  WASQLiteVFS
+} from '@powersync/web';
 import React, { Suspense } from 'react';
 import { NavigationPanelContextProvider } from '../navigation/NavigationPanelContext';
 
@@ -12,8 +19,15 @@ export const useSupabase = () => React.useContext(SupabaseContext);
 
 export const db = new PowerSyncDatabase({
   schema: AppSchema,
-  database: {
-    dbFilename: 'example.db'
+  database: new WASQLiteOpenFactory({
+    dbFilename: 'example.db',
+    vfs: WASQLiteVFS.OPFSCoopSyncVFS,
+    flags: {
+      enableMultiTabs: typeof SharedWorker !== 'undefined'
+    }
+  }),
+  flags: {
+    enableMultiTabs: typeof SharedWorker !== 'undefined'
   }
 });
 

packages/web/src/db/adapters/LockedAsyncDatabaseAdapter.ts

@@ -26,6 +26,7 @@ export interface LockedAsyncDatabaseAdapterOptions {
   openConnection: () => Promise<AsyncDatabaseConnection>;
   debugMode?: boolean;
   logger?: ILogger;
+  defaultLockTimeoutMs?: number;
 }
 
 export type LockedAsyncDatabaseAdapterListener = DBAdapterListener & {
@@ -196,7 +197,7 @@ export class LockedAsyncDatabaseAdapter
     return this.acquireLock(
       async () => fn(this.generateDBHelpers({ execute: this._execute, executeRaw: this._executeRaw })),
       {
-        timeoutMs: options?.timeoutMs
+        timeoutMs: options?.timeoutMs ?? this.options.defaultLockTimeoutMs
       }
     );
   }
@@ -206,7 +207,7 @@ export class LockedAsyncDatabaseAdapter
     return this.acquireLock(
       async () => fn(this.generateDBHelpers({ execute: this._execute, executeRaw: this._executeRaw })),
       {
-        timeoutMs: options?.timeoutMs
+        timeoutMs: options?.timeoutMs ?? this.options.defaultLockTimeoutMs
       }
     );
   }

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

@@ -157,36 +157,8 @@ export class SharedWebStreamingSyncImplementation extends WebStreamingSyncImplem
      * DB worker, but a port to the DB worker can be transferred to the
      * sync worker.
      */
-    const { crudUploadThrottleMs, identifier, retryDelayMs } = this.options;
-    const flags = { ...this.webOptions.flags, workers: undefined };
-
-    // Request a random lock until this client is disposed. The name of the lock is sent to the shared worker, which
-    // will also attempt to acquire it. Since the lock is returned when the tab is closed, this allows the share worker
-    // to free resources associated with this tab.
-    // We take hold of this lock as soon-as-possible in order to cater for potentially closed tabs.
-    getNavigatorLocks().request(`tab-close-signal-${crypto.randomUUID()}`, async (lock) => {
-      if (!this.abortOnClose.signal.aborted) {
-        // Awaiting here ensures the worker is waiting for the lock
-        await this.syncManager.addLockBasedCloseSignal(lock!.name);
-
-        await new Promise<void>((r) => {
-          this.abortOnClose.signal.onabort = () => r();
-        });
-      }
-    });
 
-    this.isInitialized = this.syncManager.setParams(
-      {
-        dbParams: this.dbAdapter.getConfiguration(),
-        streamOptions: {
-          crudUploadThrottleMs,
-          identifier,
-          retryDelayMs,
-          flags: flags
-        }
-      },
-      options.subscriptions
-    );
+    this.isInitialized = this._init();
 
     /**
      * Pass along any sync status updates to this listener
@@ -198,13 +170,62 @@ export class SharedWebStreamingSyncImplementation extends WebStreamingSyncImplem
       },
       options.db
     );
+  }
+
+  protected async _init() {
+    /**
+     * The general flow of initialization is:
+     *  - The client requests a unique navigator lock.
+     *    - Once the lock is acquired, we register the lock with the shared worker.
+     *    - 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 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.
+     */
+    await new Promise<void>((resolve) => {
+      // Request a random lock until this client is disposed. The name of the lock is sent to the shared worker, which
+      // will also attempt to acquire it. Since the lock is returned when the tab is closed, this allows the share worker
+      // to free resources associated with this tab.
+      // We take hold of this lock as soon-as-possible in order to cater for potentially closed tabs.
+      getNavigatorLocks().request(`tab-close-signal-${crypto.randomUUID()}`, async (lock) => {
+        if (!this.abortOnClose.signal.aborted) {
+          // Awaiting here ensures the worker is waiting for the lock
+          await this.syncManager.addLockBasedCloseSignal(lock!.name);
+
+          // The lock has been registered, we can continue with the initialization
+          resolve();
+
+          await new Promise<void>((r) => {
+            this.abortOnClose.signal.onabort = () => r();
+          });
+        }
+      });
+    });
+
+    const { crudUploadThrottleMs, identifier, retryDelayMs } = this.options;
+    const flags = { ...this.webOptions.flags, workers: undefined };
 
     /**
      * The sync worker will call this client provider when it needs
      * to fetch credentials or upload data.
      * This performs bi-directional method calling.
      */
     Comlink.expose(this.clientProvider, this.messagePort);
+
+    await this.syncManager.setParams(
+      {
+        dbParams: this.dbAdapter.getConfiguration(),
+        streamOptions: {
+          crudUploadThrottleMs,
+          identifier,
+          retryDelayMs,
+          flags: flags
+        }
+      },
+      this.options.subscriptions
+    );
   }
 
   /**

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

@@ -76,6 +76,10 @@ 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;
 };
 
 /**
@@ -142,7 +146,9 @@ export class SharedSyncImplementation extends BaseObserver<SharedSyncImplementat
       createSyncImplementation: async () => {
         return this.portMutex.runExclusive(async () => {
           await this.waitForReady();
+          this.logger.debug('Creating sync implementation');
           if (!this.dbAdapter) {
+            this.logger.debug('Opening internal DB');
             await this.openInternalDB();
           }
 
@@ -171,6 +177,19 @@ export class SharedSyncImplementation extends BaseObserver<SharedSyncImplementat
     return this.connectionManager.syncStreamImplementation?.isConnected ?? false;
   }
 
+  /**
+   * 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) {
+        return this.ports[i];
+      }
+    }
+    return;
+  }
+
   async waitForStatus(status: SyncStatusOptions): Promise<void> {
     return this.withSyncImplementation(async (sync) => {
       return sync.waitForStatus(status);
@@ -276,7 +295,8 @@ export class SharedSyncImplementation extends BaseObserver<SharedSyncImplementat
         port,
         clientProvider: Comlink.wrap<AbstractSharedSyncClientProvider>(port),
         currentSubscriptions: [],
-        closeListeners: []
+        closeListeners: [],
+        isProtectedFromClose: false
       } satisfies WrappedSyncPort;
       this.ports.push(portProvider);
 
@@ -298,11 +318,11 @@ export class SharedSyncImplementation extends BaseObserver<SharedSyncImplementat
     // Remove the port within a mutex context.
     // Warns if the port is not found. This should not happen in practice.
     // We return early if the port is not found.
-    const { trackedPort, shouldReconnect } = await this.portMutex.runExclusive(async () => {
+    return await this.portMutex.runExclusive(async () => {
       const index = this.ports.findIndex((p) => p == 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];
@@ -322,7 +342,6 @@ export class SharedSyncImplementation extends BaseObserver<SharedSyncImplementat
       });
 
       const shouldReconnect = !!this.connectionManager.syncStreamImplementation && this.ports.length > 0;
-
       /**
        * If the current database adapter is the one that is being closed, we need to disconnect from the backend.
        * We can disconnect in the portMutex lock. This ensures the disconnect is not affected by potential other
@@ -338,30 +357,25 @@ export class SharedSyncImplementation extends BaseObserver<SharedSyncImplementat
           .catch((ex) => this.logger.warn('Error while disconnecting. Will attempt to reconnect.', ex));
       }
 
-      return {
-        shouldReconnect,
-        trackedPort
-      };
-    });
-
-    if (!trackedPort) {
-      // We could not find the port to remove
-      return () => {};
-    }
-
-    for (const closeListener of trackedPort.closeListeners) {
-      await closeListener();
-    }
+      for (const closeListener of trackedPort.closeListeners) {
+        await closeListener();
+      }
 
-    if (shouldReconnect) {
-      await this.connectionManager.connect(CONNECTOR_PLACEHOLDER, this.lastConnectOptions ?? {});
-    }
+      try {
+        await trackedPort.db?.close();
+      } catch (ex) {
+        this.logger.warn('error closing database', ex);
+      }
 
-    // Re-index subscriptions, the subscriptions of the removed port would no longer be considered.
-    this.collectActiveSubscriptions();
+      // Re-index subscriptions, the subscriptions of the removed port would no longer be considered.
+      this.collectActiveSubscriptions();
+      if (shouldReconnect) {
+        // The internals of this needs a port mutex lock. It should be safe to start this operation here, but we cannot and don't need to await it.
+        this.connectionManager.connect(CONNECTOR_PLACEHOLDER, this.lastConnectOptions ?? {});
+      }
 
-    // Release proxy
-    return () => trackedPort.clientProvider[Comlink.releaseProxy]();
+      return () => trackedPort.clientProvider[Comlink.releaseProxy]();
+    });
   }
 
   triggerCrudUpload() {
@@ -410,7 +424,10 @@ export class SharedSyncImplementation extends BaseObserver<SharedSyncImplementat
       remote: new WebRemote(
         {
           invalidateCredentials: async () => {
-            const lastPort = this.ports[this.ports.length - 1];
+            const lastPort = this.lastWrappedPort;
+            if (!lastPort) {
+              throw new Error('No client port found to invalidate credentials');
+            }
             try {
               this.logger.log('calling the last port client provider to invalidate credentials');
               lastPort.clientProvider.invalidateCredentials();
@@ -419,7 +436,10 @@ export class SharedSyncImplementation extends BaseObserver<SharedSyncImplementat
             }
           },
           fetchCredentials: async () => {
-            const lastPort = this.ports[this.ports.length - 1];
+            const lastPort = this.lastWrappedPort;
+            if (!lastPort) {
+              throw new Error('No client port found to fetch credentials');
+            }
             return new Promise(async (resolve, reject) => {
               const abortController = new AbortController();
               this.fetchCredentialsController = {
@@ -442,7 +462,10 @@ export class SharedSyncImplementation extends BaseObserver<SharedSyncImplementat
         this.logger
       ),
       uploadCrud: async () => {
-        const lastPort = this.ports[this.ports.length - 1];
+        const lastPort = this.lastWrappedPort;
+        if (!lastPort) {
+          throw new Error('No client port found to upload crud');
+        }
 
         return new Promise(async (resolve, reject) => {
           const abortController = new AbortController();
@@ -470,7 +493,7 @@ export class SharedSyncImplementation extends BaseObserver<SharedSyncImplementat
   }
 
   protected async openInternalDB() {
-    const lastClient = this.ports[this.ports.length - 1];
+    const lastClient = this.lastWrappedPort;
     if (!lastClient) {
       // Should not really happen in practice
       throw new Error(`Could not open DB connection since no client is connected.`);
@@ -481,6 +504,7 @@ export class SharedSyncImplementation extends BaseObserver<SharedSyncImplementat
     const db = await remote(this.syncParams!.dbParams);
     const locked = new LockedAsyncDatabaseAdapter({
       name: identifier,
+      defaultLockTimeoutMs: 20_000, // Max wait time for a lock request (we will retry failed attempts)
       openConnection: async () => {
         const wrapped = new WorkerWrappedAsyncDatabaseConnection({
           remote,

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

@@ -17,6 +17,7 @@ import {
  */
 export class WorkerClient {
   private resolvedPort: WrappedSyncPort | null = null;
+  protected resolvedPortPromise: Promise<WrappedSyncPort> | null = null;
 
   constructor(
     private readonly sync: SharedSyncImplementation,
@@ -37,7 +38,15 @@ export class WorkerClient {
       }
     });
 
-    this.resolvedPort = await this.sync.addPort(this.port);
+    /**
+     * 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() {
@@ -61,6 +70,19 @@ export class WorkerClient {
    * it can consider the connection to be closed.
    */
   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 {
+      this.resolvedPortPromise.then((wrappedPort) => {
+        /**
+         * 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;
+      });
+    }
     getNavigatorLocks().request(name, async () => {
       await this.removePort();
     });