feat: Update ajv-threads package to version with worker pool (#3177)

This will make schema validation in js-ceramic truly parallel for the first time
ceramicnetwork · Mar 12, 2024 · 3a1e0db · 3a1e0db
1 parent b8c2a3a
commit 3a1e0db
Show file tree

Hide file tree

Showing 6 changed files with 13 additions and 6 deletions.
diff --git a/packages/core/package.json b/packages/core/package.json
@@ -71,7 +71,7 @@
     "@stablelib/random": "^1.0.1",
     "@stablelib/sha256": "^1.0.1",
     "@stablelib/uuid": "^1.0.1",
-    "ajv-threads": "^1.0.3",
+    "ajv-threads": "^1.0.4",
     "await-semaphore": "^0.1.3",
     "cartonne": "^3.0.1",
     "codeco": "^1.1.0",

diff --git a/packages/core/src/ceramic.ts b/packages/core/src/ceramic.ts
@@ -244,7 +244,10 @@ export class Ceramic implements StreamReaderWriter, StreamStateLoader {
 
     this._ipfs = modules.ipfs
 
-    this._schemaValidator = new SchemaValidation()
+    const numCores = os.cpus().length
+    // Use number of threads equal to half the available cores for schema validation. Leave the
+    // other half for signature validation.
+    this._schemaValidator = new SchemaValidation(Math.max(1, Math.ceil(numCores / 2)))
     this._streamHandlers = HandlersMap.makeWithDefaultHandlers(this._logger, this._schemaValidator)
 
     // This initialization block below has to be redone.
@@ -484,6 +487,7 @@ export class Ceramic implements StreamReaderWriter, StreamStateLoader {
         this._logger.warn(`Starting in read-only mode. All write operations will fail`)
       }
 
+      await this._schemaValidator.init()
       await this.repository.init()
       await this.dispatcher.init()
 

diff --git a/packages/core/src/stream-loading/__tests__/state-manipulator.test.ts b/packages/core/src/stream-loading/__tests__/state-manipulator.test.ts
@@ -62,7 +62,8 @@ describeIfV3('StateManipulator test', () => {
     const logger = new LoggerProvider().getDiagnosticsLogger()
     logSyncer = new LogSyncer(dispatcher)
 
-    schemaValidator = new SchemaValidation()
+    schemaValidator = new SchemaValidation(1)
+    await schemaValidator.init()
     const handlers = HandlersMap.makeWithDefaultHandlers(logger, schemaValidator)
     stateManipulator = new StateManipulator(logger, handlers, logSyncer, ceramic)
 

diff --git a/packages/core/src/stream-loading/__tests__/stream-loader.test.ts b/packages/core/src/stream-loading/__tests__/stream-loader.test.ts
@@ -85,7 +85,8 @@ describeIfV3('Streamloader', () => {
         ceramic.anchorService.validator
       )
 
-      schemaValidator = new SchemaValidation()
+      schemaValidator = new SchemaValidation(1)
+      await schemaValidator.init()
       const handlers = HandlersMap.makeWithDefaultHandlers(logger, schemaValidator)
       const stateManipulator = new StateManipulator(logger, handlers, logSyncer, ceramic)
       streamLoader = new StreamLoader(

diff --git a/packages/stream-model-instance-handler/package.json b/packages/stream-model-instance-handler/package.json
@@ -43,7 +43,7 @@
     "@ceramicnetwork/stream-model": "^4.2.0-rc.0",
     "@ceramicnetwork/stream-model-instance": "^4.3.0-rc.0",
     "@ceramicnetwork/streamid": "^5.0.0",
-    "ajv-threads": "^1.0.3",
+    "ajv-threads": "^1.0.4",
     "fast-json-patch": "^3.1.0",
     "least-recent": "^1.0.3",
     "lodash.clonedeep": "^4.5.0",

diff --git a/packages/stream-model-instance-handler/src/__tests__/model-instance-document-handler.test.ts b/packages/stream-model-instance-handler/src/__tests__/model-instance-document-handler.test.ts
@@ -423,7 +423,8 @@ describe('ModelInstanceDocumentHandler', () => {
   let ipfs: IpfsApi
 
   beforeAll(async () => {
-    schemaValidator = new SchemaValidation()
+    schemaValidator = new SchemaValidation(1)
+    await schemaValidator.init()
     const recs: Record<string, any> = {}
     ipfs = {
       dag: {