fix: use Barrier to ensure deterministic concurrent conflict tests

Author: CMCDragonkai

package-lock.json

@@ -28,7 +28,7 @@
   },
   "dependencies": {
     "@matrixai/async-init": "^1.8.1",
-    "@matrixai/async-locks": "^3.0.0",
+    "@matrixai/async-locks": "^3.1.1",
     "@matrixai/errors": "^1.1.2",
     "@matrixai/logger": "^2.3.0",
     "@matrixai/resources": "^1.1.3",

package.json

@@ -4,7 +4,7 @@ import path from 'path';
 import fs from 'fs';
 import Logger, { LogLevel, StreamHandler } from '@matrixai/logger';
 import { withF } from '@matrixai/resources';
-import { errors as locksErrors } from '@matrixai/async-locks';
+import { Barrier, errors as locksErrors } from '@matrixai/async-locks';
 import DB from '@/DB';
 import DBTransaction from '@/DBTransaction';
 import * as errors from '@/errors';
@@ -260,6 +260,8 @@ describe(DBTransaction.name, () => {
     expect(await db.get('hello')).toBeUndefined();
   });
   test('getForUpdate addresses write-skew by promoting gets into same-value puts', async () => {
+    // Ensure deterministic concurrency
+    const barrier = await Barrier.createBarrier(2);
     // Snapshot isolation allows write skew anomalies to occur
     // A write skew means that 2 transactions concurrently read from overlapping keys
     // then make disjoint updates to the keys, that breaks a consistency constraint on those keys
@@ -272,13 +274,15 @@ describe(DBTransaction.name, () => {
     const t1 = withF([db.transaction()], async ([tran]) => {
       let balance1 = parseInt((await tran.getForUpdate('balance1'))!);
       const balance2 = parseInt((await tran.getForUpdate('balance2'))!);
+      await barrier.wait();
       balance1 -= 100;
       expect(balance1 + balance2).toBeGreaterThanOrEqual(0);
       await tran.put('balance1', balance1.toString());
     });
     const t2 = withF([db.transaction()], async ([tran]) => {
       const balance1 = parseInt((await tran.getForUpdate('balance1'))!);
       let balance2 = parseInt((await tran.getForUpdate('balance2'))!);
+      await barrier.wait();
       balance2 -= 100;
       expect(balance1 + balance2).toBeGreaterThanOrEqual(0);
       await tran.put('balance2', balance2.toString());
@@ -298,16 +302,20 @@ describe(DBTransaction.name, () => {
     ).toBe(true);
   });
   test('locking to prevent thrashing for racing counters', async () => {
+    // Ensure deterministic concurrency
+    const barrier = await Barrier.createBarrier(2);
     await db.put('counter', '0');
     let t1 = withF([db.transaction()], async ([tran]) => {
       // Can also use `getForUpdate`, but a conflict exists even for `get`
       let counter = parseInt((await tran.get('counter'))!);
+      await barrier.wait();
       counter++;
       await tran.put('counter', counter.toString());
     });
     let t2 = withF([db.transaction()], async ([tran]) => {
       // Can also use `getForUpdate`, but a conflict exists even for `get`
       let counter = parseInt((await tran.get('counter'))!);
+      await barrier.wait();
       counter++;
       await tran.put('counter', counter.toString());
     });

tests/DBTransaction.test.ts

@@ -2,6 +2,7 @@ import type { RocksDBDatabase } from '@/native/types';
 import os from 'os';
 import path from 'path';
 import fs from 'fs';
+import { Barrier } from '@matrixai/async-locks';
 import rocksdbP from '@/native/rocksdbP';
 
 describe('rocksdbP', () => {
@@ -240,6 +241,8 @@ describe('rocksdbP', () => {
         );
       });
       test('transactionGetForUpdate addresses write skew by promoting gets into same-value puts', async () => {
+        // Ensure deterministic concurrency
+        const barrier = await Barrier.createBarrier(2);
         // Snapshot isolation allows write skew anomalies to occur
         // A write skew means that 2 transactions concurrently read from overlapping keys
         // then make disjoint updates to the keys, that breaks a consistency constraint on those keys
@@ -257,6 +260,7 @@ describe('rocksdbP', () => {
           const balance2 = parseInt(
             await rocksdbP.transactionGetForUpdate(tran1, 'balance2', {}),
           );
+          await barrier.wait();
           balance1 -= 100;
           expect(balance1 + balance2).toBeGreaterThanOrEqual(0);
           await rocksdbP.transactionPut(tran1, 'balance1', balance1.toString());
@@ -270,6 +274,7 @@ describe('rocksdbP', () => {
           let balance2 = parseInt(
             await rocksdbP.transactionGetForUpdate(tran2, 'balance2', {}),
           );
+          await barrier.wait();
           balance2 -= 100;
           expect(balance1 + balance2).toBeGreaterThanOrEqual(0);
           await rocksdbP.transactionPut(tran2, 'balance2', balance2.toString());
@@ -292,6 +297,8 @@ describe('rocksdbP', () => {
         ).toBe(true);
       });
       test('transactionMultiGetForUpdate addresses write skew by promoting gets into same-value puts', async () => {
+        // Ensure deterministic concurrency
+        const barrier = await Barrier.createBarrier(2);
         // Snapshot isolation allows write skew anomalies to occur
         // A write skew means that 2 transactions concurrently read from overlapping keys
         // then make disjoint updates to the keys, that breaks a consistency constraint on those keys
@@ -321,6 +328,7 @@ describe('rocksdbP', () => {
               )
             )[0],
           );
+          await barrier.wait();
           balance1 -= 100;
           expect(balance1 + balance2).toBeGreaterThanOrEqual(0);
           await rocksdbP.transactionPut(tran1, 'balance1', balance1.toString());
@@ -346,6 +354,7 @@ describe('rocksdbP', () => {
               )
             )[0],
           );
+          await barrier.wait();
           balance2 -= 100;
           expect(balance1 + balance2).toBeGreaterThanOrEqual(0);
           await rocksdbP.transactionPut(tran2, 'balance2', balance2.toString());