fix: fixing up @matrixai/workers usage

Author: tegefaulkes

scripts/prebuild.mjs

@@ -7,7 +7,7 @@ import url from 'node:url';
 import process from 'node:process';
 import childProcess from 'node:child_process';
 import semver from 'semver';
-import packageJSON from '../package.json' assert { type: "json" };
+import packageJSON from '../package.json' assert { type: 'json' };
 
 const projectPath = path.dirname(
   path.dirname(url.fileURLToPath(import.meta.url)),

src/DB.ts

@@ -13,9 +13,11 @@ import type {
   DBClearOptions,
   DBCountOptions,
 } from './types.js';
-import type { RocksDBDatabase, RocksDBDatabaseOptions } from './native/index.js';
+import type {
+  RocksDBDatabase,
+  RocksDBDatabaseOptions,
+} from './native/index.js';
 import nodeFs from 'node:fs';
-import { Transfer } from 'threads';
 import Logger from '@matrixai/logger';
 import { withF, withG } from '@matrixai/resources';
 import {
@@ -652,18 +654,20 @@ class DB {
         // Slice-copy for transferring to worker threads
         const key = utils.toArrayBuffer(this.crypto.key);
         const plainText = utils.toArrayBuffer(plainTextBuf);
-        cipherText = await this.workerManager.call(async (w) => {
-          return await w.encrypt(
-            Transfer(key),
-            // @ts-ignore: threads.js types are wrong
-            Transfer(plainText),
-          );
-        });
+        const result = await this.workerManager.methods.encrypt(
+          { key, plainText },
+          [key, plainText],
+        );
+        cipherText = result.data;
       } else {
-        cipherText = await this.crypto.ops.encrypt(
-          this.crypto.key,
-          plainTextBuf,
+        const result = await this.crypto.ops.encrypt(
+          {
+            key: this.crypto.key,
+            plainText: plainTextBuf,
+          },
+          [this.crypto.key, plainTextBuf],
         );
+        cipherText = result.data;
       }
       return utils.fromArrayBuffer(cipherText);
     }
@@ -689,18 +693,23 @@ class DB {
         // Slice-copy for transferring to worker threads
         const key = utils.toArrayBuffer(this.crypto.key);
         const cipherText = utils.toArrayBuffer(cipherTextBuf);
-        decrypted = await this.workerManager.call(async (w) => {
-          return await w.decrypt(
-            Transfer(key),
-            // @ts-ignore: threads.js types are wrong
-            Transfer(cipherText),
-          );
-        });
+        const result = await this.workerManager.methods.decrypt(
+          {
+            key: this.crypto.key,
+            cipherText: cipherTextBuf,
+          },
+          [this.crypto.key, cipherTextBuf],
+        );
+        decrypted = result.data;
       } else {
-        decrypted = await this.crypto.ops.decrypt(
-          this.crypto.key,
-          cipherTextBuf,
+        const result = await this.crypto.ops.decrypt(
+          {
+            key: this.crypto.key,
+            cipherText: cipherTextBuf,
+          },
+          [this.crypto.key, cipherTextBuf],
         );
+        decrypted = result.data;
       }
       if (decrypted == null) {
         throw new errors.ErrorDBDecrypt();

src/types.ts

@@ -1,5 +1,5 @@
 import type fs from 'node:fs';
-import type { WorkerManagerInterface } from '@matrixai/workers';
+import type { WorkerManager } from '@matrixai/workers';
 import type {
   RocksDBDatabaseOptions,
   RocksDBIteratorOptions,
@@ -50,19 +50,24 @@ interface FileSystem {
   };
 }
 
-/**
- * Crypto utility object
- * Remember ever Node Buffer is an ArrayBuffer
- */
 type Crypto = {
-  encrypt(key: ArrayBuffer, plainText: ArrayBuffer): Promise<ArrayBuffer>;
+  encrypt(
+    data: { key: ArrayBuffer; plainText: ArrayBuffer },
+    transferList: [ArrayBuffer, ArrayBuffer],
+  ): Promise<{ data: ArrayBuffer; transferList: [ArrayBuffer] }>;
   decrypt(
-    key: ArrayBuffer,
-    cipherText: ArrayBuffer,
-  ): Promise<ArrayBuffer | undefined>;
+    data: {
+      key: ArrayBuffer;
+      cipherText: ArrayBuffer;
+    },
+    transferList: [ArrayBuffer, ArrayBuffer],
+  ): Promise<
+    | { data: ArrayBuffer; transferList: [ArrayBuffer] }
+    | { data: undefined; transferList: [] }
+  >;
 };
 
-type DBWorkerManagerInterface = WorkerManagerInterface<Crypto>;
+type DBWorkerManagerInterface = WorkerManager<Crypto>;
 
 /**
  * Path to a key

tests/DB.test.ts

@@ -1,33 +1,28 @@
 import type DBTransaction from '#DBTransaction.js';
 import type { KeyPath } from '#types.js';
 import type { ResourceRelease } from '@matrixai/resources';
-import type { DBWorkerModule } from './workers/dbWorkerModule.js';
+import type { DBWorker } from './workers/dbWorker.js';
 import os from 'node:os';
 import path from 'node:path';
 import fs from 'node:fs';
 import nodeCrypto from 'node:crypto';
+import { Worker } from 'node:worker_threads';
 import Logger, { LogLevel, StreamHandler } from '@matrixai/logger';
-// import { WorkerManager } from '@matrixai/workers';
-import matrixaiWorkers from '@matrixai/workers';
+import { WorkerManager } from '@matrixai/workers';
 import { withF } from '@matrixai/resources';
-import { spawn, Worker } from 'threads';
+import * as testsUtils from './utils.js';
+import dbWorker from './workers/dbWorker.js';
 import DB from '#DB.js';
 import * as errors from '#errors.js';
 import * as utils from '#utils.js';
-import * as testsUtils from './utils.js';
-
-const { WorkerManager } = matrixaiWorkers;
 
 describe(DB.name, () => {
   const logger = new Logger(`${DB.name} Test`, LogLevel.WARN, [
     new StreamHandler(),
   ]);
   const crypto = {
     key: testsUtils.generateKeySync(256),
-    ops: {
-      encrypt: testsUtils.encrypt,
-      decrypt: testsUtils.decrypt,
-    },
+    ops: dbWorker,
   };
   let dataDir: string;
   beforeEach(async () => {
@@ -346,12 +341,12 @@ describe(DB.name, () => {
   test('parallelized get and put and del', async () => {
     const dbPath = `${dataDir}/db`;
     const db = await DB.createDB({ dbPath, crypto, logger });
-    const workerManager =
-      await WorkerManager.createWorkerManager<DBWorkerModule>({
-        workerFactory: () => spawn(new Worker('./workers/dbWorker')),
-        cores: 1,
-        logger,
-      });
+    const workerManager = await WorkerManager.createWorkerManager<DBWorker>({
+      workerFactory: () => new Worker('./workers/dbWorker'),
+      cores: 1,
+      manifest: dbWorker,
+      logger,
+    });
     db.setWorkerManager(workerManager);
     await db.start();
     await db.put('a', 'value0');
@@ -408,12 +403,12 @@ describe(DB.name, () => {
   test('parallelized batch put and del', async () => {
     const dbPath = `${dataDir}/db`;
     const db = await DB.createDB({ dbPath, crypto, logger });
-    const workerManager =
-      await WorkerManager.createWorkerManager<DBWorkerModule>({
-        workerFactory: () => spawn(new Worker('./workers/dbWorker')),
-        cores: 4,
-        logger,
-      });
+    const workerManager = await WorkerManager.createWorkerManager<DBWorker>({
+      workerFactory: () => new Worker('./workers/dbWorker'),
+      cores: 4,
+      manifest: dbWorker,
+      logger,
+    });
     db.setWorkerManager(workerManager);
     await db.start();
     await db.batch([

tests/DBIterator.test.ts

@@ -1,26 +1,24 @@
-import type { KeyPath } from '@/types';
+import type { KeyPath } from '#types.js';
 import os from 'os';
 import path from 'path';
 import fs from 'fs';
 import nodeCrypto from 'crypto';
 import nodeUtil from 'util';
 import lexi from 'lexicographic-integer';
 import Logger, { LogLevel, StreamHandler } from '@matrixai/logger';
-import DB from '@/DB';
-import DBIterator from '@/DBIterator';
-import rocksdbP from '@/native/rocksdbP';
-import * as testsUtils from './utils';
+import * as testsUtils from './utils.js';
+import dbWorker from './workers/dbWorker.js';
+import DB from '#DB.js';
+import DBIterator from '#DBIterator.js';
+import rocksdbP from '#native/rocksdbP.js';
 
 describe(DBIterator.name, () => {
   const logger = new Logger(`${DBIterator.name} test`, LogLevel.WARN, [
     new StreamHandler(),
   ]);
   const crypto = {
     key: testsUtils.generateKeySync(256),
-    ops: {
-      encrypt: testsUtils.encrypt,
-      decrypt: testsUtils.decrypt,
-    },
+    ops: dbWorker,
   };
   let dataDir: string;
   let db: DB;

tests/DBTransaction.test.ts

@@ -1,25 +1,23 @@
-import type { KeyPath } from '@/types';
+import type { KeyPath } from '#types.js';
 import os from 'os';
 import path from 'path';
 import fs from 'fs';
 import Logger, { LogLevel, StreamHandler } from '@matrixai/logger';
 import { withF } from '@matrixai/resources';
 import { Barrier, errors as locksErrors } from '@matrixai/async-locks';
-import DB from '@/DB';
-import DBTransaction from '@/DBTransaction';
-import * as errors from '@/errors';
-import * as testsUtils from './utils';
+import * as testsUtils from './utils.js';
+import dbWorker from './workers/dbWorker.js';
+import DB from '#DB.js';
+import DBTransaction from '#DBTransaction.js';
+import * as errors from '#errors.js';
 
 describe(DBTransaction.name, () => {
   const logger = new Logger(`${DBTransaction.name} test`, LogLevel.WARN, [
     new StreamHandler(),
   ]);
   const crypto = {
     key: testsUtils.generateKeySync(256),
-    ops: {
-      encrypt: testsUtils.encrypt,
-      decrypt: testsUtils.decrypt,
-    },
+    ops: dbWorker,
   };
   let dataDir: string;
   let db: DB;

tests/rocksdb/rocksdbP.test.ts

@@ -1,9 +1,9 @@
-import type { RocksDBDatabase } from '@/native/types';
+import type { RocksDBDatabase } from '#native/types.js';
 import os from 'os';
 import path from 'path';
 import fs from 'fs';
 import { Barrier } from '@matrixai/async-locks';
-import rocksdbP from '@/native/rocksdbP';
+import rocksdbP from '#native/rocksdbP.js';
 
 describe('rocksdbP', () => {
   let dataDir: string;

tests/utils.test.ts

@@ -1,7 +1,7 @@
 import type { KeyPath } from '#types.js';
 import nodeCrypto from 'node:crypto';
-import * as utils from '#utils.js';
 import * as testUtils from './utils.js';
+import * as utils from '#utils.js';
 
 describe('utils', () => {
   const keyPaths: Array<KeyPath> = [

tests/workers/dbWorker.ts

@@ -1,7 +1,50 @@
-import type { DBWorkerModule } from './dbWorkerModule.js';
-import { expose } from 'threads/worker';
-import dbWorker from './dbWorkerModule.js';
+import type { Crypto } from '#types.js';
+import type { WorkerManifest } from '@matrixai/workers';
+import { expose } from '@matrixai/workers';
+import * as utils from '../utils.js';
+
+const dbWorker: Crypto = {
+  async encrypt(
+    {
+      key,
+      plainText,
+    }: {
+      key: ArrayBuffer;
+      plainText: ArrayBuffer;
+    },
+    // eslint-disable-line @typescript-eslint/no-unused-vars
+    transferList: [ArrayBuffer, ArrayBuffer],
+  ): Promise<{ data: ArrayBuffer; transferList: [ArrayBuffer] }> {
+    const cipherText = await utils.encrypt(key, plainText);
+    return { data: cipherText, transferList: [cipherText] };
+  },
+  async decrypt(
+    {
+      key,
+      cipherText,
+    }: {
+      key: ArrayBuffer;
+      cipherText: ArrayBuffer;
+    },
+    // eslint-disable-line @typescript-eslint/no-unused-vars
+    transferList: [ArrayBuffer, ArrayBuffer],
+  ): Promise<
+    | { data: ArrayBuffer; transferList: [ArrayBuffer] }
+    | { data: undefined; transferList: [] }
+  > {
+    const plainText = await utils.decrypt(key, cipherText);
+    if (plainText != null) {
+      return { data: plainText, transferList: [plainText] };
+    } else {
+      return { data: undefined, transferList: [] };
+    }
+  },
+} satisfies WorkerManifest;
 
 expose(dbWorker);
 
-export type { DBWorkerModule };
+type DBWorker = typeof dbWorker;
+
+export type { DBWorker };
+
+export default dbWorker;