Merge pull request #53 from MatrixAI/feature-dangling-transactions

Fix committing or rollbacking dangling transactions

Author: CMCDragonkai

src/DB.ts

@@ -168,7 +168,15 @@ class DB {
       await iterator.destroy();
     }
     for (const transaction of this._transactionRefs) {
-      await transaction.rollback();
+      if (!transaction.committing && !transaction.rollbacking) {
+        // If any transactions is still pending at this point
+        // then if they try to commit, that will be an error because
+        // the transaction is already rollbacked
+        await transaction.rollback();
+      } else {
+        // This will wait for committing or rollbacking to complete
+        await transaction.destroy();
+      }
     }
     await rocksdbP.dbClose(this._db);
     this.logger.info(`Stopped ${this.constructor.name}`);
@@ -212,6 +220,9 @@ class DB {
               await tran.rollback(e);
             }
           } finally {
+            // If already destroyed, this is a noop
+            // this will only have affect if there was an
+            // exception during commit or rollback
             await tran.destroy();
           }
         },

src/DBTransaction.ts

@@ -20,7 +20,7 @@ import type {
 } from './native/types';
 import Logger from '@matrixai/logger';
 import { CreateDestroy, ready } from '@matrixai/async-init/dist/CreateDestroy';
-import { RWLockWriter } from '@matrixai/async-locks';
+import { Lock, RWLockWriter } from '@matrixai/async-locks';
 import DBIterator from './DBIterator';
 import { rocksdbP } from './native';
 import * as utils from './utils';
