Add maxWorkflowThreadHeapMiB to prevent OOM crashes from workflow cache

packages/test/src/test-threaded-vm.ts

@@ -0,0 +1,73 @@
+/**
+ * Unit tests for {@link WorkerThreadClient}'s handling of Workflow Worker Thread `error` events,
+ * in particular its ability to detect out-of-memory (OOM) errors and log a more actionable
+ * message pointing users at `maxCachedWorkflows` / `maxWorkflowThreadHeapMiB`.
+ */
+import { EventEmitter } from 'node:events';
+import test from 'ava';
+import type { LogEntry } from '@temporalio/worker';
+import { DefaultLogger } from '@temporalio/worker';
+import { WorkerThreadClient } from '@temporalio/worker/lib/workflow/threaded-vm';
+
+/**
+ * A minimal stand-in for a `node:worker_threads` `Worker`. `WorkerThreadClient` only ever
+ * registers listeners on it (via `.on`) in its constructor, so a plain `EventEmitter` is
+ * sufficient to drive its `error` handling logic without spinning up a real thread.
+ */
+class FakeWorkerThread extends EventEmitter {
+  postMessage(..._args: unknown[]): void {
+    // no-op: tests only exercise event handling, not actual message passing
+  }
+}
+
+function createClient(): { logs: Array<Omit<LogEntry, 'timestampNanos'>>; workerThread: FakeWorkerThread } {
+  const logs: Array<Omit<LogEntry, 'timestampNanos'>> = [];
+  const logger = new DefaultLogger('TRACE', ({ level, message, meta }) => logs.push({ level, message, meta }));
+  const workerThread = new FakeWorkerThread();
+  void new WorkerThreadClient(workerThread as any, logger);
+  return { logs, workerThread };
+}
+
+for (const message of [
+  'JS heap out of memory',
+  'Error [ERR_WORKER_OUT_OF_MEMORY]: Worker terminated due to reaching memory limit',
+  'Allocation failed - JavaScript heap out of memory',
+]) {
+  test(`logs an actionable OOM message when the worker thread errors with "${message}"`, (t) => {
+    const { logs, workerThread } = createClient();
+
+    workerThread.emit('error', new Error(message));
+
+    const errorLogs = logs.filter((l) => l.level === 'ERROR');
+    t.is(errorLogs.length, 1);
+    t.is(
+      errorLogs[0].message,
+      'Workflow Worker Thread ran out of memory. ' +
+        'Consider reducing maxCachedWorkflows or setting maxWorkflowThreadHeapMiB.'
+    );
+  });
+}
+
+test('logs a generic failure message for non-OOM worker thread errors', (t) => {
+  const { logs, workerThread } = createClient();
+
+  workerThread.emit('error', new Error('some unrelated failure'));
+
+  const errorLogs = logs.filter((l) => l.level === 'ERROR');
+  t.is(errorLogs.length, 1);
+  t.is(errorLogs[0].message, 'Workflow Worker Thread failed: Error: some unrelated failure');
+});
+
+test('worker thread errors reject pending completions after exit', async (t) => {
+  const logger = new DefaultLogger('TRACE', () => void 0);
+  const workerThread = new FakeWorkerThread();
+  const client = new WorkerThreadClient(workerThread as any, logger);
+
+  const sendPromise = client.send({ type: 'destroy' } as any);
+
+  workerThread.emit('error', new Error('JS heap out of memory'));
+  workerThread.emit('exit', 1);
+
+  const err = await t.throwsAsync(sendPromise);
+  t.regex((err as Error).message, /Workflow Worker Thread exited prematurely/);
+});

packages/test/src/test-worker-options.ts

@@ -42,3 +42,39 @@ for (const value of [-1, 1.5, Number.NaN]) {
     );
   });
 }
+
+test('forwards a positive maxWorkflowThreadHeapMiB to the compiled options', (t) => {
+  const runtime = Runtime.instance();
+  const compiled = compileWorkerOptions(
+    { ...defaultOptions, maxWorkflowThreadHeapMiB: 512 },
+    runtime.logger,
+    runtime.metricMeter
+  );
+
+  t.is(compiled.maxWorkflowThreadHeapMiB, 512);
+});
+
+test('leaves maxWorkflowThreadHeapMiB undefined by default', (t) => {
+  const runtime = Runtime.instance();
+  const compiled = compileWorkerOptions({ ...defaultOptions }, runtime.logger, runtime.metricMeter);
+
+  t.is(compiled.maxWorkflowThreadHeapMiB, undefined);
+});
+
+for (const value of [-1, 0]) {
+  test(`rejects invalid maxWorkflowThreadHeapMiB ${value}`, (t) => {
+    const runtime = Runtime.instance();
+    t.throws(
+      () =>
+        compileWorkerOptions(
+          { ...defaultOptions, maxWorkflowThreadHeapMiB: value },
+          runtime.logger,
+          runtime.metricMeter
+        ),
+      {
+        instanceOf: TypeError,
+        message: 'maxWorkflowThreadHeapMiB must be a positive number',
+      }
+    );
+  });
+}

