feat: Add startTime to WorkflowInfo, and more (#1031)

Author: mjameswh

packages/test/src/integration-tests.ts

@@ -185,7 +185,8 @@ export function runIntegrationTests(codec?: PayloadCodec): void {
           if (
             !history?.events?.some(
               ({ workflowTaskFailedEventAttributes }) =>
-                workflowTaskFailedEventAttributes?.failure?.message === "'not-found' is not a function"
+                workflowTaskFailedEventAttributes?.failure?.message ===
+                "Failed to initialize workflow of type 'not-found': no such function is exported by the workflow bundle"
             )
           ) {
             throw new Error('Cannot find workflow task failed event');
@@ -646,7 +647,8 @@ export function runIntegrationTests(codec?: PayloadCodec): void {
       CustomKeywordField: ['test-value'],
       CustomIntField: [1, 2],
       CustomDatetimeField: [date.toISOString(), date.toISOString()],
-      datetimeInstanceofWorks: [false],
+      datetimeInstanceofWorks: [true],
+      arrayInstanceofWorks: [true],
       datetimeType: ['Date'],
     });
   });
@@ -712,6 +714,8 @@ export function runIntegrationTests(codec?: PayloadCodec): void {
       workflowType: 'returnWorkflowInfo',
       workflowId,
       historyLength: 3,
+      startTime: result.startTime,
+      runStartTime: result.runStartTime,
       // unsafe.now is a function, so doesn't make it through serialization, but .now is required, so we need to cast
       unsafe: { isReplaying: false } as UnsafeWorkflowInfo,
     });

packages/test/src/test-sinks.ts

@@ -33,34 +33,43 @@ if (RUN_INTEGRATION_TESTS) {
       fn: string;
     }
 
