fix(tokens,extended-typeorm): release locks when an update or transaction fails

.changeset/brave-locks-release.md

@@ -0,0 +1,7 @@
+---
+'@rosen-bridge/tokens': patch
+'@rosen-bridge/extended-typeorm': patch
+---
+
+fix(tokens): release the update semaphore when a TokenMap config update fails, so later updates no longer wait forever
+fix(extended-typeorm): release the CustomQueryRunner mutex when starting a transaction or rolling one back fails

packages/extended-typeorm/lib/customQueryRunner.ts

@@ -4,7 +4,7 @@ import { SqliteQueryRunner } from 'typeorm/driver/sqlite/SqliteQueryRunner';
 import { IsolationLevel } from 'typeorm/driver/types/IsolationLevel';
 
 class CustomQueryRunner extends SqliteQueryRunner {
-  releaseMutex: MutexInterface.Releaser | null;
+  releaseMutex: MutexInterface.Releaser | null = null;
   readonly mutex: Mutex;
 
   constructor(driver: SqliteDriver, mutex: Mutex) {
@@ -13,14 +13,24 @@ class CustomQueryRunner extends SqliteQueryRunner {
   }
 
   startTransaction = async (isolationLevel?: IsolationLevel): Promise<void> => {
-    this.releaseMutex = await this.mutex.acquire();
-    return super.startTransaction(isolationLevel);
+    const release = await this.mutex.acquire();
+    try {
+      await super.startTransaction(isolationLevel);
+    } catch (e) {
+      // the transaction never started, so nothing else can release the mutex
+      release();
+      throw e;
+    }
+    this.releaseMutex = release;
   };
 
   commitTransaction = async (): Promise<void> => {
     if (!this.releaseMutex) {
       throw new Error('Cannot commit transaction before starting it');
     }
+    // the mutex is released only after a successful commit: when a commit
+    // fails the transaction is still active, and the caller's rollback (e.g.
+    // by EntityManager.transaction) is what ends it and releases the mutex
     await super.commitTransaction();
     this.releaseMutex();
     this.releaseMutex = null;
@@ -30,9 +40,14 @@ class CustomQueryRunner extends SqliteQueryRunner {
     if (!this.releaseMutex) {
       throw new Error('Cannot rollback transaction before starting it');
     }
-    await super.rollbackTransaction();
-    this.releaseMutex();
-    this.releaseMutex = null;
+    try {
+      await super.rollbackTransaction();
+    } finally {
+      // the transaction is over either way; keeping the mutex after a failed
+      // rollback would block every later transaction on this runner forever
+      this.releaseMutex();
+      this.releaseMutex = null;
+    }
   };
 }
 

packages/extended-typeorm/tests/customQueryRunner.spec.ts

@@ -1,6 +1,39 @@
+import { CustomQueryRunner } from '../lib/customQueryRunner';
 import { createAndInitializeDataSource } from './customQueryRunnerTestUtils';
 
 describe('CustomQueryRunner', () => {
+  describe('startTransaction', () => {
+    /**
+     * @target CustomQueryRunner.startTransaction should release the mutex when starting a transaction fails
+     * @dependencies
+     * @scenario
+     * - create a datasource
+     * - create a query runner and connect to it
+     * - call startTransaction with an isolation level SQLite does not support, so it throws after the mutex was acquired
+     * - check the mutex state
+     * - start and commit a valid transaction
+     * @expected
+     * - the failed start should throw
+     * - the mutex should not be locked and no releaser should be kept after the failure
+     * - the valid transaction should start and commit successfully
+     */
+    it('should release the mutex when starting a transaction fails', async () => {
+      const dataSource = await createAndInitializeDataSource();
+      const queryRunner = dataSource.createQueryRunner() as CustomQueryRunner;
+      await queryRunner.connect();
+      await expect(
+        queryRunner.startTransaction('READ COMMITTED'),
+      ).rejects.toThrow();
+      expect(queryRunner.releaseMutex).toEqual(null);
+      expect(queryRunner.mutex.isLocked()).toEqual(false);
+
+      await queryRunner.startTransaction();
+      await queryRunner.commitTransaction();
+      expect(queryRunner.releaseMutex).toEqual(null);
+      expect(queryRunner.mutex.isLocked()).toEqual(false);
+    });
+  });
+
   describe('commitTransaction', () => {
     /**
      * @target CustomQueryRunner.commitTransaction should throw Error when no transaction started
@@ -37,5 +70,30 @@ describe('CustomQueryRunner', () => {
       await queryRunner.connect();
       await expect(() => queryRunner.rollbackTransaction()).rejects.toThrow();
     });
+
+    /**
+     * @target CustomQueryRunner.rollbackTransaction should release the mutex when the rollback fails
+     * @dependencies
+     * @scenario
+     * - create a datasource
+     * - create a query runner and connect to it
+     * - start a transaction
+     * - end the transaction with a raw COMMIT so the rollback fails in SQLite
+     * - call rollbackTransaction
+     * - check the mutex state
+     * @expected
+     * - the rollback should throw
+     * - the mutex should not be locked and no releaser should be kept after the failure
+     */
+    it('should release the mutex when the rollback fails', async () => {
+      const dataSource = await createAndInitializeDataSource();
+      const queryRunner = dataSource.createQueryRunner() as CustomQueryRunner;
+      await queryRunner.connect();
+      await queryRunner.startTransaction();
+      await queryRunner.query('COMMIT');
+      await expect(queryRunner.rollbackTransaction()).rejects.toThrow();
+      expect(queryRunner.releaseMutex).toEqual(null);
+      expect(queryRunner.mutex.isLocked()).toEqual(false);
+    });
   });
 });

packages/tokens/lib/tokenMap.ts

@@ -80,7 +80,10 @@ export class TokenMap {
    * @param tokens
    */
   updateConfigByJson = async (tokens: RosenTokens) => {
-    await this.updateSemaphore.acquire().then(async (release) => {
+    // `use` releases the semaphore on every path: releasing only on success
+    // would block every later update forever after a single failure (e.g. a
+    // corrupted config or a throwing callback)
+    await this.updateSemaphore.use(async () => {
       const newTokenConfig: RosenTokens = [];
       const newUnbridgeableTokens: RosenTokens = [];
       tokens.forEach((tokenSet) => {
@@ -101,7 +104,6 @@ export class TokenMap {
       this.tokensConfig = newTokenConfig;
       this.unbridgeableTokens = newUnbridgeableTokens;
       for (const callback of this.callbacks.values()) callback();
-      release();
     });
   };
 

packages/tokens/tests/tokenMap.spec.ts

@@ -65,6 +65,67 @@ describe('TokenMap', () => {
       }).rejects.toThrow(CorruptedConfigError);
       expect(mockedCallback).not.toHaveBeenCalled();
     });
+
+    /**
+     * @target TokenMap.updateConfigByJson should accept a new config after a corrupted config update failed
+     * @dependencies
+     * - RosenToken json
+     * @scenario
+     * - call updateConfigByJson with a corrupted config and check it throws
+     * - call updateConfigByJson with a valid config
+     * @expected
+     * - the valid update should complete (the failed update must not keep the semaphore)
+     * - the config should equal the valid config
+     */
+    it('should accept a new config after a corrupted config update failed', async () => {
+      const tokenMap = new TokenMap();
+
+      await expect(async () => {
+        await tokenMap.updateConfigByJson([
+          ...firstTokenMapWithUnbridgeableTokens,
+          invalidTokenSet,
+        ]);
+      }).rejects.toThrow(CorruptedConfigError);
+
+      const updated = await Promise.race([
+        tokenMap.updateConfigByJson(firstTokenMap).then(() => true),
+        new Promise<false>((resolve) => setTimeout(() => resolve(false), 1000)),
+      ]);
+      expect(updated).toEqual(true);
+      expect(tokenMap.getConfig()).toEqual(firstTokenMap);
+    });
+
+    /**
+     * @target TokenMap.updateConfigByJson should accept a new config after a registered callback throws
+     * @dependencies
+     * - RosenToken json
+     * @scenario
+     * - register a callback that throws
+     * - call updateConfigByJson and check it throws
+     * - unregister the callback
+     * - call updateConfigByJson again
+     * @expected
+     * - the second update should complete (the throwing callback must not keep the semaphore)
+     * - the config should equal the valid config
+     */
+    it('should accept a new config after a registered callback throws', async () => {
+      const tokenMap = new TokenMap();
+      const callbackId = tokenMap.registerCallback(() => {
+        throw new Error('callback failed');
+      });
+
+      await expect(async () => {
+        await tokenMap.updateConfigByJson(firstTokenMap);
+      }).rejects.toThrow('callback failed');
+
+      tokenMap.unregisterCallback(callbackId);
+      const updated = await Promise.race([
+        tokenMap.updateConfigByJson(firstTokenMap).then(() => true),
+        new Promise<false>((resolve) => setTimeout(() => resolve(false), 1000)),
+      ]);
+      expect(updated).toEqual(true);
+      expect(tokenMap.getConfig()).toEqual(firstTokenMap);
+    });
   });
 
   describe('search', () => {