packages/worker/src/worker-options.ts

@@ -436,6 +436,24 @@ export interface WorkerOptions {
    */
   maxCachedWorkflows?: number;
 
+  /**
+   * Maximum old-generation heap size (in MiB) for each workflow worker thread.
+   *
+   * When set, each workflow worker thread is created with
+   * {@link https://nodejs.org/api/worker_threads.html#new-workerfilename-options | resourceLimits.maxOldGenerationSizeMb}
+   * so that a single thread's V8 heap cannot grow beyond this limit. Without this, an unbounded workflow cache can
+   * exhaust the process heap and crash the entire worker with an unrecoverable OOM error.
+   *
+   * Setting this value converts a fatal process-level OOM into a per-thread error that the worker handles gracefully
+   * (the affected thread exits and its workflows are reported as failed). The worker itself continues running.
+   *
+   * A good starting point is to divide the available heap (`--max-old-space-size` or the Node.js default) by the
+   * number of workflow threads, leaving headroom for the main thread.
+   *
+   * @default undefined (no per-thread heap limit; the process-level limit applies)
+   */
+  maxWorkflowThreadHeapMiB?: number;
+
   /**
    * Controls the number of threads to be created for executing Workflow Tasks.
    *
@@ -1121,6 +1139,9 @@ export function compileWorkerOptions(
     logger.warn('maxCachedWorkflows must be either 0 (ie. cache is disabled) or greater than 1. Defaulting to 2.');
     opts.maxCachedWorkflows = 2;
   }
+  if (opts.maxWorkflowThreadHeapMiB !== undefined && opts.maxWorkflowThreadHeapMiB <= 0) {
+    throw new TypeError('maxWorkflowThreadHeapMiB must be a positive number');
+  }
 
   if (opts.maxConcurrentWorkflowTaskExecutions !== undefined) {
     if (opts.maxCachedWorkflows > 0 && opts.maxConcurrentWorkflowTaskExecutions > opts.maxCachedWorkflows) {

packages/worker/src/worker.ts

@@ -608,6 +608,7 @@ export class Worker {
         registeredActivityNames,
         logger,
         patchActivationCallback: compiledOptions.patchActivationCallback,
+        maxOldGenerationSizeMb: compiledOptions.maxWorkflowThreadHeapMiB,
       });
     }
   }

packages/worker/src/workflow/threaded-vm.ts

@@ -98,9 +98,21 @@ export class WorkerThreadClient {
       completion.resolve(result.output);
     });
     workerThread.on('error', (err) => {
-      logger.error(`Workflow Worker Thread failed: ${err}`, err);
+      const isOOM =
+        err instanceof Error &&
+        (err.message.includes('out of memory') ||
+          err.message.includes('ERR_WORKER_OUT_OF_MEMORY') ||
+          err.message.includes('Allocation failed'));
+      if (isOOM) {
+        logger.error(
+          'Workflow Worker Thread ran out of memory. ' +
+            'Consider reducing maxCachedWorkflows or setting maxWorkflowThreadHeapMiB.',
+          err
+        );
+      } else {
+        logger.error(`Workflow Worker Thread failed: ${err}`, err);
+      }
       this.exitError = new UnexpectedError(`Workflow Worker Thread exited prematurely: ${err}`, err);
-      // Node will automatically terminate the Worker Thread, immediately after this event.
     });
     workerThread.on('exit', (exitCode) => {
       logger.trace(`Workflow Worker Thread exited with code ${exitCode}`, { exitError: this.exitError });
@@ -216,6 +228,7 @@ export interface ThreadedVMWorkflowCreatorOptions {
   registeredActivityNames: Set<string>;
   logger: Logger;
   patchActivationCallback?: PatchActivationCallback;
+  maxOldGenerationSizeMb?: number;
 }
 
 /**
@@ -235,13 +248,15 @@ export class ThreadedVMWorkflowCreator implements WorkflowCreator {
     registeredActivityNames,
     logger,
     patchActivationCallback,
+    maxOldGenerationSizeMb,
   }: ThreadedVMWorkflowCreatorOptions): Promise<ThreadedVMWorkflowCreator> {
+    const resourceLimits = maxOldGenerationSizeMb ? { maxOldGenerationSizeMb } : undefined;
     const workerThreadClients = Array(threadPoolSize)
       .fill(0)
       .map(
         () =>
           new WorkerThreadClient(
-            new NodeWorker(require.resolve('./workflow-worker-thread')),
+            new NodeWorker(require.resolve('./workflow-worker-thread'), { resourceLimits }),
             logger,
             patchActivationCallback
           )