feat: introduce WORKERS_POOL_ENABLED config to optionally disable worker thread pool

This change allows the relay to run CPU-intensive tasks on the main event loop, reducing the memory floor for resource-constrained environments (e.g. 128MB RAM).

Key changes:
- Added WORKERS_POOL_ENABLED configuration variable (default true).
- Updated WorkersPool.ts to supporting local execution via dynamic imports.
- Extracted WorkersErrorUtils.ts to handle error serialization/deserialization across thread boundaries.
- Updated blockWorker.ts and commonWorker.ts to use the new error utility.
- Added comprehensive unit tests for WorkersPool and WorkersErrorUtils.

Signed-off-by: Logan Nguyen <logan.nguyen@swirldslabs.com>

Author: quiet-node

packages/config-service/src/services/globalConfig.ts

@@ -776,6 +776,11 @@ const _CONFIG = {
     required: false,
     defaultValue: true,
   },
+  WORKERS_POOL_ENABLED: {
+    type: 'boolean',
+    required: false,
+    defaultValue: true,
+  },
   WORKERS_POOL_MAX_THREADS: {
     type: 'number',
     required: false,

packages/relay/src/lib/services/ethService/blockService/blockWorker.ts

@@ -23,7 +23,7 @@ import {
 import { Block, Log, Transaction } from '../../../model';
 import { IContractResultsParams, ITransactionReceipt, MirrorNodeBlock, RequestDetails } from '../../../types';
 import { IReceiptRlpInput } from '../../../types/IReceiptRlpInput';
-import { WorkersPool } from '../../workersService/WorkersPool';
+import { wrapError } from '../../workersService/WorkersErrorUtils';
 import { CommonService } from '../ethCommonService/CommonService';
 
 /**
@@ -375,7 +375,7 @@ export async function getBlock(
       receiptsRoot,
     });
   } catch (e: unknown) {
-    throw WorkersPool.wrapError(e);
+    throw wrapError(e);
   }
 }
 
@@ -467,7 +467,7 @@ export async function getBlockReceipts(
 
     return sortedReceipts as ITransactionReceipt[];
   } catch (e: unknown) {
-    throw WorkersPool.wrapError(e);
+    throw wrapError(e);
   }
 }
 
@@ -482,8 +482,8 @@ export async function getBlockReceipts(
  * @param blockHashOrBlockNumber - Block hash (0x-prefixed) or block number string
  * @param requestDetails - The request details for logging and tracking
  * @returns Promise of an array of hex-encoded receipt strings (RLP), or empty array if
- *   the block has no contract results and no logs. Re-throws errors wrapped with
- *   {@link WorkersPool.wrapError}.
+ *   the block has no contract results and no logs. Re-throws errors via {@link wrapError}
+ *   when running inside a worker thread, or propagates natively on the main thread.
  */
 export async function getRawReceipts(
   blockHashOrBlockNumber: string,
@@ -527,7 +527,7 @@ export async function getRawReceipts(
 
     return encodedReceipts;
   } catch (e: unknown) {
-    throw WorkersPool.wrapError(e);
+    throw wrapError(e);
   }
 }
 

packages/relay/src/lib/services/ethService/ethCommonService/commonWorker.ts

@@ -7,7 +7,7 @@ import { MirrorNodeClient } from '../../../clients/mirrorNodeClient';
 import { CacheClientFactory } from '../../../factories/cacheClientFactory';
 import { RegistryFactory } from '../../../factories/registryFactory';
 import { RequestDetails } from '../../../types';
-import { WorkersPool } from '../../workersService/WorkersPool';
+import { wrapError } from '../../workersService/WorkersErrorUtils';
 import { CommonService } from './CommonService';
 
 const logger = pino({ level: ConfigService.get('LOG_LEVEL') || 'trace' });
@@ -57,6 +57,6 @@ export async function getLogs(
 
     return await commonService.getLogsWithParams(address, params, requestDetails, sliceCountWrapper.value);
   } catch (e: unknown) {
-    throw WorkersPool.wrapError(e);
+    throw wrapError(e);
   }
 }

packages/relay/src/lib/services/workersService/WorkersErrorUtils.ts

@@ -0,0 +1,103 @@
+// SPDX-License-Identifier: Apache-2.0
+
+import { parentPort } from 'worker_threads';
+
+import { JsonRpcError, predefined } from '../../errors/JsonRpcError';
+import { MirrorNodeClientError } from '../../errors/MirrorNodeClientError';
+
+/**
+ * Plain JSON representation of a serializable error that can be safely transported
+ * across worker thread boundaries using the Structured Clone algorithm.
+ *
+ * Piscina communicates task results and errors between threads via `postMessage`, which
+ * uses Structured Clone. Only plain data survives this boundary — class instances, prototype
+ * chains, and private fields are stripped. This envelope captures the fields needed to
+ * reconstruct supported error types on the receiving thread.
+ */
+interface ErrorEnvelope {
+  name: string;
+  message: string;
+  code?: number;
+  statusCode?: number;
+  data?: string;
+  detail?: string;
+}
+
+/**
+ * Conditionally serializes an error for cross-thread transport.
+ *
+ * When invoked inside a Piscina worker thread (`parentPort` is non-null), serializes the
+ * error into a standard {@link Error} whose `message` contains the JSON-encoded payload.
+ * This ensures that rich error types (e.g. {@link JsonRpcError}, {@link MirrorNodeClientError})
+ * survive the Structured Clone boundary used by `postMessage`, which otherwise strips
+ * prototype chains and class-specific fields.
+ *
+ * When invoked on the main thread (`parentPort` is null), returns the original error
+ * unchanged. This avoids the CPU/memory overhead of unnecessary serialization-deserialization
+ * cycles during local execution on the main thread (e.g. when WORKERS_POOL_ENABLED is false)
+ * and preserves the original object's identity.
+ *
+ * @param err - The error-like value to wrap if in a worker context.
+ * @returns The original error when called outside a worker, or a JSON-encoded `Error` inside one.
+ */
+export function wrapError(err: unknown): unknown {
+  if (!parentPort) {
+    return err;
+  }
+  return new Error(JSON.stringify(err));
+}
+
+/**
+ * Reconstructs the original typed error from an error previously produced by {@link wrapError}.
+ *
+ * Parses the JSON payload embedded in `err.message` and attempts to reconstruct one of the
+ * supported rich error types. Returns a predefined internal error if the input is not an
+ * `Error`, the message is not valid JSON, or the error name is unrecognised.
+ *
+ * Supported error types:
+ * - {@link JsonRpcError}
+ * - {@link MirrorNodeClientError}
+ *
+ * @param err - An error whose `message` is expected to contain a JSON-encoded {@link ErrorEnvelope}.
+ * @returns The reconstructed typed error, or a {@link predefined.INTERNAL_ERROR} if the
+ *   envelope cannot be parsed or the type is unsupported.
+ */
+export function unwrapError(err: unknown): Error {
+  if (!(err instanceof Error)) {
+    return predefined.INTERNAL_ERROR('Failed unwrapping piscina error: value is not an Error instance.');
+  }
+
+  let parsedErr: ErrorEnvelope;
+  try {
+    parsedErr = JSON.parse(err.message) as ErrorEnvelope;
+  } catch {
+    return predefined.INTERNAL_ERROR('Failed parsing wrapped piscina error while unwrapping.');
+  }
+
+  switch (parsedErr?.name) {
+    case JsonRpcError.name: {
+      if (typeof parsedErr.code !== 'number') {
+        return predefined.INTERNAL_ERROR(
+          'Failed unwrapping piscina error: missing numeric code in JsonRpcError envelope.',
+        );
+      }
+      return new JsonRpcError({
+        code: parsedErr.code,
+        data: parsedErr.data,
+        message: parsedErr.message,
+      });
+    }
+
+    case MirrorNodeClientError.name: {
+      if (typeof parsedErr.statusCode !== 'number') {
+        return predefined.INTERNAL_ERROR(
+          'Failed unwrapping piscina error: missing numeric statusCode in MirrorNodeClientError envelope.',
+        );
+      }
+      return MirrorNodeClientError.fromJSON(parsedErr.statusCode, parsedErr.message, parsedErr.data, parsedErr.detail);
+    }
+
+    default:
+      return predefined.INTERNAL_ERROR('Failed unwrapping piscina error.');
+  }
+}

packages/relay/src/lib/services/workersService/WorkersPool.ts

@@ -7,21 +7,9 @@ import { parentPort } from 'worker_threads';
 
 import { MeasurableCache, MirrorNodeClient } from '../../clients';
 import { ICacheClient } from '../../clients/cache/ICacheClient';
-import { JsonRpcError, predefined } from '../../errors/JsonRpcError';
-import { MirrorNodeClientError } from '../../errors/MirrorNodeClientError';
 import { RegistryFactory } from '../../factories/registryFactory';
-
-/**
- * Plain JSON representation of a serialized error that can be safely transferred across worker or process boundaries.
- */
-interface ErrorEnvelope {
-  name: string;
-  message: string;
-  code?: number;
-  statusCode?: number;
-  data?: string;
-  detail?: string;
-}
+import type { WorkerTask } from './workers';
+import { unwrapError } from './WorkersErrorUtils';
 
 /**
  * A wrapper around a shared Piscina worker thread pool.
@@ -80,6 +68,13 @@ export class WorkersPool {
    */
   private static workerPoolQueueSizeGauge: Gauge;
 
+  /**
+   * Cached reference to the local task handler function used when {@link WORKERS_POOL_ENABLED} is `false`.
+   *
+   * Populated on the first invocation of {@link run} to avoid repeating the dynamic import on every call.
+   */
+  private static handleTaskFn: ((task: WorkerTask) => Promise<any>) | null = null;
+
   /**
    * Updates a metric either by delegating the update to a worker thread
    * or by executing the update locally when no worker context is present.
@@ -219,14 +214,36 @@ export class WorkersPool {
   }
 
   /**
-   * Executes a worker task using the shared Piscina pool.
+   * Executes a worker task using either the shared Piscina thread pool or the main thread,
+   * depending on the {@link WORKERS_POOL_ENABLED} configuration flag.
+   *
+   * When the pool is enabled, the task is dispatched to a Piscina worker thread.
    *
-   * @param options - The data passed to the worker.
-   * @param mirrorNodeClient - The mirror node client instance.
-   * @param cacheService - The cache service instance.
-   * @returns A promise resolving to the worker's result.
+   * When the pool is disabled, the task is executed locally on the main thread and entirely
+   * bypasses the worker pool. In this mode `mirrorNodeClient` and `cacheService` are not
+   * forwarded to the task handler — the worker modules maintain their own module-level
+   * instances initialised on first use.
+   *
+   * @param options - The task descriptor forwarded to the worker handler.
+   * @param mirrorNodeClient - Mirror node client instance used to forward inter-thread metrics.
+   *   Unused when {@link WORKERS_POOL_ENABLED} is `false`.
+   * @param cacheService - Cache service instance used to forward inter-thread metrics.
+   *   Unused when {@link WORKERS_POOL_ENABLED} is `false`.
+   * @returns A promise that resolves to the task handler's return value.
+   * @throws The original error from the task handler; reconstructed from its serialized form
+   *   in worker mode, or native in local mode.
    */
-  static async run(options: unknown, mirrorNodeClient: MirrorNodeClient, cacheService: ICacheClient): Promise<any> {
+  static async run(options: WorkerTask, mirrorNodeClient: MirrorNodeClient, cacheService: ICacheClient): Promise<any> {
+    if (!ConfigService.get('WORKERS_POOL_ENABLED')) {
+      if (!this.handleTaskFn) {
+        // Dynamic import to defer loading worker modules and their dependencies until actually needed,
+        // ensuring any module-level instances are created once and reused across all local task executions.
+        const mod = await import('./workers');
+        this.handleTaskFn = mod.default;
+      }
+      return this.handleTaskFn(options);
+    }
+
     this.mirrorNodeClient = mirrorNodeClient;
     this.cacheService = cacheService as MeasurableCache;
 
@@ -240,7 +257,7 @@ export class WorkersPool {
     const result = await this.getInstance()
       .run(options)
       .catch((error: unknown) => {
-        const unwrappedErr = WorkersPool.unwrapError(error);
+        const unwrappedErr = unwrapError(error);
 
         this.workerTaskDurationSecondsHistogram
           ?.labels(taskType)
@@ -259,62 +276,4 @@ export class WorkersPool {
 
     return result;
   }
-
-  /**
-   * Wraps an error into a standard `Error` instance by serializing its plain JSON representation into
-   * the error message. Intended for transporting rich error information (including custom error types) across
-   * boundaries such as Piscina worker threads.
-   *
-   * @param err - An error-like object that implements `toJSON()`.
-   * @returns A new `Error` whose `message` contains a JSON-encoded
-   */
-  static wrapError(err: any): Error {
-    return new Error(JSON.stringify(err));
-  }
-
-  /**
-   * Unwraps an error previously wrapped with {@link wrapError} and attempts to reconstruct the original error
-   * instance. If parsing fails or the error type is unsupported, a predefined internal error is returned instead.
-   *
-   * Supported error types:
-   * - {@link JsonRpcError}
-   * - {@link MirrorNodeClientError}
-   *
-   * @param err - An error whose `message` is expected to contain a JSON-encoded {@link ErrorEnvelope}.
-   * @returns The reconstructed error instance, or an internal error if unwrapping fails.
-   */
-  static unwrapError(err: unknown): Error {
-    if (!(err instanceof Error)) {
-      return predefined.INTERNAL_ERROR(`Failed unwrapping piscina error: value is not an Error instance.`);
-    }
-
-    let parsedErr: ErrorEnvelope;
-    try {
-      parsedErr = JSON.parse(err.message) as ErrorEnvelope;
-    } catch {
-      return predefined.INTERNAL_ERROR(`Failed parsing wrapped piscina error while unwrapping.`);
-    }
-
-    switch (parsedErr?.name) {
-      case JsonRpcError.name: {
-        return new JsonRpcError({
-          code: parsedErr.code!,
-          data: parsedErr.data!,
-          message: parsedErr.message,
-        });
-      }
-
-      case MirrorNodeClientError.name: {
-        return MirrorNodeClientError.fromJSON(
-          parsedErr.statusCode!,
-          parsedErr.message,
-          parsedErr.data,
-          parsedErr.detail,
-        );
-      }
-
-      default:
-        return predefined.INTERNAL_ERROR('Failed unwrapping piscina error.');
-    }
-  }
 }

packages/relay/src/lib/services/workersService/workers.ts

@@ -34,11 +34,17 @@ interface GetRawReceiptsTask {
   requestDetails: RequestDetails;
 }
 
-type WorkerTask = GetLogsTask | GetBlockTask | GetBlockReceiptsTask | GetRawReceiptsTask;
+export type WorkerTask = GetLogsTask | GetBlockTask | GetBlockReceiptsTask | GetRawReceiptsTask;
 
 /**
- * Main worker export - handles different task types.
- * This function is called by Piscina with the task data.
+ * Dispatches a worker task to the appropriate handler function.
+ *
+ * Invoked either by Piscina on a dedicated worker thread, or directly on the main
+ * thread when {@link WORKERS_POOL_ENABLED} is `false` (local execution mode).
+ *
+ * @param task - Discriminated-union descriptor for the task to execute.
+ * @returns A promise that resolves to the handler's result.
+ * @throws {Error} If `task.type` does not match any known task variant.
  */
 export default async function handleTask(task: WorkerTask): Promise<any> {
   switch (task.type) {