Start looking at OPFS tab close issue

Author: simolus3

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

@@ -203,6 +203,8 @@ export class LockedAsyncDatabaseAdapter
     );
   }
 
+  static dbLockRequestId = 0;
+
   protected async acquireLock(callback: () => Promise<any>, options?: { timeoutMs?: number }): Promise<any> {
     await this.waitForInitialized();
 
@@ -221,13 +223,17 @@ export class LockedAsyncDatabaseAdapter
         }, timeoutMs)
       : null;
 
-    return getNavigatorLocks().request(`db-lock-${this._dbIdentifier}`, { signal: abortController.signal }, () => {
-      this.pendingAbortControllers.delete(abortController);
-      if (timoutId) {
-        clearTimeout(timoutId);
-      }
-      return callback();
-    });
+    const id = LockedAsyncDatabaseAdapter.dbLockRequestId++;
+    console.trace('Requesting database lock', this._dbIdentifier, id);
+    return getNavigatorLocks()
+      .request(`db-lock-${this._dbIdentifier}`, { signal: abortController.signal }, () => {
+        this.pendingAbortControllers.delete(abortController);
+        if (timoutId) {
+          clearTimeout(timoutId);
+        }
+        return callback();
+      })
+      .finally(() => console.log('returning database lock', id));
   }
 
   async readTransaction<T>(fn: (tx: Transaction) => Promise<T>, options?: DBLockOptions | undefined): Promise<T> {

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

@@ -15,6 +15,7 @@ export type SharedConnectionWorker = {
 export type WrappedWorkerConnectionOptions<Config extends ResolvedWebSQLOpenOptions = ResolvedWebSQLOpenOptions> = {
   baseConnection: AsyncDatabaseConnection;
   identifier: string;
+  remoteCanCloseUnexpectedly: boolean;
   /**
    * Need a remote in order to keep a reference to the Proxied worker
    */
@@ -29,10 +30,13 @@ export type WrappedWorkerConnectionOptions<Config extends ResolvedWebSQLOpenOpti
 export class WorkerWrappedAsyncDatabaseConnection<Config extends ResolvedWebSQLOpenOptions = ResolvedWebSQLOpenOptions>
   implements AsyncDatabaseConnection
 {
-  protected lockAbortController: AbortController;
+  protected lockAbortController = new AbortController();
+  protected notifyRemoteClosed: AbortController | undefined;
 
   constructor(protected options: WrappedWorkerConnectionOptions<Config>) {
-    this.lockAbortController = new AbortController();
+    if (options.remoteCanCloseUnexpectedly) {
+      this.notifyRemoteClosed = new AbortController();
+    }
   }
 
   protected get baseConnection() {
@@ -43,6 +47,48 @@ export class WorkerWrappedAsyncDatabaseConnection<Config extends ResolvedWebSQLO
     return this.baseConnection.init();
   }
 
+  /**
+   * Marks the remote as closed.
+   *
+   * This can sometimes happen outside of our control, e.g. when a shared worker requests a connection from a tab. When
+   * it happens, all methods on the {@link baseConnection} would never resolve. To avoid livelocks in this scenario, we
+   * throw on all outstanding promises and forbid new calls.
+   */
+  markRemoteClosed() {
+    // Can non-null assert here because this function is only supposed to be called when remoteCanCloseUnexpectedly was
+    // set.
+    this.notifyRemoteClosed!.abort();
+  }
+
+  private withRemote<T>(inner: () => Promise<T>): Promise<T> {
+    const controller = this.notifyRemoteClosed;
+    if (controller) {
+      return new Promise((resolve, reject) => {
+        if (controller.signal.aborted) {
+          reject(new Error('Called operation on closed remote'));
+        }
+
+        function handleAbort() {
+          reject(new Error('Remote peer closed with request in flight'));
+        }
+
+        function markResolved(inner: () => void) {
+          controller!.signal.removeEventListener('abort', handleAbort);
+          inner();
+        }
+
+        controller.signal.addEventListener('abort', handleAbort);
+
+        inner()
+          .then((data) => markResolved(() => resolve(data)))
+          .catch((e) => markResolved(() => reject(e)));
+      });
+    } else {
+      // Can't close, so just return the inner promise unguarded.
+      return inner();
+    }
+  }
+
   /**
    * Get a MessagePort which can be used to share the internals of this connection.
    */

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

@@ -75,6 +75,7 @@ export type WrappedSyncPort = {
   clientProvider: Comlink.Remote<AbstractSharedSyncClientProvider>;
   db?: DBAdapter;
   currentSubscriptions: SubscribedStream[];
+  closeListeners: (() => void)[];
 };
 
 /**
@@ -274,7 +275,8 @@ export class SharedSyncImplementation extends BaseObserver<SharedSyncImplementat
       const portProvider = {
         port,
         clientProvider: Comlink.wrap<AbstractSharedSyncClientProvider>(port),
-        currentSubscriptions: []
+        currentSubscriptions: [],
+        closeListeners: []
       } satisfies WrappedSyncPort;
       this.ports.push(portProvider);
 
@@ -331,10 +333,13 @@ export class SharedSyncImplementation extends BaseObserver<SharedSyncImplementat
       return () => {};
     }
 
+    for (const closeListener of trackedPort.closeListeners) {
+      closeListener();
+    }
+
     if (this.dbAdapter && this.dbAdapter == trackedPort.db) {
-      if (shouldReconnect) {
-        await this.connectionManager.disconnect();
-      }
+      // Unconditionally close the connection because the database it's writing to has just been closed.
+      await this.connectionManager.disconnect();
 
       // Clearing the adapter will result in a new one being opened in connect
       this.dbAdapter = null;
@@ -473,11 +478,16 @@ export class SharedSyncImplementation extends BaseObserver<SharedSyncImplementat
     const locked = new LockedAsyncDatabaseAdapter({
       name: identifier,
       openConnection: async () => {
-        return new WorkerWrappedAsyncDatabaseConnection({
+        const wrapped = new WorkerWrappedAsyncDatabaseConnection({
           remote,
           baseConnection: db,
           identifier
         });
+        lastClient.closeListeners.push(() => {
+          wrapped.markRemoteClosed();
+        });
+
+        return wrapped;
       },
       logger: this.logger
     });

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

@@ -41,7 +41,9 @@ export class WorkerClient {
 
   private async removePort() {
     if (this.resolvedPort) {
-      const release = await this.sync.removePort(this.resolvedPort);
+      const resolved = this.resolvedPort;
+      this.resolvedPort = null;
+      const release = await this.sync.removePort(resolved);
       this.resolvedPort = null;
       this.port.postMessage({
         event: SharedSyncClientEvent.CLOSE_ACK,