Handle multiple transactions in parallel

Instead of assuming one global transaction ID, track knex transaction
IDs with Aurora Data API transaction IDs so multiple transactions can
occur in parallel.

Author: txase

index.js

@@ -103,6 +103,7 @@ class Client_AuroraDataMySQL extends Client_MySQL { // eslint-disable-line camel
         promise: Promise.resolve({
           client: this.driver,
           parameters,
+          transactions: {},
           __knexUid: this.knexUid++
         })
       }),
@@ -207,13 +208,19 @@ class Client_AuroraDataMySQL extends Client_MySQL { // eslint-disable-line camel
   }
 
   async _query (connection, obj) {
+    const params = {
+      ...connection.parameters,
+      includeResultMetadata: true,
+      sql: obj.sql,
+      parameters: obj.bindings
+    };
+
+    if ('__knexTxId' in connection) {
+      params.transactionId = connection.transactions[connection.__knexTxId];
+    }
+
     obj.data = await connection.client
-      .executeStatement({
-        ...connection.parameters,
-        includeResultMetadata: true,
-        sql: obj.sql,
-        parameters: obj.bindings
-      })
+      .executeStatement(params)
       .promise();
 
     return obj;

tests/constants.js

@@ -2178,5 +2178,9 @@ module.exports = {
   },
   ROLLBACK_TRANSACTION_DATA: {
     transactionStatus: 'Rollback Complete'
+  },
+
+  BEGIN_TRANSACTION_DATA_2: {
+    transactionId: 'lksdjfojf_2'
   }
 };

tests/test.js

@@ -627,4 +627,80 @@ describe('Query statement tests', () => {
     expect(mockExecuteStatement).toHaveBeenCalledTimes(0);
     expect(mockCommitTransaction).toHaveBeenCalledTimes(0);
   });
+
+  test('Two transactions in parallel', async () => {
+    mockBeginTransactionPromise
+      .mockResolvedValueOnce(constants.BEGIN_TRANSACTION_DATA)
+      .mockResolvedValueOnce(constants.BEGIN_TRANSACTION_DATA_2);
+    mockExecuteStatementPromise.mockResolvedValue(constants.ALL_QUERY_RESPONSE_DATA);
+    mockCommitTransactionPromise.mockResolvedValue(constants.COMMIT_TRANSACTION_DATA);
+
+    let firstTrxInProgress = false;
+
+    const [rows, rows2] = await Promise.all([
+      knex.transaction(async trx => {
+        firstTrxInProgress = true;
+        await new Promise(resolve => setTimeout(resolve, 100));
+        const rows = trx.select('*').from('foo');
+        firstTrxInProgress = false;
+        return rows;
+      }),
+      knex.transaction(async trx => {
+        if (!firstTrxInProgress) {
+          throw new Error('First transaction not concurrently in progress');
+        }
+        await new Promise(resolve => setTimeout(resolve, 100));
+        return trx.select('*').from('foo');
+      })
+    ]);
+
+    expect(mockBeginTransaction).toHaveBeenCalledTimes(2);
+    expect(mockBeginTransaction).toHaveBeenNthCalledWith(1, {
+      resourceArn: constants.AURORA_CLUSTER_ARN,
+      secretArn: constants.SECRET_ARN,
+      database: constants.DATABASE
+    });
+    expect(mockBeginTransaction).toHaveBeenNthCalledWith(2, {
+      resourceArn: constants.AURORA_CLUSTER_ARN,
+      secretArn: constants.SECRET_ARN,
+      database: constants.DATABASE
+    });
+
+    expect(mockExecuteStatement).toHaveBeenCalledTimes(2);
+    expect(mockExecuteStatement).toHaveBeenNthCalledWith(1, {
+      resourceArn: constants.AURORA_CLUSTER_ARN,
+      secretArn: constants.SECRET_ARN,
+      database: constants.DATABASE,
+      transactionId: constants.BEGIN_TRANSACTION_DATA.transactionId,
+      sql: 'select * from `foo`',
+      parameters: [],
+      includeResultMetadata: true
+    });
+    expect(mockExecuteStatement).toHaveBeenNthCalledWith(2, {
+      resourceArn: constants.AURORA_CLUSTER_ARN,
+      secretArn: constants.SECRET_ARN,
+      database: constants.DATABASE,
+      transactionId: constants.BEGIN_TRANSACTION_DATA_2.transactionId,
+      sql: 'select * from `foo`',
+      parameters: [],
+      includeResultMetadata: true
+    });
+
+    expect(mockCommitTransaction).toHaveBeenCalledTimes(2);
+    expect(mockCommitTransaction).toHaveBeenNthCalledWith(1, {
+      resourceArn: constants.AURORA_CLUSTER_ARN,
+      secretArn: constants.SECRET_ARN,
+      transactionId: constants.BEGIN_TRANSACTION_DATA.transactionId
+    });
+    expect(mockCommitTransaction).toHaveBeenNthCalledWith(2, {
+      resourceArn: constants.AURORA_CLUSTER_ARN,
+      secretArn: constants.SECRET_ARN,
+      transactionId: constants.BEGIN_TRANSACTION_DATA_2.transactionId
+    });
+
+    expect(mockRollbackTransaction).toHaveBeenCalledTimes(0);
+
+    expect(rows).toEqual(constants.ALL_QUERY_RESPONSE_ROWS);
+    expect(rows2).toEqual(constants.ALL_QUERY_RESPONSE_ROWS);
+  });
 });