@@ -49,8 +49,11 @@ class DBTransaction {
   protected _callbacksSuccess: Array<() => any> = [];
   protected _callbacksFailure: Array<(e?: Error) => any> = [];
   protected _callbacksFinally: Array<(e?: Error) => any> = [];
+  protected _committing: boolean = false;
   protected _committed: boolean = false;
+  protected _rollbacking: boolean = false;
   protected _rollbacked: boolean = false;
+  protected commitOrRollbackLock: Lock = new Lock();
 
   public constructor({
     db,
@@ -86,9 +89,12 @@ class DBTransaction {
    */
   public async destroy() {
     this.logger.debug(`Destroying ${this.constructor.name} ${this.id}`);
-    if (!this._committed && !this._rollbacked) {
+    if (!this._committing && !this._rollbacking) {
       throw new errors.ErrorDBTransactionNotCommittedNorRollbacked();
     }
+    // Wait for commit or rollback to finish
+    // this then allows the destruction to proceed
+    await this.commitOrRollbackLock.waitForUnlock();
     this._db.transactionRefs.delete(this);
     // Unlock all locked keys in reverse
     const lockedKeys = [...this._locks.keys()].reverse();
@@ -116,10 +122,30 @@ class DBTransaction {
     return this._callbacksFinally;
   }
 
+  /**
+   * Indicates when `this.commit` is first called
+   */
+  get committing(): boolean {
+    return this._committing;
+  }
+
+  /**
+   * Indicates when the transaction is committed
+   */
   get committed(): boolean {
     return this._committed;
   }
 
+  /**
+   * Indicates when `this.rollback` is first called
+   */
+  get rollbacking(): boolean {
+    return this._rollbacking;
+  }
+
+  /**
+   * Indicates when the transaction is rollbacked
+   */
   get rollbacked(): boolean {
     return this._rollbacked;
   }
@@ -437,75 +463,79 @@ class DBTransaction {
 
   @ready(new errors.ErrorDBTransactionDestroyed())
   public async commit(): Promise<void> {
-    if (this._rollbacked) {
+    if (this._rollbacking) {
       throw new errors.ErrorDBTransactionRollbacked();
     }
-    if (this._committed) {
+    if (this._committing) {
       return;
     }
+    this._committing = true;
     this.logger.debug(`Committing ${this.constructor.name} ${this.id}`);
-    for (const iterator of this._iteratorRefs) {
-      await iterator.destroy();
-    }
-    this._committed = true;
-    try {
+    await this.commitOrRollbackLock.withF(async () => {
+      for (const iterator of this._iteratorRefs) {
+        await iterator.destroy();
+      }
       try {
-        // If this fails, the `DBTransaction` is still considered committed
-        // it must be destroyed, it cannot be reused
-        await rocksdbP.transactionCommit(this._transaction);
-      } catch (e) {
-        if (e.code === 'TRANSACTION_CONFLICT') {
-          this.logger.debug(
-            `Failed Committing ${this.constructor.name} ${this.id} due to ${errors.ErrorDBTransactionConflict.name}`,
-          );
-          throw new errors.ErrorDBTransactionConflict(undefined, {
-            cause: e,
-          });
-        } else {
-          this.logger.debug(
-            `Failed Committing ${this.constructor.name} ${this.id} due to ${e.message}`,
-          );
-          throw e;
+        try {
+          // If this fails, the `DBTransaction` is still considered committed
+          // it must be destroyed, it cannot be reused
+          await rocksdbP.transactionCommit(this._transaction);
+        } catch (e) {
+          if (e.code === 'TRANSACTION_CONFLICT') {
+            this.logger.debug(
+              `Failed Committing ${this.constructor.name} ${this.id} due to ${errors.ErrorDBTransactionConflict.name}`,
+            );
+            throw new errors.ErrorDBTransactionConflict(undefined, {
+              cause: e,
+            });
+          } else {
+            this.logger.debug(
+              `Failed Committing ${this.constructor.name} ${this.id} due to ${e.message}`,
+            );
+            throw e;
+          }
+        }
+        for (const f of this._callbacksSuccess) {
+          await f();
+        }
+      } finally {
+        for (const f of this._callbacksFinally) {
+          await f();
         }
       }
-      for (const f of this._callbacksSuccess) {
-        await f();
-      }
-    } finally {
-      for (const f of this._callbacksFinally) {
-        await f();
-      }
-    }
-    await this.destroy();
+      this._committed = true;
+    });
     this.logger.debug(`Committed ${this.constructor.name} ${this.id}`);
   }
 
   @ready(new errors.ErrorDBTransactionDestroyed())
   public async rollback(e?: Error): Promise<void> {
-    if (this._committed) {
+    if (this._committing) {
       throw new errors.ErrorDBTransactionCommitted();
     }
-    if (this._rollbacked) {
+    if (this._rollbacking) {
       return;
     }
+    this._rollbacking = true;
     this.logger.debug(`Rollbacking ${this.constructor.name} ${this.id}`);
-    for (const iterator of this._iteratorRefs) {
-      await iterator.destroy();
-    }
-    this._rollbacked = true;
-    try {
-      // If this fails, the `DBTransaction` is still considered rollbacked
-      // it must be destroyed, it cannot be reused
-      await rocksdbP.transactionRollback(this._transaction);
-      for (const f of this._callbacksFailure) {
-        await f(e);
+    await this.commitOrRollbackLock.withF(async () => {
+      for (const iterator of this._iteratorRefs) {
+        await iterator.destroy();
       }
-    } finally {
-      for (const f of this._callbacksFinally) {
-        await f(e);
+      try {
+        // If this fails, the `DBTransaction` is still considered rollbacked
+        // it must be destroyed, it cannot be reused
+        await rocksdbP.transactionRollback(this._transaction);
+        for (const f of this._callbacksFailure) {
+          await f(e);
+        }
+      } finally {
+        for (const f of this._callbacksFinally) {
+          await f(e);
+        }
       }
-    }
-    await this.destroy();
+      this._rollbacked = true;
+    });
     this.logger.debug(`Rollbacked ${this.constructor.name} ${this.id}`);
   }
 

tests/DB.test.ts

@@ -1,4 +1,6 @@
+import type DBTransaction from '@/DBTransaction';
 import type { KeyPath } from '@/types';
+import type { ResourceRelease } from '@matrixai/resources';
 import type { DBWorkerModule } from './workers/dbWorkerModule';
 import os from 'os';
 import path from 'path';
@@ -569,4 +571,68 @@ describe(DB.name, () => {
     ]);
     await db.stop();
   });
+  test('dangling transactions are automatically rollbacked', async () => {
+    const dbPath = `${dataDir}/db`;
+    let db = await DB.createDB({ dbPath, crypto, logger });
+    // This is a pending transaction to be committed
+    // however it hasn't started committing until it's already rollbacked
+    const p = expect(
+      db.withTransactionF(async (tran) => {
+        await tran.put('foo', 'bar');
+        expect(tran.committing).toBe(false);
+      }),
+    ).rejects.toThrow(errors.ErrorDBTransactionRollbacked);
+    await db.stop();
+    await p;
+    db = await DB.createDB({ dbPath, crypto, logger });
+    const acquireTran = db.transaction();
+    const [releaseTran, tran] = (await acquireTran()) as [
+      ResourceRelease,
+      DBTransaction,
+    ];
+    await tran.put('foo', 'bar');
+    expect(tran.rollbacking).toBe(false);
+    expect(tran.rollbacked).toBe(false);
+    await db.stop();
+    expect(tran.rollbacking).toBe(true);
+    expect(tran.rollbacked).toBe(true);
+    await expect(releaseTran()).rejects.toThrow(
+      errors.ErrorDBTransactionRollbacked,
+    );
+  });
+  test('dangling committing transactions are waited for', async () => {
+    const dbPath = `${dataDir}/db`;
+    const db = await DB.createDB({ dbPath, crypto, logger });
+    const acquireTran = db.transaction();
+    const [releaseTran, tran] = (await acquireTran()) as [
+      ResourceRelease,
+      DBTransaction,
+    ];
+    await tran.put('foo', 'bar');
+    const p = releaseTran();
+    expect(tran.committing).toBe(true);
+    expect(tran.committed).toBe(false);
+    // This will wait for the transaction to be committed
+    await db.stop();
+    await p;
+    expect(tran.committing).toBe(true);
+    expect(tran.committed).toBe(true);
+  });
+  test('dangling rollbacking transactions are waited for', async () => {
+    const dbPath = `${dataDir}/db`;
+    const db = await DB.createDB({ dbPath, crypto, logger });
+    const acquireTran = db.transaction();
+    const [releaseTran, tran] = (await acquireTran()) as [
+      ResourceRelease,
+      DBTransaction,
+    ];
+    await tran.put('foo', 'bar');
+    const p = releaseTran(new Error('Trigger Rollback'));
+    expect(tran.rollbacking).toBe(true);
+    expect(tran.rollbacked).toBe(false);
+    await db.stop();
+    await p;
+    expect(tran.rollbacking).toBe(true);
+    expect(tran.rollbacked).toBe(true);
+  });
 });

tests/DBIterator.test.ts

@@ -1,4 +1,4 @@
-import type { KeyPath } from '@';
+import type { KeyPath } from '@/types';
 import os from 'os';
 import path from 'path';
 import fs from 'fs';