fix(worker): Don't fail Worker on Activity Task decode failures (#1473)

Author: mjameswh

packages/client/src/helpers.ts

@@ -7,7 +7,7 @@ import {
   SearchAttributes,
 } from '@temporalio/common';
 import { Replace } from '@temporalio/common/lib/type-helpers';
-import { optionalTsToDate, tsToDate } from '@temporalio/common/lib/time';
+import { optionalTsToDate, requiredTsToDate } from '@temporalio/common/lib/time';
 import { decodeMapFromPayloads } from '@temporalio/common/lib/internal-non-workflow/codec-helpers';
 import { temporal, google } from '@temporalio/proto';
 import { RawWorkflowExecutionInfo, WorkflowExecutionInfo, WorkflowExecutionStatusName } from './types';
@@ -62,7 +62,7 @@ export async function executionInfoFromRaw<T>(
     // Exact truncation for multi-petabyte histories
     // historySize === 0 means WFT was generated by pre-1.20.0 server, and the history size is unknown
     historySize: raw.historySizeBytes?.toNumber() || undefined,
-    startTime: tsToDate(raw.startTime!),
+    startTime: requiredTsToDate(raw.startTime, 'startTime'),
     executionTime: optionalTsToDate(raw.executionTime),
     closeTime: optionalTsToDate(raw.closeTime),
     memo: await decodeMapFromPayloads(dataConverter, raw.memo?.fields),

packages/client/src/schedule-client.ts

@@ -8,7 +8,13 @@ import {
   filterNullAndUndefined,
 } from '@temporalio/common/lib/internal-non-workflow';
 import { temporal } from '@temporalio/proto';
-import { optionalDateToTs, optionalTsToDate, optionalTsToMs, tsToDate } from '@temporalio/common/lib/time';
+import {
+  optionalDateToTs,
+  optionalTsToDate,
+  optionalTsToMs,
+  requiredTsToDate,
+  tsToDate,
+} from '@temporalio/common/lib/time';
 import { SymbolBasedInstanceOfError } from '@temporalio/common/lib/type-helpers';
 import { CreateScheduleInput, CreateScheduleOutput, ScheduleClientInterceptor } from './interceptors';
 import { WorkflowService } from './types';
@@ -430,8 +436,7 @@ export class ScheduleClient extends BaseClient {
           info: {
             recentActions: decodeScheduleRecentActions(raw.info?.recentActions),
             nextActionTimes: raw.info?.futureActionTimes?.map(tsToDate) ?? [],
-            // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
-            createdAt: tsToDate(raw.info!.createTime!),
+            createdAt: requiredTsToDate(raw.info?.createTime, 'createTime'),
             lastUpdatedAt: optionalTsToDate(raw.info?.updateTime),
             runningActions: decodeScheduleRunningActions(raw.info?.runningWorkflows),
             numActionsMissedCatchupWindow: raw.info?.missedCatchupWindow?.toNumber() ?? 0,

packages/client/src/schedule-helpers.ts

@@ -23,7 +23,7 @@ import {
   optionalDateToTs,
   optionalTsToDate,
   optionalTsToMs,
-  tsToDate,
+  requiredTsToDate,
 } from '@temporalio/common/lib/time';
 import {
   CalendarSpec,
@@ -405,10 +405,8 @@ export function decodeScheduleRecentActions(
       } else throw new TypeError('Unsupported schedule action');
 
       return {
-        // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
-        scheduledAt: tsToDate(executionResult.scheduleTime!),
-        // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
-        takenAt: tsToDate(executionResult.actualTime!),
+        scheduledAt: requiredTsToDate(executionResult.scheduleTime, 'scheduleTime'),
+        takenAt: requiredTsToDate(executionResult.actualTime, 'actualTime'),
         action,
       };
     }

packages/common/src/time.ts

@@ -28,6 +28,17 @@ export function optionalTsToMs(ts: Timestamp | null | undefined): number | undef
   return tsToMs(ts);
 }
 
+/**
+ * Lossy conversion function from Timestamp to number due to possible overflow.
+ * If ts is null or undefined, throws a TypeError, with error message including the name of the field.
+ */
+export function requiredTsToMs(ts: Timestamp | null | undefined, fieldName: string): number {
+  if (ts === undefined || ts === null) {
+    throw new TypeError(`Expected ${fieldName} to be a timestamp, got ${ts}`);
+  }
+  return tsToMs(ts);
+}
+
 /**
  * Lossy conversion function from Timestamp to number due to possible overflow
  */
@@ -83,6 +94,11 @@ export function tsToDate(ts: Timestamp): Date {
   return new Date(tsToMs(ts));
 }
 
+// ts-prune-ignore-next
+export function requiredTsToDate(ts: Timestamp | null | undefined, fieldName: string): Date {
+  return new Date(requiredTsToMs(ts, fieldName));
+}
+
 export function optionalTsToDate(ts: Timestamp | null | undefined): Date | undefined {
   if (ts === undefined || ts === null) {
     return undefined;

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

@@ -222,6 +222,66 @@ test('Worker fails activity with proper message when it is not registered', asyn
   });
 });
 
+test('Worker fails activity with proper message if activity info contains null ScheduledTime', async (t) => {
+  const worker = isolateFreeWorker({
+    ...defaultOptions,
+    activities: {
+      async dummy(): Promise<void> {},
+    },
+  });
+  t.context.worker = worker;
+
+  await runWorker(t, async () => {
+    const taskToken = Buffer.from(uuid4());
+    const { result } = await worker.native.runActivityTask({
+      taskToken,
+      start: {
+        activityType: 'dummy',
+        workflowExecution: { workflowId: 'wfid', runId: 'runId' },
+        input: toPayloads(defaultPayloadConverter),
+        scheduledTime: null,
+      },
+    });
+    t.is(worker.getState(), 'RUNNING');
+    t.is(result?.failed?.failure?.applicationFailureInfo?.type, 'TypeError');
+    t.is(result?.failed?.failure?.message, 'Expected scheduledTime to be a timestamp, got null');
+    t.true(/worker\.[jt]s/.test(result?.failed?.failure?.stackTrace ?? ''));
+  });
+});
+
+test('Worker fails activity task if interceptor factory throws', async (t) => {
+  const worker = isolateFreeWorker({
+    ...defaultOptions,
+    activities: {
+      async dummy(): Promise<void> {},
+    },
+    interceptors: {
+      activity: [
+        () => {
+          throw new Error('I am a bad interceptor');
+        },
+      ],
+    },
+  });
+  t.context.worker = worker;
+
+  await runWorker(t, async () => {
+    const taskToken = Buffer.from(uuid4());
+    const { result } = await worker.native.runActivityTask({
+      taskToken,
+      start: {
+        activityType: 'dummy',
+        workflowExecution: { workflowId: 'wfid', runId: 'runId' },
+        input: toPayloads(defaultPayloadConverter),
+      },
+    });
+    t.is(worker.getState(), 'RUNNING');
+    t.is(result?.failed?.failure?.applicationFailureInfo?.type, 'Error');
+    t.is(result?.failed?.failure?.message, 'I am a bad interceptor');
+    t.true(/test-worker-activities\.[tj]s/.test(result?.failed?.failure?.stackTrace ?? ''));
+  });
+});
+
 test('Non ApplicationFailure TemporalFailures thrown from Activity are wrapped with ApplicationFailure', async (t) => {
   const worker = isolateFreeWorker({
     ...defaultOptions,