transaction.js

@@ -14,9 +14,9 @@ class Transaction_AuroraDataMySQL extends Transaction { // eslint-disable-line c
 
   async begin (conn) {
     /* istanbul ignore next */
-    if (conn.parameters.transactionId) {
+    if (conn.__knexTxId in conn.transactions) {
       throw new Error(
-        `Attempted to begin a new transaction for connection with existing transaction ${conn.parameters.transactionId}`
+        `Attempted to begin a new transaction for connection ${conn.__knexUid} transaction ${conn.__knexTxId} with existing Aurora Data API transaction ID ${conn.transactions[conn.__knexTxId]}`
       );
     }
 
@@ -25,49 +25,55 @@ class Transaction_AuroraDataMySQL extends Transaction { // eslint-disable-line c
       .promise();
     debug(`Transaction begun with id ${transactionId}`);
 
-    conn.parameters.transactionId = transactionId;
+    conn.transactions[conn.__knexTxId] = transactionId;
   }
 
   async commit (conn, value) {
     // When a transaction is explicitly rolled back this method is still called
     // at the end of the transaction block after the transaction no longer
     // exists.
-    if ('transactionId' in conn.parameters) {
-      const params = { ...conn.parameters };
+    if (conn.__knexTxId in conn.transactions) {
+      const params = {
+        ...conn.parameters,
+        transactionId: conn.transactions[conn.__knexTxId]
+      };
       delete params.database;
 
       const { transactionStatus } = await conn.client
         .commitTransaction(params)
         .promise();
       debug(
-        `Transaction ${conn.parameters.transactionId} commit status: ${transactionStatus}`
+        `Transaction ${conn.transactions[conn.__knexTxId]} commit status: ${transactionStatus}`
       );
 
-      delete conn.parameters.transactionId;
+      delete conn.transactions[conn.__knexTxId];
     }
 
     this._resolver(value);
   }
 
   async rollback (conn, error) {
     /* istanbul ignore next */
-    if (!('transactionId' in conn.parameters)) {
+    if (!(conn.__knexTxId in conn.transactions)) {
       throw new Error(
         'Attempted to rollback a transaction when one is not in progress'
       );
     }
 
-    const params = { ...conn.parameters };
+    const params = {
+      ...conn.parameters,
+      transactionId: conn.transactions[conn.__knexTxId]
+    };
     delete params.database;
 
     const { transactionStatus } = await conn.client
       .rollbackTransaction(params)
       .promise();
     debug(
-      `Transaction ${conn.parameters.transactionId} rollback status: ${transactionStatus}`
+      `Transaction ${conn.transactions[conn.__knexTxId]} rollback status: ${transactionStatus}`
     );
 
-    delete conn.parameters.transactionId;
+    delete conn.transactions[conn.__knexTxId];
 
     if (error === undefined) {
       if (this.doNotRejectOnRollback) {