Merge branch '55-fix-storage-indexeddb-adapter-issues' into 'development'

Resolve "fix(Storage): IndexedDB Adapter issues"

Closes #55

See merge request GeneralProtocols/xo/stack!128

package-lock.json

@@ -3563,15 +3563,6 @@
       "dev": true,
       "license": "MIT"
     },
-    "node_modules/@types/trusted-types": {
-      "version": "2.0.7",
-      "resolved": "https://registry.npmjs.org/@types/trusted-types/-/trusted-types-2.0.7.tgz",
-      "integrity": "sha512-ScaPdn1dQczgbl0QFTeTOmVHFULt394XJgOQNoyVhZ6r2vLnMLJfBPd53SB52T/3G36VI1/g2MZaX0cwDuXsfw==",
-      "dev": true,
-      "license": "MIT",
-      "optional": true,
-      "peer": true
-    },
     "node_modules/@types/unist": {
       "version": "3.0.3",
       "resolved": "https://registry.npmjs.org/@types/unist/-/unist-3.0.3.tgz",
@@ -14828,6 +14819,7 @@
         "@electrum-cash/protocol": "2.1.0-development.8075178089",
         "@generalprotocols/oracle-client": "^0.0.1",
         "@xocash/primitives": "*",
+        "async-mutex": "^0.5.0",
         "better-sqlite3": "^11.8.1",
         "idb": "8.0.3",
         "json-to-graphql-query": "^2.3.0",

packages/stack/package.json

@@ -40,6 +40,7 @@
     "@electrum-cash/protocol": "2.1.0-development.8075178089",
     "@generalprotocols/oracle-client": "^0.0.1",
     "@xocash/primitives": "*",
+    "async-mutex": "^0.5.0",
     "better-sqlite3": "^11.8.1",
     "idb": "8.0.3",
     "json-to-graphql-query": "^2.3.0",

packages/stack/src/storage/store-indexeddb.ts

@@ -8,13 +8,16 @@ import { Encrypter } from '@xocash/primitives';
 import { pack, unpack } from 'msgpackr';
 
 import { openDB, deleteDB, IDBPDatabase } from 'idb';
+import { Mutex } from 'async-mutex';
 
 export class StorageIndexedDB extends BaseStorage {
   private db: IDBPDatabase;
   private dbName: string;
-  private versionChangePromise: Promise<void> | null = null;
   private activeStores: Set<StoreIndexedDB> = new Set(); // Track all active stores
 
+  // Mutex to prevent multiple upgrades from happening at the same time, resulting in a blocked upgrade state
+  private upgradeMutex = new Mutex();
+
   private constructor(db: IDBPDatabase, dbName: string) {
     super();
     this.db = db;
@@ -26,11 +29,32 @@ export class StorageIndexedDB extends BaseStorage {
     });
   }
 
-  // Wait for any ongoing version changes to complete
+  private async upgradeDb(upgrade: (db: IDBPDatabase) => void) {
-  private async waitForVersionChange(): Promise<void> {
+    const currentVersion = this.db.version;
-    if (this.versionChangePromise) {
+    const currentDB = this.db;
-      await this.versionChangePromise;
+
-    }
+    const newDb = await openDB(this.dbName, currentVersion + 1, {
+      upgrade(db) {
+        upgrade(db);
+      },
+      blocked() {
+        console.warn(`Database upgrade is blocked, waiting...`);
+        currentDB.close();
+      },
+      blocking() {
+        console.warn(`This connection is blocking upgrades`);
+      },
+      terminated() {
+        console.error(
+          `Database connection was terminated unexpectedly during upgrade`,
+        );
+      },
+    });
+
+    // Update the database reference
+    this.db = newDb;
+
+    return newDb;
   }
 
   public static async createOrOpen(dbName: string) {
@@ -65,66 +89,46 @@ export class StorageIndexedDB extends BaseStorage {
   }
 
   async listStores() {
-    await this.waitForVersionChange();
+    await this.upgradeMutex.waitForUnlock();
     return Array.from(this.db.objectStoreNames);
   }
 
   async createStore(storeName: string, opts: StorageOpts = {}) {
-    await this.waitForVersionChange();
+    // Acquire the mutex so we always have the newest database reference when upgrading
+    const release = await this.upgradeMutex.acquire();
+    try {
+      if (!this.db.objectStoreNames.contains(storeName)) {
+        const newDb = await this.upgradeDb((db) => {
+          db.createObjectStore(storeName);
+        });
 
-    if (!this.db.objectStoreNames.contains(storeName)) {
+        // Update db references in all active stores
-      const currentVersion = this.db.version;
+        for (const store of this.activeStores) {
-
+          store.updateDatabaseReference(newDb);
-      // Set up the version change promise
-      this.versionChangePromise = (async () => {
-        try {
-          const newDb = await openDB(this.dbName, currentVersion + 1, {
-            upgrade(db) {
-              db.createObjectStore(storeName);
-            },
-            blocked() {
-              console.warn(`Database upgrade is blocked, waiting...`);
-            },
-            blocking() {
-              console.warn(`This connection is blocking upgrades`);
-            },
-            terminated() {
-              console.error(
-                `Database connection was terminated unexpectedly during upgrade`,
-              );
-            },
-          });
-
-          // Update the database reference
-          this.db = newDb;
-
-          // Update db references in all active stores
-          for (const store of this.activeStores) {
-            store.updateDatabaseReference(newDb);
-          }
-        } finally {
-          this.versionChangePromise = null;
         }
-      })();
+      }
 
-      // Wait for the version change to complete
+      const store = await StoreIndexedDB.create(
-      await this.versionChangePromise;
+        this.db,
+        storeName,
+        this.upgradeMutex,
+        opts.encryption,
+        this, // Pass a reference to the parent storage
+      );
+
+      this.activeStores.add(store); // Track this store
+      return opts.syncInMemory
+        ? await StoreInMemorySynced.create(store)
+        : store;
+    } finally {
+      // Release the mutex
+      release();
     }
-
-    const store = await StoreIndexedDB.create(
-      this.db,
-      storeName,
-      opts.encryption,
-      this, // Pass a reference to the parent storage
-    );
-
-    this.activeStores.add(store); // Track this store
-    return opts.syncInMemory ? await StoreInMemorySynced.create(store) : store;
   }
 
   async getStore(storeName: string, opts: StorageOpts = {}) {
     try {
-      await this.waitForVersionChange();
+      await this.upgradeMutex.waitForUnlock();
 
       if (!this.db.objectStoreNames.contains(storeName)) {
         return null;
@@ -133,6 +137,7 @@ export class StorageIndexedDB extends BaseStorage {
       const store = await StoreIndexedDB.create(
         this.db,
         storeName,
+        this.upgradeMutex,
         opts.encryption,
         this, // Pass reference to parent
       );
@@ -155,42 +160,20 @@ export class StorageIndexedDB extends BaseStorage {
   }
 
   async deleteStore(storeName: string) {
-    await this.waitForVersionChange();
+    const release = await this.upgradeMutex.acquire();
+    try {
+      if (this.db.objectStoreNames.contains(storeName)) {
+        const newDb = await this.upgradeDb((db) => {
+          db.deleteObjectStore(storeName);
+        });
 
-    if (this.db.objectStoreNames.contains(storeName)) {
+        // Update db references in all active stores
-      const currentVersion = this.db.version;
+        for (const store of this.activeStores) {
-
+          store.updateDatabaseReference(newDb);
-      this.versionChangePromise = (async () => {
-        try {
-          const newDb = await openDB(this.dbName, currentVersion + 1, {
-            upgrade(db) {
-              db.deleteObjectStore(storeName);
-            },
-            blocked() {
-              console.warn(`Database upgrade is blocked, waiting...`);
-            },
-            blocking() {
-              console.warn(`This connection is blocking upgrades`);
-            },
-            terminated() {
-              console.error(
-                `Database connection was terminated unexpectedly during upgrade`,
-              );
-            },
-          });
-
-          this.db = newDb;
-
-          // Update db references in all active stores
-          for (const store of this.activeStores) {
-            store.updateDatabaseReference(newDb);
-          }
-        } finally {
-          this.versionChangePromise = null;
         }
-      })();
+      }
-
+    } finally {
-      await this.versionChangePromise;
+      release();
     }
   }
 
@@ -218,6 +201,7 @@ export class StoreIndexedDB extends BaseStore {
   constructor(
     public db: IDBPDatabase,
     public storeName: string,
+    public databaseUpgradeMutex: Mutex,
     public encryption?: Encrypter,
     private parentStorage?: StorageIndexedDB, // Reference to parent storage
   ) {
@@ -235,29 +219,42 @@ export class StoreIndexedDB extends BaseStore {
    *
    * @param db - An instance of an upgraded and opened IndexedDB database
    * @param storeName - The name of the object store within the database to use
+   * @param databaseUpgradeMutex - A mutex handed down from the parent StorageIndexedDB instance to wait for any upgrades to complete before accessing the database
    * @param encryption - Optional encrypter for encrypting/decrypting data
    * @param parentStorage - Reference to the parent StorageIndexedDB instance
    *
    * @example
    * ```typescript
-   * const store = await StoreIndexedDB.create(db, 'myStore');
+   * const store = await StoreIndexedDB.create(db, 'myStore', databaseUpgradeMutex);
    * ```
    */
   static async create(
     db: IDBPDatabase,
     storeName: string,
+    databaseUpgradeMutex: Mutex,
     encryption?: Encrypter,
     parentStorage?: StorageIndexedDB,
   ) {
-    return new this(db, storeName, encryption, parentStorage);
+    return new this(
+      db,
+      storeName,
+      databaseUpgradeMutex,
+      encryption,
+      parentStorage,
+    );
   }
 
   async all<T>() {
+    // Wait for any upgrades to complete
+    await this.databaseUpgradeMutex.waitForUnlock();
+
     const tx = this.db.transaction(this.storeName, 'readonly');
     const store = tx.objectStore(this.storeName);
     const keys = await store.getAllKeys();
     const values = await store.getAll();
 
+    await tx.done;
+
     return keys.reduce(
       (acc, key, index) => {
         acc[key.toString()] = unpack(values[index]);
@@ -268,13 +265,22 @@ export class StoreIndexedDB extends BaseStore {
   }
 
   async keys() {
+    // Wait for any upgrades to complete
+    await this.databaseUpgradeMutex.waitForUnlock();
+
     const tx = this.db.transaction(this.storeName, 'readonly');
     const store = tx.objectStore(this.storeName);
     const keys = await store.getAllKeys();
+
+    await tx.done;
+
     return keys.map((key) => key.toString());
   }
 
   async get<T>(key: string) {
+    // Wait for any upgrades to complete
+    await this.databaseUpgradeMutex.waitForUnlock();
+
     const encodedValue = await this.db.get(this.storeName, key);
 
     if (!encodedValue) {
@@ -289,6 +295,9 @@ export class StoreIndexedDB extends BaseStore {
   }
 
   async set<T>(key: string, value: T) {
+    // Wait for any upgrades to complete
+    await this.databaseUpgradeMutex.waitForUnlock();
+
     const encodedValue = pack(value);
 
     const encryptedValue = this.encryption
@@ -300,11 +309,17 @@ export class StoreIndexedDB extends BaseStore {
   }
 
   async delete(key: string) {
+    // Wait for any upgrades to complete
+    await this.databaseUpgradeMutex.waitForUnlock();
+
     await this.db.delete(this.storeName, key);
     this.emit('delete', { key });
   }
 
   async batchGet<T>(keys: string[]): Promise<Map<string, T>> {
+    // Wait for any upgrades to complete
+    await this.databaseUpgradeMutex.waitForUnlock();
+
     const tx = this.db.transaction(this.storeName, 'readonly');
     const store = tx.objectStore(this.storeName);
 
@@ -323,6 +338,9 @@ export class StoreIndexedDB extends BaseStore {
   }
 
   async batchSet<T>(items: Map<string, T>) {
+    // Wait for any upgrades to complete
+    await this.databaseUpgradeMutex.waitForUnlock();
+
     const tx = this.db.transaction(this.storeName, 'readwrite');
     const store = tx.objectStore(this.storeName);
 
@@ -344,14 +362,23 @@ export class StoreIndexedDB extends BaseStore {
   }
 
   async clear() {
+    // Wait for any upgrades to complete
+    await this.databaseUpgradeMutex.waitForUnlock();
+
     await this.db.clear(this.storeName);
     this.emit('clear', undefined);
   }
 
   async has(key: string) {
+    // Wait for any upgrades to complete
+    await this.databaseUpgradeMutex.waitForUnlock();
+
     const tx = this.db.transaction(this.storeName, 'readonly');
     const store = tx.objectStore(this.storeName);
     const result = await store.getKey(key);
+
+    await tx.done;
+
     return result !== undefined;
   }