Merge pull request #112 from pbbadenhorst/fix/serialize-multiple-transactions

Author: Oscar Franco

example/src/tests/rawQueries.spec.ts

@@ -288,13 +288,13 @@ export function registerBaseTests() {
       });
     });
 
-    it('Async transaction, auto commit', done => {
+    it('Async transaction, auto commit', async () => {
       const id = chance.integer();
       const name = chance.name();
       const age = chance.integer();
       const networth = chance.floating();
 
-      db.transactionAsync(async tx => {
+      await db.transactionAsync(async tx => {
         const res = await tx.executeAsync(
           'INSERT INTO "User" (id, name, age, networth) VALUES(?, ?, ?, ?)',
           [id, name, age, networth],
@@ -308,87 +308,156 @@ export function registerBaseTests() {
         expect(res.rows.item).to.be.a('function');
       });
 
-      setTimeout(() => {
-        const res = db.execute('SELECT * FROM User');
-        expect(res.rows?._array).to.eql([
-          {
-            id,
-            name,
-            age,
-            networth,
-          },
-        ]);
-        done();
-      }, 200);
+      const res = db.execute('SELECT * FROM User');
+      expect(res.rows?._array).to.eql([
+        {
+          id,
+          name,
+          age,
+          networth,
+        },
+      ]);
     });
 
-    it('Async transaction, auto rollback', done => {
-      const id = chance.string();
+    it('Async transaction, auto rollback', async () => {
+      const id = chance.string(); // Causes error because it should be an integer
       const name = chance.name();
       const age = chance.integer();
       const networth = chance.floating();
 
-      db.transactionAsync(async tx => {
-        await tx.executeAsync(
-          'INSERT INTO "User" (id, name, age, networth) VALUES(?, ?, ?, ?)',
-          [id, name, age, networth],
-        );
-      });
+      try {
+        await db.transactionAsync(async tx => {
+          await tx.executeAsync(
+            'INSERT INTO "User" (id, name, age, networth) VALUES(?, ?, ?, ?)',
+            [id, name, age, networth],
+          );
+        });
+      } catch (error) {
+        expect(error).to.be.instanceOf(Error)
+        expect((error as Error).message)
+          .to.include('SQL execution error')
+          .and
+          .to.include('cannot store TEXT value in INT column User.id');
 
-      setTimeout(() => {
         const res = db.execute('SELECT * FROM User');
         expect(res.rows?._array).to.eql([]);
-        done();
-      }, 200);
+      }
     });
 
-    it('Async transaction, manual commit', done => {
+    it('Async transaction, manual commit', async () => {
       const id = chance.integer();
       const name = chance.name();
       const age = chance.integer();
       const networth = chance.floating();
 
-      db.transactionAsync(async tx => {
+      await db.transactionAsync(async tx => {
         await tx.executeAsync(
           'INSERT INTO "User" (id, name, age, networth) VALUES(?, ?, ?, ?)',
           [id, name, age, networth],
         );
         tx.commit();
       });
 
-      setTimeout(() => {
-        const res = db.execute('SELECT * FROM User');
-        expect(res.rows?._array).to.eql([
-          {
-            id,
-            name,
-            age,
-            networth,
-          },
-        ]);
-        done();
-      }, 1000);
+      const res = db.execute('SELECT * FROM User');
+      expect(res.rows?._array).to.eql([
+        {
+          id,
+          name,
+          age,
+          networth,
+        },
+      ]);
     });
 
-    it('Async transaction, manual rollback', done => {
+    it('Async transaction, manual rollback', async () => {
       const id = chance.integer();
       const name = chance.name();
       const age = chance.integer();
       const networth = chance.floating();
 
-      db.transactionAsync(async tx => {
+      await db.transactionAsync(async tx => {
         await tx.executeAsync(
           'INSERT INTO "User" (id, name, age, networth) VALUES(?, ?, ?, ?)',
           [id, name, age, networth],
         );
         tx.rollback();
       });
 
-      setTimeout(() => {
-        const res = db.execute('SELECT * FROM User');
-        expect(res.rows?._array).to.eql([]);
-        done();
-      }, 1000);
+      const res = db.execute('SELECT * FROM User');
+      expect(res.rows?._array).to.eql([]);
+    });
+
+    it('Async transaction, upsert and select', async () => {
+      // ARRANGE: Setup for multiple transactions
+      const iterations = 10;
+      const actual = new Set();
+
+      // ARRANGE: Generate expected data
+      const id = chance.integer();
+      const name = chance.name();
+      const age = chance.integer();
+
+      // ACT: Start multiple async transactions to upsert and select the same record
+      const promises = [];
+      for (let iteration = 1; iteration <= iterations; iteration++) {
+        const promised = db.transactionAsync(async (tx) => {
+          // ACT: Upsert statement to create record / increment the value
+          await tx.executeAsync(`
+          INSERT OR REPLACE INTO [User] ([id], [name], [age], [networth])
+          SELECT ?, ?, ?,
+            IFNULL((
+              SELECT [networth] + 1000
+              FROM [User]
+              WHERE [id] = ?
+            ), 1000)
+          `, [id, name, age, id]);
+
+          // ACT: Select statement to get incremented value and store it for checking later
+          const results = await tx.executeAsync('SELECT [networth] FROM [User] WHERE [id] = ?', [id]);
+
+          actual.add(results.rows._array[0].networth);
+        })
+
+        promises.push(promised);
+      }
+
+      // ACT: Wait for all transactions to complete
+      await Promise.all(promises);
+
+      // ASSERT: That the expected values where returned
+      expect(actual.size).to.equal(iterations, 'Each transaction should read a different value');
+    });
+
+    it('Async transaction, rejects on callback error', async () => {
+      const promised = db.transactionAsync(async (tx) => {
+        throw new Error('Error from callback');
+      });
+
+      // ASSERT: should return a promise that eventually rejects
+      expect(promised).to.have.property('then').that.is.a('function');
+      try {
+        await promised;
+        expect.fail('Should not resolve');
+      } catch (e) {
+        expect(e).to.be.a.instanceof(Error);
+        expect((e as Error)?.message).to.equal('Error from callback');
+      }
+    });
+
+    it('Async transaction, rejects on invalid query', async () => {
+      const promised = db.transactionAsync(async (tx) => {
+        await tx.executeAsync('SELECT * FROM [tableThatDoesNotExist];');
+      })
+
+      // ASSERT: should return a promise that eventually rejects
+      expect(promised).to.have.property('then').that.is.a('function');
+      try {
+        await promised;
+        expect.fail('Should not resolve');
+      } catch (e) {
+        expect(e).to.be.a.instanceof(Error);
+        expect((e as Error)?.message).to.include('no such table: tableThatDoesNotExist');
+      }
     });
 
     it('Batch execute', () => {

src/index.ts

@@ -140,7 +140,7 @@ interface ISQLite {
   transactionAsync: (
     dbName: string,
     fn: (tx: TransactionAsync) => Promise<any>
-  ) => void;
+  ) => Promise<void>;
   transaction: (dbName: string, fn: (tx: Transaction) => void) => void;
   execute: (dbName: string, query: string, params?: any[]) => QueryResult;
   executeAsync: (
@@ -240,12 +240,22 @@ QuickSQLite.transaction = (
   };
 
   const commit = () => {
+    if (isFinalized) {
+      throw Error(
+        `Quick SQLite Error: Cannot execute commit on finalized transaction: ${dbName}`
+      );
+    }
     const result = QuickSQLite.execute(dbName, 'COMMIT');
     isFinalized = true;
     return result;
   };
 
   const rollback = () => {
+    if (isFinalized) {
+      throw Error(
+        `Quick SQLite Error: Cannot execute rollback on finalized transaction: ${dbName}`
+      );
+    }
     const result = QuickSQLite.execute(dbName, 'ROLLBACK');
     isFinalized = true;
     return result;
@@ -276,7 +286,7 @@ QuickSQLite.transaction = (
   startNextTransaction(dbName);
 };
 
-QuickSQLite.transactionAsync = (
+QuickSQLite.transactionAsync = async (
   dbName: string,
   callback: (tx: TransactionAsync) => Promise<any>
 ) => {
@@ -306,69 +316,84 @@ QuickSQLite.transactionAsync = (
   };
 
   const commit = () => {
+    if (isFinalized) {
+      throw Error(
+        `Quick SQLite Error: Cannot execute commit on finalized transaction: ${dbName}`
+      );
+    }
     const result = QuickSQLite.execute(dbName, 'COMMIT');
     isFinalized = true;
     return result;
   };
 
   const rollback = () => {
+    if (isFinalized) {
+      throw Error(
+        `Quick SQLite Error: Cannot execute rollback on finalized transaction: ${dbName}`
+      );
+    }
     const result = QuickSQLite.execute(dbName, 'ROLLBACK');
     isFinalized = true;
     return result;
   };
 
-  const tx: PendingTransaction = {
-    start: async () => {
-      try {
-        QuickSQLite.execute(dbName, 'BEGIN TRANSACTION');
-        await callback({
-          commit,
-          execute,
-          executeAsync,
-          rollback,
-        });
-
-        if (!isFinalized) {
-          commit();
-        }
-      } catch (e: any) {
-        if (!isFinalized) {
-          rollback();
-        }
+  return await new Promise((resolve, reject) => {
+    const tx: PendingTransaction = {
+      start: async () => {
+        try {
+          QuickSQLite.execute(dbName, 'BEGIN TRANSACTION');
+          await callback({
+            commit,
+            execute,
+            executeAsync,
+            rollback,
+          });
+
+          if (!isFinalized) {
+            commit();
+          }
 
-        // Do not throw an error, because the transaction is executed with a setImmediate call
-        // This errors are uncatchable
-        // https://stackoverflow.com/questions/51081892/nodejs-asynchronous-exceptions-are-uncatchable
+          resolve();
+        } catch (e) {
+          if (!isFinalized) {
+            try {
+              rollback();
+            } catch (rollbackError) {
+              reject(rollbackError);
+            }
+          }
 
-        // throw e;
-      } finally {
-        locks[dbName].inProgress = false;
-        isFinalized = false;
-        startNextTransaction(dbName);
-      }
-    },
-  };
+          reject(e);
+        } finally {
+          locks[dbName].inProgress = false;
+          isFinalized = false;
+          startNextTransaction(dbName);
+        }
+      },
+    };
 
-  locks[dbName].queue.push(tx);
-  startNextTransaction(dbName);
+    locks[dbName].queue.push(tx);
+    startNextTransaction(dbName);
+  });
 };
 
 const startNextTransaction = (dbName: string) => {
+  if (!locks[dbName]) {
+    throw Error(`Lock not found for db: ${dbName}`);
+  }
+
   if (locks[dbName].inProgress) {
     // Transaction is already in process bail out
     return;
   }
 
-  setImmediate(() => {
-    if (!locks[dbName]) {
-      throw Error(`Lock not found for db: ${dbName}`);
-    }
-
-    if (locks[dbName].queue.length) {
-      locks[dbName].inProgress = true;
-      locks[dbName].queue.shift().start();
-    }
-  });
+  if (locks[dbName].queue.length) {
+    locks[dbName].inProgress = true;
+    const tx = locks[dbName].queue.shift();
+    setImmediate(() => {
+      tx.start();
+    });
+  }
 };
 
 //   _________     _______  ______ ____  _____  __  __            _____ _____
@@ -413,7 +438,7 @@ export const typeORMDriver = {
             fail(e);
           }
         },
-        transaction: (fn: (tx: Transaction) => Promise<void>): void => {
+        transaction: (fn: (tx: Transaction) => Promise<void>): Promise<void> => {
           return QuickSQLite.transactionAsync(options.name, fn);
         },
         close: (ok: any, fail: any) => {