fix opfs deadlocks

Author: stevensJourney

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

@@ -6,16 +6,16 @@ import {
   SyncStatusOptions
 } from '@powersync/common';
 import * as Comlink from 'comlink';
+import { getNavigatorLocks } from '../../shared/navigator';
 import { AbstractSharedSyncClientProvider } from '../../worker/sync/AbstractSharedSyncClientProvider';
 import { ManualSharedSyncPayload, SharedSyncClientEvent } from '../../worker/sync/SharedSyncImplementation';
-import { DEFAULT_CACHE_SIZE_KB, resolveWebSQLFlags, TemporaryStorageOption } from '../adapters/web-sql-flags';
+import { WorkerClient } from '../../worker/sync/WorkerClient';
 import { WebDBAdapter } from '../adapters/WebDBAdapter';
+import { DEFAULT_CACHE_SIZE_KB, TemporaryStorageOption, resolveWebSQLFlags } from '../adapters/web-sql-flags';
 import {
   WebStreamingSyncImplementation,
   WebStreamingSyncImplementationOptions
 } from './WebStreamingSyncImplementation';
-import { WorkerClient } from '../../worker/sync/WorkerClient';
-import { getNavigatorLocks } from '../../shared/navigator';
 
 /**
  * The shared worker will trigger methods on this side of the message port
@@ -160,6 +160,21 @@ export class SharedWebStreamingSyncImplementation extends WebStreamingSyncImplem
     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(),
@@ -190,19 +205,6 @@ export class SharedWebStreamingSyncImplementation extends WebStreamingSyncImplem
      * This performs bi-directional method calling.
      */
     Comlink.expose(this.clientProvider, this.messagePort);
-
-    // 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.
-    getNavigatorLocks().request(`tab-close-signal-${crypto.randomUUID()}`, async (lock) => {
-      if (!this.abortOnClose.signal.aborted) {
-        this.syncManager.addLockBasedCloseSignal(lock!.name);
-
-        await new Promise<void>((r) => {
-          this.abortOnClose.signal.onabort = () => r();
-        });
-      }
-    });
   }
 
   /**

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

@@ -322,6 +322,22 @@ 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
+       * connect operations coming from other tabs.
+       */
+      if (this.dbAdapter && this.dbAdapter == trackedPort.db) {
+        this.logger.debug('disconnecting due to closed database', shouldReconnect);
+        this.dbAdapter = null;
+        // Unconditionally close the connection because the database it's writing to has just been closed.
+        // The connection has been closed previously, this might throw. We should be able to ignore it.
+        await this.connectionManager
+          .disconnect()
+          .catch((ex) => this.logger.warn('Error while disconnecting. Will attempt to reconnect.', ex));
+      }
+
       return {
         shouldReconnect,
         trackedPort
@@ -337,19 +353,8 @@ export class SharedSyncImplementation extends BaseObserver<SharedSyncImplementat
       await closeListener();
     }
 
-    if (this.dbAdapter && this.dbAdapter == trackedPort.db) {
-      // Unconditionally close the connection because the database it's writing to has just been closed.
-      // The connection has been closed previously, this might throw. We should be able to ignore it.
-      await this.connectionManager
-        .disconnect()
-        .catch((ex) => this.logger.warn('Error while disconnecting. Will attempt to reconnect.', ex));
-
-      // 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 ?? {});
-      }
+    if (shouldReconnect) {
+      await this.connectionManager.connect(CONNECTOR_PLACEHOLDER, this.lastConnectOptions ?? {});
     }
 
     // Re-index subscriptions, the subscriptions of the removed port would no longer be considered.
@@ -487,7 +492,12 @@ export class SharedSyncImplementation extends BaseObserver<SharedSyncImplementat
         });
         lastClient.closeListeners.push(async () => {
           this.logger.info('Aborting open connection because associated tab closed.');
-          await wrapped.close().catch((ex) => this.logger.warn('error closing database connection', ex));
+          /**
+           * Don't await this close operation. It might never resolve if the tab is closed.
+           * We run the close operation first, before marking the remote as closed. This gives the database some chance
+           * to close the connection.
+           */
+          wrapped.close().catch((ex) => this.logger.warn('error closing database connection', ex));
           wrapped.markRemoteClosed();
         });
 

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

@@ -1,13 +1,13 @@
+import { ILogLevel, PowerSyncConnectionOptions, SubscribedStream, SyncStatusOptions } from '@powersync/common';
 import * as Comlink from 'comlink';
+import { getNavigatorLocks } from '../../shared/navigator';
 import {
   ManualSharedSyncPayload,
   SharedSyncClientEvent,
   SharedSyncImplementation,
   SharedSyncInitOptions,
   WrappedSyncPort
 } from './SharedSyncImplementation';
-import { ILogLevel, PowerSyncConnectionOptions, SubscribedStream, SyncStatusOptions } from '@powersync/common';
-import { getNavigatorLocks } from '../../shared/navigator';
 
 /**
  * A client to the shared sync worker.
@@ -21,7 +21,9 @@ export class WorkerClient {
   constructor(
     private readonly sync: SharedSyncImplementation,
     private readonly port: MessagePort
-  ) {}
+  ) {
+    Comlink.expose(this, this.port);
+  }
 
   async initialize() {
     /**
@@ -36,7 +38,6 @@ export class WorkerClient {
     });
 
     this.resolvedPort = await this.sync.addPort(this.port);
-    Comlink.expose(this, this.port);
   }
 
   private async removePort() {