+    const dummyDate = new Date(2000, 1, 0, 0, 0, 0);
+    function fixWorkflowInfoDates(input: WorkflowInfo): WorkflowInfo {
+      return {
+        ...input,
+        startTime: dummyDate,
+        runStartTime: dummyDate,
+      };
+    }
+
     const recordedCalls: RecordedCall[] = [];
     const taskQueue = `${__filename}-${t.title}`;
     const thrownErrors = Array<DependencyError>();
     const sinks: InjectedSinks<workflows.TestSinks> = {
       success: {
         runAsync: {
           async fn(info, counter) {
-            recordedCalls.push({ info, counter, fn: 'success.runAsync' });
+            recordedCalls.push({ info: fixWorkflowInfoDates(info), counter, fn: 'success.runAsync' });
           },
         },
         runSync: {
           fn(info, counter) {
-            recordedCalls.push({ info, counter, fn: 'success.runSync' });
+            recordedCalls.push({ info: fixWorkflowInfoDates(info), counter, fn: 'success.runSync' });
           },
         },
       },
       error: {
         throwAsync: {
           async fn(info, counter) {
-            recordedCalls.push({ info, counter, fn: 'error.throwAsync' });
+            recordedCalls.push({ info: fixWorkflowInfoDates(info), counter, fn: 'error.throwAsync' });
             const error = new DependencyError('error', 'throwAsync');
             thrownErrors.push(error);
             throw error;
           },
         },
         throwSync: {
           fn(info, counter) {
-            recordedCalls.push({ info, counter, fn: 'error.throwSync' });
+            recordedCalls.push({ info: fixWorkflowInfoDates(info), counter, fn: 'error.throwSync' });
             const error = new DependencyError('error', 'throwSync');
             thrownErrors.push(error);
             throw error;
@@ -102,6 +111,8 @@ if (RUN_INTEGRATION_TESTS) {
       parent: undefined,
       searchAttributes: {},
       historyLength: 3,
+      startTime: dummyDate,
+      runStartTime: dummyDate,
       // unsafe.now() doesn't make it through serialization, but .now is required, so we need to cast
       unsafe: { isReplaying: false } as UnsafeWorkflowInfo,
     };
@@ -114,7 +125,13 @@ if (RUN_INTEGRATION_TESTS) {
     ]);
 
     t.deepEqual(
-      recordedLogs,
+      recordedLogs.map((x) => ({
+        ...x,
+        meta: {
+          ...x.meta,
+          workflowInfo: fixWorkflowInfoDates(x.meta.workflowInfo),
+        },
+      })),
       thrownErrors.map((error) => ({
         level: 'ERROR',
         message: 'External sink function threw an error',

packages/test/src/test-workflows.ts

@@ -94,6 +94,8 @@ async function createWorkflow(
       searchAttributes: {},
       historyLength: 3,
       unsafe: { isReplaying: false, now: Date.now },
+      startTime: new Date(),
+      runStartTime: new Date(),
     },
     randomnessSeed: Long.fromInt(1337).toBytes(),
     now: startTime,

packages/test/src/workflows/return-search-attributes.ts

@@ -7,5 +7,6 @@ export async function returnSearchAttributes(): Promise<SearchAttributes | undef
     ...sa,
     datetimeType: [Object.getPrototypeOf(datetime).constructor.name],
     datetimeInstanceofWorks: [datetime instanceof Date],
+    arrayInstanceofWorks: [sa.CustomIntField instanceof Array],
   };
 }

packages/worker/src/worker-options.ts

@@ -457,8 +457,8 @@ export function appendDefaultInterceptors(
 }
 
 export function addDefaultWorkerOptions(options: WorkerOptions): WorkerOptionsWithDefaults {
-  const { maxCachedWorkflows, showStackTraceSources, debugMode, namespace, ...rest } = options;
-
+  const { maxCachedWorkflows, showStackTraceSources, namespace, ...rest } = options;
+  const debugMode = options.debugMode || isSet(process.env.TEMPORAL_DEBUG);
   return {
     namespace: namespace ?? 'default',
     identity: `${process.pid}@${os.hostname()}`,
@@ -484,6 +484,12 @@ export function addDefaultWorkerOptions(options: WorkerOptions): WorkerOptionsWi
   };
 }
 
+function isSet(env: string | undefined): boolean {
+  if (env === undefined) return false;
+  env = env.toLocaleLowerCase();
+  return env === '1' || env === 't' || env === 'true';
+}
+
 export function compileWorkerOptions(opts: WorkerOptionsWithDefaults): CompiledWorkerOptions {
   return {
     ...opts,

packages/worker/src/worker.ts

@@ -50,7 +50,7 @@ import {
   TASK_TOKEN_ATTR_KEY,
 } from '@temporalio/common/lib/otel';
 import { historyFromJSON } from '@temporalio/common/lib/proto-utils';
-import { optionalTsToDate, optionalTsToMs, tsToMs } from '@temporalio/common/lib/time';
+import { optionalTsToDate, optionalTsToMs, tsToDate, tsToMs } from '@temporalio/common/lib/time';
 import { errorMessage } from '@temporalio/common/lib/type-helpers';
 import * as native from '@temporalio/core-bridge';
 import { UnexpectedError } from '@temporalio/core-bridge';
@@ -254,7 +254,7 @@ function formatTaskToken(taskToken: Uint8Array) {
 /**
  * Notify that an activity has started, used as input to {@link Worker.activityHeartbeatSubject}
  *
- * Used to detect rouge activities.
+ * Used to detect rogue activities.
  */
 interface HeartbeatCreateNotification {
   type: 'create';
@@ -575,7 +575,10 @@ export class Worker {
     }
     const workflowCreator = await this.createWorkflowCreator(bundle, compiledOptions);
     const replayHandle = await nativeWorkerCtor.createReplay(addBuildIdIfMissing(compiledOptions, bundle.code));
-    return [new this(replayHandle.worker, workflowCreator, compiledOptions), replayHandle.historyPusher];
+    return [
+      new this(replayHandle.worker, workflowCreator, compiledOptions, undefined, true),
+      replayHandle.historyPusher,
+    ];
   }
 
   /**
@@ -680,7 +683,8 @@ export class Worker {
      */
     protected readonly workflowCreator: WorkflowCreator | undefined,
     public readonly options: CompiledWorkerOptions,
-    protected readonly connection?: NativeConnection
+    protected readonly connection?: NativeConnection,
+    protected readonly isReplayWorker: boolean = false
   ) {
     this.tracer = getTracer(options.enableSDKTracing);
     this.workflowCodecRunner = new WorkflowCodecRunner(options.loadedDataConverter.payloadCodecs);
@@ -1133,15 +1137,16 @@ export class Worker {
                       if (
                         !(
                           startWorkflow &&
-                          startWorkflow.workflowId &&
-                          startWorkflow.workflowType &&
-                          startWorkflow.randomnessSeed
+                          startWorkflow.workflowId != null &&
+                          startWorkflow.workflowType != null &&
+                          startWorkflow.randomnessSeed != null &&
+                          startWorkflow.firstExecutionRunId != null &&
+                          startWorkflow.attempt != null &&
+                          startWorkflow.startTime != null
                         )
                       ) {
                         throw new TypeError(
-                          `Expected StartWorkflow with workflowId, workflowType and randomnessSeed, got ${JSON.stringify(
-                            maybeStartWorkflow
-                          )}`
+                          `Malformed StartWorkflow activation: ${JSON.stringify(maybeStartWorkflow)}`
                         );
                       }
                       if (activation.timestamp == null) {
@@ -1168,12 +1173,6 @@ export class Worker {
                         searchAttributes,
                       } = startWorkflow;
 
-                      if (firstExecutionRunId === null || firstExecutionRunId === undefined) {
-                        throw new TypeError(`Unexpected value: \`firstExecutionRunId\` is ${firstExecutionRunId}`);
-                      }
-                      if (attempt === null || attempt === undefined) {
-                        throw new TypeError(`Unexpected value: \`attempt\` is ${attempt}`);
-                      }
                       const workflowInfo: WorkflowInfo = {
                         workflowId,
                         runId: activation.runId,
@@ -1198,6 +1197,8 @@ export class Worker {
                         namespace: this.options.namespace,
                         firstExecutionRunId,
                         continuedFromExecutionRunId: continuedFromExecutionRunId || undefined,
+                        startTime: tsToDate(startWorkflow.startTime),
+                        runStartTime: tsToDate(activation.timestamp),
                         executionTimeoutMs: optionalTsToMs(workflowExecutionTimeout),
                         executionExpirationTime: optionalTsToDate(workflowExecutionExpirationTime),
                         runTimeoutMs: optionalTsToMs(workflowRunTimeout),
@@ -1624,7 +1625,7 @@ export class Worker {
   protected activity$(): Observable<void> {
     // This Worker did not register any activities, return early
     if (this.options.activities === undefined || Object.keys(this.options.activities).length === 0) {
-      this.log.warn('No activities registered, not polling for activity tasks');
+      if (!this.isReplayWorker) this.log.warn('No activities registered, not polling for activity tasks');
       this.activityPollerStateSubject.next('SHUTDOWN');
       return EMPTY;
     }

packages/workflow/src/interfaces.ts

@@ -81,11 +81,15 @@ export interface WorkflowInfo {
    */
   continuedFromExecutionRunId?: string;
 
-  // TODO expose from Core
   /**
-   * Time at which the Workflow Run started
+   * Time at which this [Workflow Execution Chain](https://docs.temporal.io/workflows#workflow-execution-chain) was started
    */
-  // startTime: Date;
+  startTime: Date;
+
+  /**
+   * Time at which the current Workflow Run started
+   */
+  runStartTime: Date;
 
   /**
    * Milliseconds after which the Workflow Execution is automatically terminated by Temporal Server. Set via {@link WorkflowOptions.workflowExecutionTimeout}.

packages/workflow/src/worker-interface.ts

@@ -10,7 +10,7 @@ import type { coresdk } from '@temporalio/proto';
 import { storage } from './cancellation-scope';
 import { DeterminismViolationError } from './errors';
 import { WorkflowInterceptorsFactory } from './interceptors';
-import { WorkflowCreateOptionsWithSourceMap } from './interfaces';
+import { WorkflowCreateOptionsWithSourceMap, WorkflowInfo } from './interfaces';
 import { Activator, getActivator } from './internals';
 import { SinkCall } from './sinks';
 
@@ -93,9 +93,9 @@ export function overrideGlobals(): void {
  * Sets required internal state and instantiates the workflow and interceptors.
  */
 export function initRuntime(options: WorkflowCreateOptionsWithSourceMap): void {
-  const { info } = options;
+  const info: WorkflowInfo = fixPrototypes(options.info);
   info.unsafe.now = OriginalDate.now;
-  const activator = new Activator(options);
+  const activator = new Activator({ ...options, info });
   // There's on activator per workflow instance, set it globally on the context.
   // We do this before importing any user code so user code can statically reference @temporalio/workflow functions
   // as well as Date and Math.random.
@@ -126,7 +126,7 @@ export function initRuntime(options: WorkflowCreateOptionsWithSourceMap): void {
     const factory: WorkflowInterceptorsFactory = mod.interceptors;
     if (factory !== undefined) {
       if (typeof factory !== 'function') {
-        throw new TypeError(`interceptors must be a function, got: ${factory}`);
+        throw new TypeError(`Failed to initialize workflows interceptors: expected a function, but got: '${factory}'`);
       }
       const interceptors = factory();
       activator.interceptors.inbound.push(...(interceptors.inbound ?? []));
@@ -138,11 +138,34 @@ export function initRuntime(options: WorkflowCreateOptionsWithSourceMap): void {
   const mod = importWorkflows();
   const workflow = mod[info.workflowType];
   if (typeof workflow !== 'function') {
-    throw new TypeError(`'${info.workflowType}' is not a function`);
+    const details =
+      workflow === undefined
+        ? 'no such function is exported by the workflow bundle'
+        : `expected a function, but got: '${typeof info.workflowType}'`;
+    throw new TypeError(`Failed to initialize workflow of type '${info.workflowType}': ${details}`);
   }
   activator.workflow = workflow;
 }
 
+/**
+ * Objects transfered to the VM from outside have prototypes belonging to the
+ * outer context, which means that instanceof won't work inside the VM. This
+ * function recursively walks over the content of an object, and recreate some
+ * of these objects (notably Array, Date and Objects).
+ */
+function fixPrototypes<X>(obj: X): X {
+  if (obj != null && typeof obj === 'object') {
+    switch (Object.getPrototypeOf(obj)?.constructor?.name) {
+      case 'Array':
+        return Array.from((obj as Array<unknown>).map(fixPrototypes)) as X;
+      case 'Date':
+        return new Date(obj as unknown as Date) as X;
+      default:
+        return Object.fromEntries(Object.entries(obj).map(([k, v]): [string, any] => [k, fixPrototypes(v)])) as X;
+    }
+  } else return obj;
+}
+
 /**
  * Run a chunk of activation jobs
  * @returns a boolean indicating whether job was processed or ignored