fix: Fix nightly and activity failing with non ApplicationFailure (#751)

Author: bergundy

packages/common/src/failure.ts

@@ -360,14 +360,15 @@ export function errorToFailure(err: unknown, payloadConverter: PayloadConverter)
 /**
  * If `err` is an Error it is turned into an `ApplicationFailure`.
  *
- * If `err` was already a `TemporalFailure`, returns the original error.
+ * If `err` was already a `ApplicationFailure`, returns the original error.
  *
  * Otherwise returns an `ApplicationFailure` with `String(err)` as the message.
  */
-export function ensureTemporalFailure(err: unknown): TemporalFailure {
-  if (err instanceof TemporalFailure) {
+export function ensureApplicationFailure(err: unknown): ApplicationFailure {
+  if (err instanceof ApplicationFailure) {
     return err;
-  } else if (err instanceof Error) {
+  }
+  if (err instanceof Error) {
     const name = err.constructor?.name ?? err.name;
     const failure = new ApplicationFailure(err.message, name, false);
     failure.stack = err.stack;
@@ -379,6 +380,20 @@ export function ensureTemporalFailure(err: unknown): TemporalFailure {
   }
 }
 
+/**
+ * If `err` is an Error it is turned into an `ApplicationFailure`.
+ *
+ * If `err` was already a `TemporalFailure`, returns the original error.
+ *
+ * Otherwise returns an `ApplicationFailure` with `String(err)` as the message.
+ */
+export function ensureTemporalFailure(err: unknown): TemporalFailure {
+  if (err instanceof TemporalFailure) {
+    return err;
+  }
+  return ensureApplicationFailure(err);
+}
+
 /**
  * Converts a Failure proto message to a JS Error object if defined or returns undefined.
  */

packages/test/src/load/args.ts

@@ -62,6 +62,7 @@ export interface WorkerArgSpec extends Spec {
   '--server-address': typeof String;
   '--otel-url': typeof String;
   '--status-port': typeof Number;
+  '--shutdown-grace-time-ms': typeof String;
 }
 
 export const workerArgSpec: WorkerArgSpec = {
@@ -77,6 +78,7 @@ export const workerArgSpec: WorkerArgSpec = {
   '--server-address': String,
   '--otel-url': String,
   '--status-port': Number,
+  '--shutdown-grace-time-ms': String,
 };
 
 export interface WrapperArgSpec extends Spec {

packages/test/src/load/worker.ts

@@ -107,6 +107,7 @@ async function main() {
   const namespace = getRequired(args, '--ns');
   const taskQueue = getRequired(args, '--task-queue');
   const statusPort = args['--status-port'];
+  const shutdownGraceTime = args['--shutdown-grace-time'] || '30s';
 
   const telemetryOptions: TelemetryOptions = {
     tracingFilter: `temporal_sdk_core=${logLevel}`,
@@ -149,6 +150,7 @@ async function main() {
       maxConcurrentLocalActivityExecutions,
       maxConcurrentWorkflowTaskExecutions,
       maxCachedWorkflows,
+      shutdownGraceTime,
       interceptors: {
         activityInbound: [() => new ConnectionInjectorInterceptor(clientConnection)],
       },

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

@@ -1,6 +1,6 @@
 /* eslint-disable @typescript-eslint/no-non-null-assertion */
 import * as activity from '@temporalio/activity';
-import { defaultPayloadConverter, toPayloads } from '@temporalio/common';
+import { TemporalFailure, defaultPayloadConverter, toPayloads } from '@temporalio/common';
 import { coresdk } from '@temporalio/proto';
 import anyTest, { ExecutionContext, TestInterface } from 'ava';
 import dedent from 'dedent';
@@ -253,3 +253,27 @@ test('Worker cancels activities after shutdown', async (t) => {
   t.truthy(result?.failed);
   t.true(activityCancelled);
 });
+
+test('Non ApplicationFailure TemporalFailures thrown from Activity are wrapped with ApplicationFailure', async (t) => {
+  const worker = isolateFreeWorker({
+    ...defaultOptions,
+    activities: {
+      async throwTemporalFailure() {
+        throw new TemporalFailure('I should be valid');
+      },
+    },
+  });
+  t.context.worker = worker;
+
+  await runWorker(t, async () => {
+    const taskToken = Buffer.from(uuid4());
+    const { result } = await worker.native.runActivityTask({
+      taskToken,
+      start: {
+        activityType: 'throwTemporalFailure',
+        input: toPayloads(defaultPayloadConverter),
+      },
+    });
+    t.is(result?.failed?.failure?.applicationFailureInfo?.type, 'TemporalFailure');
+  });
+});

packages/worker/src/activity.ts

@@ -2,7 +2,7 @@ import { asyncLocalStorage, Context, Info } from '@temporalio/activity';
 import {
   ApplicationFailure,
   CancelledFailure,
-  ensureTemporalFailure,
+  ensureApplicationFailure,
   FAILURE_SOURCE,
   LoadedDataConverter,
 } from '@temporalio/common';
@@ -96,7 +96,7 @@ export class Activity {
         }
         return {
           failed: {
-            failure: await encodeErrorToFailure(this.dataConverter, ensureTemporalFailure(err)),
+            failure: await encodeErrorToFailure(this.dataConverter, ensureApplicationFailure(err)),
           },
         };
       }

packages/worker/src/worker-options.ts

@@ -93,7 +93,7 @@ export interface WorkerOptions {
    * Time to wait for pending tasks to drain after shutdown was requested.
    *
    * @format {@link https://www.npmjs.com/package/ms | ms} formatted string or number of milliseconds
-   * @default 5s
+   * @default 10s
    */
   shutdownGraceTime?: string | number;
 
@@ -323,7 +323,7 @@ export function addDefaultWorkerOptions(options: WorkerOptions): WorkerOptionsWi
   return {
     namespace: 'default',
     identity: `${process.pid}@${os.hostname()}`,
-    shutdownGraceTime: '5s',
+    shutdownGraceTime: '10s',
     maxConcurrentActivityTaskExecutions: 100,
     maxConcurrentLocalActivityExecutions: 100,
     enableNonLocalActivities: true,