Wrap entire open operation init in writeLock.

Author: stevensJourney

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

@@ -113,23 +113,28 @@ export class LockedAsyncDatabaseAdapter
   }
 
   protected async openInternalDB() {
-    // Dispose any previous table change listener.
-    this._disposeTableChangeListener?.();
-    this._disposeTableChangeListener = null;
-
-    const isReOpen = !!this._db;
-
-    this._db = await this.options.openConnection();
-    await this._db.init();
-    this._config = await this._db.getConfig();
-    await this.registerOnChangeListener(this._db);
-    if (isReOpen) {
-      this.iterateListeners((cb) => cb.databaseReOpened?.());
-    }
     /**
-     * This is only required for the long-lived shared IndexedDB connections.
+     * Execute opening of the db in a lock in order not to interfere with other operations.
      */
-    this.requiresHolds = (this._config as ResolvedWASQLiteOpenFactoryOptions).vfs == WASQLiteVFS.IDBBatchAtomicVFS;
+    return this._acquireLock(async () => {
+      // Dispose any previous table change listener.
+      this._disposeTableChangeListener?.();
+      this._disposeTableChangeListener = null;
+
+      const isReOpen = !!this._db;
+
+      this._db = await this.options.openConnection();
+      await this._db.init();
+      this._config = await this._db.getConfig();
+      await this.registerOnChangeListener(this._db);
+      if (isReOpen) {
+        this.iterateListeners((cb) => cb.databaseReOpened?.());
+      }
+      /**
+       * This is only required for the long-lived shared IndexedDB connections.
+       */
+      this.requiresHolds = (this._config as ResolvedWASQLiteOpenFactoryOptions).vfs == WASQLiteVFS.IDBBatchAtomicVFS;
+    });
   }
 
   protected _reOpen() {
@@ -252,13 +257,10 @@ export class LockedAsyncDatabaseAdapter
     );
   }
 
-  protected async acquireLock(callback: () => Promise<any>, options?: { timeoutMs?: number }): Promise<any> {
-    await this.waitForInitialized();
-
+  protected async _acquireLock(callback: () => Promise<any>, options?: { timeoutMs?: number }): Promise<any> {
     if (this.closing) {
       throw new Error(`Cannot acquire lock, the database is closing`);
     }
-
     const abortController = new AbortController();
     this.pendingAbortControllers.add(abortController);
     const { timeoutMs } = options ?? {};
@@ -278,36 +280,41 @@ export class LockedAsyncDatabaseAdapter
         if (timeoutId) {
           clearTimeout(timeoutId);
         }
-        let holdId: string | null = null;
-        try {
-          // The database is being opened in the background. Wait for it here.
-          if (this.databaseOpenPromise) {
-            /**
-             * We can't await this since it uses the same lock as we're in now.
-             */
-            throw new ConnectionClosedError('Connection is busy re-opening');
-          }
+        return await callback();
+      }
+    );
+  }
 
-          holdId = this.requiresHolds ? await this.baseDB.markHold() : null;
-          return await callback();
-        } catch (ex) {
-          if (
-            ex instanceof ConnectionClosedError ||
-            (ex instanceof Error && ex.name === 'NoModificationAllowedError')
-          ) {
-            if (this.options.reOpenOnConnectionClosed && !this.databaseOpenPromise && !this.closing) {
-              // Immediately re-open the database. We need to miss as little table updates as possible.
-              this.reOpenInternalDB();
-            }
-          }
-          throw ex;
-        } finally {
-          if (holdId) {
-            await this.baseDB.releaseHold(holdId);
+  protected async acquireLock(callback: () => Promise<any>, options?: { timeoutMs?: number }): Promise<any> {
+    await this.waitForInitialized();
+
+    return this._acquireLock(async () => {
+      let holdId: string | null = null;
+      try {
+        // The database is being opened in the background. Wait for it here.
+        if (this.databaseOpenPromise) {
+          /**
+           * 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;
+        return await callback();
+      } catch (ex) {
+        if (ex instanceof ConnectionClosedError || (ex instanceof Error && ex.name === 'NoModificationAllowedError')) {
+          if (this.options.reOpenOnConnectionClosed && !this.databaseOpenPromise && !this.closing) {
+            // Immediately re-open the database. We need to miss as little table updates as possible.
+            this.reOpenInternalDB();
           }
         }
+        throw ex;
+      } finally {
+        if (holdId) {
+          await this.baseDB.releaseHold(holdId);
+        }
       }
-    );
+    }, options);
   }
 
   async readTransaction<T>(fn: (tx: Transaction) => Promise<T>, options?: DBLockOptions | undefined): Promise<T> {

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

@@ -16,13 +16,11 @@ baseLogger.useDefaults();
 const logger = createLogger('db-worker');
 
 const DBMap = new Map<string, SharedDBWorkerConnection>();
-
+const OPEN_DB_LOCK = 'open-wasqlite-db';
 let nextClientId = 1;
 
 const openDBShared = async (options: WorkerDBOpenerOptions): Promise<AsyncDatabaseConnection> => {
-  // Prevent multiple simultaneous opens from causing race conditions
-  // Use the same lock as a write lock to prevent concurrent writes to the database.
-  return getNavigatorLocks().request(`db-lock-${options.dbFilename}`, async () => {
+  return getNavigatorLocks().request(OPEN_DB_LOCK, async () => {
     const clientId = nextClientId++;
     const { dbFilename, logLevel } = options;
 

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

@@ -93,18 +93,21 @@ export class OPFSCoopSyncVFS extends FacadeVFS {
       });
     });
 
-    finalizationRegistry.register(this, async () => {
-      for (const file of this.persistentFiles.values()) {
-        const release = this.#releaseAccessHandle(file);
-        try {
-          await this.#releaseAccessHandle(file);
-        } catch (e) {
-          this.log?.('error releasing access handle', e);
-        } finally {
-          release();
-        }
-      }
-    });
+    const releaseHandle = async () => {
+      await Promise.all(
+        this.persistentFiles.values().map(async (file) => {
+          try {
+            await this.#releaseAccessHandle(file);
+          } catch (e) {
+            this.log?.('error releasing access handle', e);
+          } finally {
+            release();
+          }
+        })
+      );
+    };
+
+    finalizationRegistry.register(this, releaseHandle);
     finalizationRegistry.register(this, this.releaser);
     const tmpDir = await root.getDirectoryHandle(tmpDirName, { create: true });
 
@@ -272,9 +275,10 @@ export class OPFSCoopSyncVFS extends FacadeVFS {
       this.mapIdToFile.delete(fileId);
 
       if (file?.flags & VFS.SQLITE_OPEN_MAIN_DB) {
-        if (file.persistentFile?.handleLockReleaser) {
-          this.#releaseAccessHandle(file);
-        }
+        // Release this either way
+        // if (file.persistentFile?.handleLockReleaser) {
+        this.#releaseAccessHandle(file);
+        // }
       } else if (file?.flags & VFS.SQLITE_OPEN_DELETEONCLOSE) {
         file.accessHandle.truncate(0);
         this.accessiblePaths.delete(file.path);