Cleanup aborted operations if aborted. Try and gracefully handle Accesshandle errors.

Author: stevensJourney

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

@@ -282,14 +282,10 @@ export class LockedAsyncDatabaseAdapter
         try {
           // The database is being opened in the background. Wait for it here.
           if (this.databaseOpenPromise) {
-            try {
-              await this.databaseOpenPromise;
-            } catch (ex) {
-              // This will cause a retry of opening the database.
-              const wrappedError = new ConnectionClosedError('Could not open database');
-              wrappedError.cause = ex;
-              throw wrappedError;
-            }
+            /**
+             * We can't await this since it uses the same lock as we're in now.
+             */
+            throw new ConnectionClosedError('Connection is busy re-opening');
           }
 
           holdId = this.requiresHolds ? await this.baseDB.markHold() : null;

packages/web/src/worker/db/opfs.ts

@@ -259,19 +259,7 @@ export class OPFSCoopSyncVFS extends FacadeVFS {
       this.mapIdToFile.delete(fileId);
 
       if (file?.flags & VFS.SQLITE_OPEN_MAIN_DB) {
-        // Always close handles directly as well to ensure cleanup
-        // This handles edge cases where the lock mechanism might not work properly
-        if (file.persistentFile) {
-          DB_RELATED_FILE_SUFFIXES.forEach((suffix) => {
-            const persistentFile = this.persistentFiles.get(file.path + suffix);
-            if (persistentFile?.accessHandle) {
-              persistentFile.accessHandle.close();
-              persistentFile.accessHandle = null;
-            }
-          });
-        }
         if (file.persistentFile?.handleLockReleaser) {
-          // Normal case: release via the lock mechanism
           this.#releaseAccessHandle(file);
         }
       } else if (file?.flags & VFS.SQLITE_OPEN_DELETEONCLOSE) {
@@ -532,7 +520,6 @@ export class OPFSCoopSyncVFS extends FacadeVFS {
         (async () => {
           // Acquire the Web Lock.
           file.persistentFile.handleLockReleaser = await this.#acquireLock(file.persistentFile);
-
           try {
             // Get access handles for the database and releated files in parallel.
             this.log?.(`creating access handles for ${file.path}`);
@@ -544,11 +531,13 @@ export class OPFSCoopSyncVFS extends FacadeVFS {
                 }
               })
             );
-            file.persistentFile.isRequestInProgress = false;
           } catch (e) {
             this.log?.(`failed to create access handles for ${file.path}`, e);
+            // Release the lock, if we failed here, we'd need to obtain the lock later in order to retry
             file.persistentFile.handleLockReleaser();
             throw e;
+          } finally {
+            file.persistentFile.isRequestInProgress = false;
           }
         })()
       );
@@ -581,6 +570,7 @@ export class OPFSCoopSyncVFS extends FacadeVFS {
    */
   #acquireLock(persistentFile) {
     return new Promise((resolve) => {
+      // Tell other connections we want the access handle.
       const lockName = persistentFile.handleRequestChannel.name;
       const notify = () => {
         this.log?.(`notifying for ${lockName}`);
@@ -590,19 +580,11 @@ export class OPFSCoopSyncVFS extends FacadeVFS {
       setTimeout(notify);
 
       this.log?.(`lock requested: ${lockName}`);
-      let releaseLock; // This will hold the resolver for the lock Promise
-      const lockPromise = new Promise((release) => {
-        releaseLock = release; // Capture the resolver
-      });
-
       navigator.locks.request(lockName, (lock) => {
+        // We have the lock. Stop asking other connections for it.
         this.log?.(`lock acquired: ${lockName}`, lock);
         clearInterval(notifyId);
-        // Resolve the outer Promise with a function that releases the lock
-        resolve(() => {
-          releaseLock(); // This resolves lockPromise, releasing the lock
-        });
-        return lockPromise; // Return the Promise that controls the lock
+        return new Promise(resolve);
       });
     });
   }

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

@@ -525,12 +525,16 @@ export class SharedSyncImplementation extends BaseObserver<SharedSyncImplementat
 
         client.closeListeners.push(closeListener);
 
-        const workerPort = await withAbort(() => client.clientProvider.getDBWorkerPort(), abortController.signal).catch(
-          (ex) => {
-            removeCloseListener();
-            throw ex;
+        const workerPort = await withAbort({
+          action: () => client.clientProvider.getDBWorkerPort(),
+          signal: abortController.signal,
+          cleanupOnAbort: (port) => {
+            port.close();
           }
-        );
+        }).catch((ex) => {
+          removeCloseListener();
+          throw ex;
+        });
 
         const remote = Comlink.wrap<OpenAsyncDatabaseConnection>(workerPort);
         const identifier = this.syncParams!.dbParams.dbFilename;
@@ -541,7 +545,13 @@ export class SharedSyncImplementation extends BaseObserver<SharedSyncImplementat
          * We typically execute the closeListeners using the portMutex in a different context.
          * We can't rely on the closeListeners to abort the operation if the tab is closed.
          */
-        const db = await withAbort(() => remote(this.syncParams!.dbParams), abortController.signal).finally(() => {
+        const db = await withAbort({
+          action: () => remote(this.syncParams!.dbParams),
+          signal: abortController.signal,
+          cleanupOnAbort: (db) => {
+            db.close();
+          }
+        }).finally(() => {
           // We can remove the close listener here since we no longer need it past this point.
           removeCloseListener();
         });
@@ -588,7 +598,12 @@ export class SharedSyncImplementation extends BaseObserver<SharedSyncImplementat
 /**
  * Runs the action with an abort controller.
  */
-function withAbort<T>(action: () => Promise<T>, signal: AbortSignal): Promise<T> {
+function withAbort<T>(options: {
+  action: () => Promise<T>;
+  signal: AbortSignal;
+  cleanupOnAbort?: (result: T) => void;
+}): Promise<T> {
+  const { action, signal, cleanupOnAbort } = options;
   return new Promise((resolve, reject) => {
     if (signal.aborted) {
       reject(new AbortOperation('Operation aborted by abort controller'));
@@ -608,7 +623,13 @@ function withAbort<T>(action: () => Promise<T>, signal: AbortSignal): Promise<T>
     }
 
     action()
-      .then((data) => completePromise(() => resolve(data)))
+      .then((data) => {
+        completePromise(() => resolve(data));
+        // We already rejected due to the abort, allow for cleanup
+        if (signal.aborted) {
+          cleanupOnAbort?.(data);
+        }
+      })
       .catch((e) => completePromise(() => reject(e)));
   });
 }