fix(worker): Gracefully shut down Worker on unexpected errors (#1539)

Author: mjameswh

package.json

@@ -23,7 +23,7 @@
   "scripts": {
     "rebuild": "npm run clean && npm run build",
     "build": "lerna run --stream build",
-    "build.watch": "npm run build:protos && tsc --build --watch packages/*",
+    "build.watch": "npm run build:protos && tsc --build --watch packages/*/tsconfig.json",
     "build:protos": "node ./packages/proto/scripts/compile-proto.js",
     "test": "lerna run --stream test",
     "test.watch": "lerna run --stream test.watch",

packages/client/src/workflow-client.ts

@@ -308,7 +308,7 @@ function ensureArgs<W extends Workflow, T extends WorkflowStartOptions<W>>(
   opts: T
 ): Omit<T, 'args'> & { args: unknown[] } {
   const { args, ...rest } = opts;
-  return { args: args ?? [], ...rest };
+  return { args: (args as unknown[]) ?? [], ...rest };
 }
 
 /**

packages/core-bridge/ts/errors.ts

@@ -18,7 +18,14 @@ export class TransportError extends Error {}
  * Something unexpected happened, considered fatal
  */
 @SymbolBasedInstanceOfError('UnexpectedError')
-export class UnexpectedError extends Error {}
+export class UnexpectedError extends Error {
+  constructor(
+    message: string,
+    public cause?: unknown
+  ) {
+    super(message);
+  }
+}
 
 export { IllegalStateError };
 
@@ -47,7 +54,7 @@ export function convertFromNamedError(e: unknown, keepStackTrace: boolean): unkn
         return newerr;
 
       case 'UnexpectedError':
-        newerr = new UnexpectedError(e.message);
+        newerr = new UnexpectedError(e.message, e);
         newerr.stack = keepStackTrace ? e.stack : undefined;
         return newerr;
     }

packages/test/src/test-integration-workflows-with-recorded-logs.ts

@@ -380,7 +380,7 @@ class UnfinishedHandlersWorkflowTerminationTypeTest {
     switch (this.handlerType) {
       case 'update':
         executeUpdate = w.executeUpdate(unfinishedHandlersWorkflowTerminationTypeUpdate, { updateId });
-        await waitUntil(() => workflowUpdateExists(w, updateId), 2000);
+        await waitUntil(() => workflowUpdateExists(w, updateId), 5000);
         break;
       case 'signal':
         await w.signal(unfinishedHandlersWorkflowTerminationTypeSignal);

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

@@ -4,11 +4,15 @@
  *
  * @module
  */
+import { setTimeout } from 'timers/promises';
+import { randomUUID } from 'crypto';
 import test from 'ava';
-import { Runtime } from '@temporalio/worker';
-import { TransportError } from '@temporalio/core-bridge';
+import { Runtime, PromiseCompletionTimeoutError } from '@temporalio/worker';
+import { TransportError, UnexpectedError } from '@temporalio/core-bridge';
+import { Client } from '@temporalio/client';
 import { RUN_INTEGRATION_TESTS, Worker } from './helpers';
 import { defaultOptions, isolateFreeWorker } from './mock-native-worker';
+import { fillMemory } from './workflows';
 
 if (RUN_INTEGRATION_TESTS) {
   test.serial('Worker shuts down gracefully', async (t) => {
@@ -28,6 +32,23 @@ if (RUN_INTEGRATION_TESTS) {
     await t.throwsAsync(worker.run(), { message: 'Poller was already started' });
   });
 
+  test.serial("Worker.runUntil doesn't hang if provided promise survives to Worker's shutdown", async (t) => {
+    const worker = await Worker.create({
+      ...defaultOptions,
+      taskQueue: t.title.replace(/ /g, '_'),
+    });
+    const p = worker.runUntil(
+      new Promise(() => {
+        /* a promise that will never unblock */
+      })
+    );
+    t.is(worker.getState(), 'RUNNING');
+    worker.shutdown();
+    t.is(worker.getState(), 'STOPPING');
+    await t.throwsAsync(p, { instanceOf: PromiseCompletionTimeoutError });
+    t.is(worker.getState(), 'STOPPED');
+  });
+
   test.serial('Worker shuts down gracefully if interrupted before running', async (t) => {
     const worker = await Worker.create({
       ...defaultOptions,
@@ -54,6 +75,55 @@ if (RUN_INTEGRATION_TESTS) {
       }
     );
   });
+
+  test.serial('Threaded VM gracely stops and fails on ERR_WORKER_OUT_OF_MEMORY', async (t) => {
+    // We internally use a timeout of 10s to catch a possible case where test would
+    // be non-conclusive. We need the test timeout to be longer than that.
+    t.timeout(30_000);
+
+    const taskQueue = t.title.replace(/ /g, '_');
+    const client = new Client();
+    const worker = await Worker.create({
+      ...defaultOptions,
+      taskQueue,
+    });
+
+    // This workflow will allocate large block of memory, hopefully causing a ERR_WORKER_OUT_OF_MEMORY.
+    // Note that due to the way Node/V8 optimize byte code, its possible that this may trigger
+    // other type of errors, including some that can't be intercepted cleanly.
+    client.workflow
+      .start(fillMemory, {
+        taskQueue,
+        workflowId: randomUUID(),
+        // Don't linger
+        workflowExecutionTimeout: '30s',
+      })
+      .catch(() => void 0);
+
+    const workerRunPromise = worker.run();
+    try {
+      // Due to various environment factors, it is possible that the worker may sometime not fail.
+      // That's obviously not what we want to assert, but that's still ok. We therefore set a
+      // timeout of 10s and simply pass if the Worker hasn't failed by then.
+      await Promise.race([setTimeout(10_000), workerRunPromise]);
+      if (worker.getState() === 'RUNNING') {
+        worker.shutdown();
+        await workerRunPromise;
+      }
+      t.log('Non-conclusive result: Worker did not fail as expected');
+      t.pass();
+    } catch (err) {
+      t.is((err as Error).name, UnexpectedError.name);
+      t.is(
+        (err as Error).message,
+        'Workflow Worker Thread exited prematurely: Error [ERR_WORKER_OUT_OF_MEMORY]: ' +
+          'Worker terminated due to reaching memory limit: JS heap out of memory'
+      );
+      t.is(worker.getState(), 'FAILED');
+    } finally {
+      if (Runtime._instance) await Runtime._instance.shutdown();
+    }
+  });
 }
 
 test.serial('Mocked run shuts down gracefully', async (t) => {

packages/test/src/test-workflow-unhandled-rejection-crash.ts

@@ -17,13 +17,17 @@ if (RUN_INTEGRATION_TESTS) {
       taskQueue,
       args: [{ crashWorker: true }],
     });
-    await t.throwsAsync(worker.run(), {
-      instanceOf: UnexpectedError,
-      message:
-        'Worker thread shut down prematurely, this could be caused by an' +
-        ' unhandled rejection in workflow code that could not be' +
-        ' associated with a workflow run',
-    });
-    await handle.terminate();
+    try {
+      await t.throwsAsync(worker.run(), {
+        instanceOf: UnexpectedError,
+        message:
+          'Workflow Worker Thread exited prematurely: UnhandledRejectionError: ' +
+          "Unhandled Promise rejection for unknown Workflow Run id='undefined': " +
+          'Error: error to crash the worker',
+      });
+      t.is(worker.getState(), 'FAILED');
+    } finally {
+      await handle.terminate();
+    }
   });
 }

packages/test/src/workflows/fill-memory.ts

@@ -0,0 +1,30 @@
+export async function fillMemory(): Promise<number> {
+  // There are different ways that Node may report "out of memory" conditions, and the criterias
+  // for each of them are essentially undocumented. The following code has been empirically
+  // designed to trigger a very specific type of error, i.e. a Worker Thread getting killed with an
+  // `ERR_WORKER_OUT_OF_MEMORY` code, with very high probability (more than 95% of the time across
+  // all of our test setups, including all platforms, and Node.js v16, v18 and v20). The size of
+  // RAM appears not to be important.
+  //
+  // Occurence of an `ERR_WORKER_OUT_OF_MEMORY` error in the following code appears to be the
+  // result of some interractions of V8's JIT compiler and the V8's heap allocator. With appropriate
+  // parameters, the code crashes on the first execution of the loop; yet, removing the loop or
+  // reducing the upper bound of the loop (exact threshold appears to be between 64 and 128) prevents
+  // the crash, presumably because the loop might no longer be deemed worth of early JIT compilation.
+  // Allocating larger arrays is also likely to prevent the crash, presumably because the arrays will
+  // then be allocated from the "large objects space" rather than the "yound gen" space, which, by
+  // default, has a limit of 16MB on 64-bit systems.
+  //
+  // We also need to prevent the JIT compiler from optimizing this code _before_ we even start
+  // executing it, as that then results in a low level error (native-like), which may terminate the
+  // whole process rather than just the Worker Thread. Hence, we wrap the code in an `eval` block.
+
+  const accumulator = [];
+  eval(`
+    for (let i = 0; i < 2048; i++) {
+      accumulator.push(new Array(1024 * 1024 * 2).fill(i));
+    }
+  `);
+
+  return accumulator.length;
+}

packages/test/src/workflows/index.ts

@@ -38,6 +38,7 @@ export * from './deprecate-patch';
 export * from './fail-signal';
 export * from './fail-unless-signaled-before-start';
 export * from './fails-workflow-task';
+export * from './fill-memory';
 export * from './global-overrides';
 export * from './handle-external-workflow-cancellation-while-activity-running';
 export * from './http';

packages/worker/src/errors.ts

@@ -8,6 +8,63 @@ import { ShutdownError, TransportError, UnexpectedError } from '@temporalio/core
 @SymbolBasedInstanceOfError('GracefulShutdownPeriodExpiredError')
 export class GracefulShutdownPeriodExpiredError extends Error {}
 
+/**
+ * Thrown from the Workflow Worker when a Promise is rejected, but there is no `catch` handler
+ * for that Promise. This error wraps the original error that was thrown from the Promise.
+ *
+ * Occurrence of this error generally indicate a missing `await` statement on a call that return
+ * a Promise. To silent rejections on a specific Promise, use `promise.catch(funcThatCantThrow)`
+ * (e.g. `promise.catch(() => void 0)` or `promise.catch((e) => logger.error(e))`).
+ */
+// FIXME: At this time, this wrapper is only used for errors that could not be associated with a
+//        specific workflow run; it should also be used for unhandled rejections in workflow code,
+//        but this is not possible at the moment as we intentionally "unhandle" non-TemporalFailure
+//        errors happening in workflow code (i.e. ALL non-TemporalFailure errors thrown from
+//        workflow code becomes Unhandled Rejection at some point in our own logic)
+@SymbolBasedInstanceOfError('UnhandledRejectionError')
+export class UnhandledRejectionError extends Error {
+  constructor(
+    message: string,
+    public cause: unknown
+  ) {
+    super(message);
+  }
+}
+
+/**
+ * Combined error information for {@link Worker.runUntil}
+ */
+export interface CombinedWorkerRunErrorCause {
+  /**
+   * Error thrown by a Worker
+   */
+  workerError: unknown;
+  /**
+   * Error thrown by the wrapped promise or function
+   */
+  innerError: unknown;
+}
+
+/**
+ * Error thrown by {@link Worker.runUntil} and {@link Worker.runReplayHistories}
+ */
+@SymbolBasedInstanceOfError('CombinedWorkerRunError')
+export class CombinedWorkerRunError extends Error {
+  public readonly cause: CombinedWorkerRunErrorCause;
+
+  constructor(message: string, { cause }: { cause: CombinedWorkerRunErrorCause }) {
+    super(message);
+    this.cause = cause;
+  }
+}
+
+/**
+ * Error thrown by {@link Worker.runUntil} if the provided Promise does not resolve within the specified
+ * {@link RunUntilOptions.promiseCompletionTimeout|timeout period} after the Worker has stopped.
+ */
+@SymbolBasedInstanceOfError('PromiseCompletionTimeoutError')
+export class PromiseCompletionTimeoutError extends Error {}
+
 /**
  * @deprecated Import error classes directly
  */