WIP

Author: CMCDragonkai

src/DBTransaction.ts

@@ -1,5 +1,5 @@
 import type { ResourceRelease } from '@matrixai/resources';
-import type { LockBox, LockRequest as AsyncLocksLockRequest } from '@matrixai/async-locks';
+import type { LockBox, MultiLockRequest as AsyncLocksMultiLockRequest } from '@matrixai/async-locks';
 import type DB from './DB';
 import type {
   ToString,
@@ -8,7 +8,7 @@ import type {
   DBIteratorOptions,
   DBClearOptions,
   DBCountOptions,
-  LockRequest,
+  MultiLockRequest,
 } from './types';
 import type {
   RocksDBTransaction,
@@ -31,7 +31,11 @@ class DBTransaction {
   protected _db: DB;
   protected logger: Logger;
   protected lockBox: LockBox<RWLockWriter>;
-  protected _locks: Map<string, [ResourceRelease, RWLockWriter]> = new Map();
+  protected _locks: Map<string, {
+    lock: RWLockWriter,
+    type: 'read' | 'write',
+    release: ResourceRelease,
+  }> = new Map();
   protected _options: RocksDBTransactionOptions;
   protected _transaction: RocksDBTransaction;
   protected _snapshot: RocksDBTransactionSnapshot;
@@ -114,7 +118,11 @@ class DBTransaction {
     return this._rollbacked;
   }
 
-  get locks(): ReadonlyMap<string, [ResourceRelease, RWLockWriter]> {
+  get locks(): ReadonlyMap<string, {
+    lock: RWLockWriter,
+    type: 'read' | 'write',
+    release: ResourceRelease,
+  }> {
     return this._locks;
   }
 
@@ -127,41 +135,55 @@ class DBTransaction {
 
   /**
    * Lock a sequence of lock requests
-   * If a lock request is not an array, it is converted to `[ToString, RWLockWriter, 'write']`
+   * If the lock request doesn't specify, it
+   * defaults to using `RWLockWriter` with `write` type
    * Keys are locked in string sorted order
    * Even though keys can be arbitrary strings, by convention, you should use
    * keys that correspond to keys in the database
    * Locking with the same key is idempotent therefore lock re-entrancy is enabled
-   * There is no deadlock detection, so becareful!
    * Keys are automatically unlocked in reverse sorted order
    * when the transaction is destroyed
+   * There is no support for lock upgrading or downgrading
+   * There is no deadlock detection
    */
-  public async lock(...requests: Array<LockRequest | string>): Promise<void> {
-    const requests_: Array<AsyncLocksLockRequest<RWLockWriter>> = [];
+  public async lock(...requests: Array<MultiLockRequest | string>): Promise<void> {
+    const requests_: Array<AsyncLocksMultiLockRequest<RWLockWriter>> = [];
     for (const request of requests) {
       if (Array.isArray(request)) {
-        const [key, ...rest] = request;
+        const [key, ...lockingParams] = request;
         const key_ = key.toString();
-        if (!this._locks.has(key_)) {
-          requests_.push([key_, RWLockWriter, ...rest]);
+        const lock = this._locks.get(key_);
+        // Default the lock type to `write`
+        const lockType = lockingParams[0] = lockingParams[0] ?? 'write';
+        if (lock == null) {
+          requests_.push([key_, RWLockWriter, ...lockingParams]);
+        } else if (lock.type !== lockType) {
+          throw new errors.ErrorDBTransactionLockType();
         }
       } else {
         const key_ = request.toString();
-        if (!this._locks.has(key_)) {
+        const lock = this._locks.get(key_);
+        if (lock == null) {
           // Default to using `RWLockWriter` write lock for just string keys
           requests_.push([key_, RWLockWriter, 'write']);
+        } else if (lock.type !== 'write') {
+          throw new errors.ErrorDBTransactionLockType();
         }
       }
     }
     if (requests_.length > 0) {
       // Duplicates are eliminated, and the returned acquisitions are sorted
       const lockAcquires = this.lockBox.lockMulti(...requests_);
-      for (const [key, lockAcquire] of lockAcquires) {
+      for (const [key, lockAcquire, ...lockingParams] of lockAcquires) {
         const [lockRelease, lock] = await lockAcquire();
         // The `Map` will maintain insertion order
         // these must be unlocked in reverse order
         // when the transaction is destroyed
-        this._locks.set(key as string, [lockRelease, lock!]);
+        this._locks.set(key as string, {
+          lock: lock!,
+          type: lockingParams[0]!, // The `type` is defaulted to `write`
+          release: lockRelease,
+        });
       }
     }
   }
@@ -178,9 +200,8 @@ class DBTransaction {
       const key_ = key.toString();
       const lock = this._locks.get(key_);
       if (lock == null) continue;
-      const [lockRelease] = lock;
       this._locks.delete(key_);
-      await lockRelease();
+      await lock.release();
     }
   }
 
@@ -426,7 +447,9 @@ class DBTransaction {
           this.logger.debug(
             `Failed Committing ${this.constructor.name} ${this.id} due to ${errors.ErrorDBTransactionConflict.name}`,
           );
-          throw new errors.ErrorDBTransactionConflict(undefined, { cause: e });
+          throw new errors.ErrorDBTransactionConflict(undefined, {
+            cause: e,
+          });
         } else {
           this.logger.debug(
             `Failed Committing ${this.constructor.name} ${this.id} due to ${e.message}`,

src/errors.ts

@@ -82,6 +82,10 @@ class ErrorDBTransactionConflict<T> extends ErrorDBTransaction<T> {
   static description = 'DBTransaction cannot commit due to conflicting writes';
 }
 
+class ErrorDBTransactionLockType<T> extends ErrorDBTransaction<T> {
+  static description = 'DBTransaction does not support upgrading or downgrading the lock type';
+}
+
 export {
   ErrorDB,
   ErrorDBRunning,
@@ -103,4 +107,5 @@ export {
   ErrorDBTransactionRollbacked,
   ErrorDBTransactionNotCommittedNorRollbacked,
   ErrorDBTransactionConflict,
+  ErrorDBTransactionLockType,
 };

src/types.ts

@@ -159,7 +159,10 @@ type DBOp =
 
 type DBOps = Array<DBOp>;
 
-type LockRequest = [key: ToString, ...lockingParams: Parameters<RWLockWriter['lock']>];
+type MultiLockRequest = [
+  key: ToString,
+  ...lockingParams: Parameters<RWLockWriter['lock']>
+];
 
 export type {
   POJO,
@@ -179,5 +182,5 @@ export type {
   DBBatch,
   DBOp,
   DBOps,
-  LockRequest
+  MultiLockRequest
 };

tests/DBTransaction.test.ts

@@ -1006,6 +1006,74 @@ describe(DBTransaction.name, () => {
       await tran.unlock('bar', 'foo');
       expect(tran.locks.size).toBe(0);
     });
+    // Duplicates are eliminated
+    await db.withTransactionF(async (tran) => {
+      await tran.lock('foo', 'foo');
+      expect(tran.locks.size).toBe(1);
+      await tran.unlock('foo', 'foo');
+      expect(tran.locks.size).toBe(0);
+    });
+  });
+  test('read and write locking', async () => {
+    await db.withTransactionF(async (tran1) => {
+      await tran1.lock(['foo', 'read']);
+      await tran1.lock(['bar', 'write']);
+      // There is no automatic lock upgrade or downgrade
+      await expect(tran1.lock(['foo', 'write'])).rejects.toThrow(errors.ErrorDBTransactionLockType);
+      await expect(tran1.lock(['bar', 'read'])).rejects.toThrow(errors.ErrorDBTransactionLockType);
+      await db.withTransactionF(async (tran2) => {
+        await tran2.lock(['foo', 'read']);
+        await expect(tran2.lock(['bar', 'write', 0])).rejects.toThrow(locksErrors.ErrorAsyncLocksTimeout);
+        expect(tran1.locks.size).toBe(2);
+        expect(tran1.locks.has('foo')).toBe(true);
+        expect(tran1.locks.get('foo')!.type).toBe('read');
+        expect(tran2.locks.size).toBe(1);
+        expect(tran2.locks.has('foo')).toBe(true);
+        expect(tran2.locks.get('foo')!.type).toBe('read');
+      });
+      expect(tran1.locks.size).toBe(2);
+      await tran1.unlock('bar');
+      await db.withTransactionF(async (tran2) => {
+        await tran2.lock(['foo', 'read']);
+        await tran2.lock(['bar', 'write']);
+        expect(tran1.locks.size).toBe(1);
+        expect(tran1.locks.has('foo')).toBe(true);
+        expect(tran1.locks.get('foo')!.type).toBe('read');
+        expect(tran2.locks.size).toBe(2);
+        expect(tran2.locks.has('foo')).toBe(true);
+        expect(tran2.locks.get('foo')!.type).toBe('read');
+        expect(tran2.locks.has('bar')).toBe(true);
+        expect(tran2.locks.get('bar')!.type).toBe('write');
+      });
+    });
+  });
+  test('locks are unlocked in reverse order', async () => {
+    let order: Array<string> = [];
+    let p1, p2;
+    await db.withTransactionF(async (tran) => {
+      // '1' and '2' are in sort order
+      await tran.lock('1');
+      await tran.lock('2');
+      p1 = db.withTransactionF(async (tran) => {
+        await tran.lock('1');
+        order.push('1');
+      });
+      p2 = db.withTransactionF(async (tran) => {
+        await tran.lock('2');
+        order.push('2');
+      });
+    });
+    await Promise.all([p2, p1]);
+    expect(order).toStrictEqual(['2', '1']);
+  });
+  test('lock re-entrancy', async () => {
+    await db.withTransactionF(async (tran) => {
+      // Locking with the same keys is idempotent
+      await tran.lock('key1', 'key2');
+      await tran.lock('key1', 'key2');
+      await tran.lock('key1');
+      await tran.lock('key2');
+    });
   });
   test('locks are isolated per transaction', async () => {
     await db.withTransactionF(async (tran1) => {
@@ -1034,13 +1102,22 @@ describe(DBTransaction.name, () => {
       expect(tran1.locks.has('key2')).toBe(true);
     });
   });
-  test('transactions support lock re-entrancy', async () => {
-    await db.withTransactionF(async (tran) => {
-      // Locking with the same keys is idempotent
-      await tran.lock('key1', 'key2');
-      await tran.lock('key1', 'key2');
-      await tran.lock('key1');
-      await tran.lock('key2');
+  test('deadlock', async () => {
+    await db.withTransactionF(async (tran1) => {
+      await db.withTransactionF(async (tran2) => {
+        await tran1.lock('foo');
+        await tran2.lock('bar');
+        // Currently a deadlock can happen, and the only way to avoid is to use timeouts
+        // In the future, we want to have `DBTransaction` detect deadlocks
+        // and automatically give us `ErrorDBTransactionDeadlock` exception
+        const p1 = tran1.lock(['bar', 'write', 50]);
+        const p2 = tran2.lock(['foo', 'write', 50]);
+        const results = await Promise.allSettled([p1, p2]);
+        expect(results.every(r =>
+          r.status === 'rejected' &&
+          r.reason instanceof locksErrors.ErrorAsyncLocksTimeout)
+        ).toBe(true);
+      });
     });
   });
 });