feat(workflow): Add workflowInfo().unsafe.now() (#882)

Author: lorensr

packages/docs/docs/index.md

@@ -5,7 +5,7 @@ slug: /
 ---
 
 - Temporal docs site: [`docs.temporal.io`](https://docs.temporal.io)
-- SDK docs: [`temporal.io/ts`](https://docs.temporal.io/typescript/introduction/)
+- SDK docs: [`t.mp/ts`](https://docs.temporal.io/typescript/introduction/)
 - API reference: 👇
 
 | Package                                                                  | Description                                                                    |

packages/test/src/integration-tests.ts

@@ -42,6 +42,7 @@ import { cleanOptionalStackTrace, u8 } from './helpers';
 import * as workflows from './workflows';
 import { withZeroesHTTPServer } from './zeroes-http-server';
 import { readFileSync } from 'node:fs';
+import { UnsafeWorkflowInfo } from '@temporalio/workflow/src/interfaces';
 
 const { EVENT_TYPE_TIMER_STARTED, EVENT_TYPE_TIMER_FIRED, EVENT_TYPE_TIMER_CANCELED } =
   iface.temporal.api.enums.v1.EventType;
@@ -711,7 +712,8 @@ export function runIntegrationTests(codec?: PayloadCodec): void {
       workflowType: 'returnWorkflowInfo',
       workflowId,
       historyLength: 3,
-      unsafe: { isReplaying: false },
+      // 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

@@ -2,6 +2,7 @@
 import { WorkflowClient } from '@temporalio/client';
 import { DefaultLogger, InjectedSinks, Runtime, Worker } from '@temporalio/worker';
 import { WorkflowInfo } from '@temporalio/workflow';
+import { UnsafeWorkflowInfo } from '@temporalio/workflow/src/interfaces';
 import test from 'ava';
 import { v4 as uuid4 } from 'uuid';
 import { RUN_INTEGRATION_TESTS } from './helpers';
@@ -101,7 +102,8 @@ if (RUN_INTEGRATION_TESTS) {
       parent: undefined,
       searchAttributes: {},
       historyLength: 3,
-      unsafe: { isReplaying: false },
+      // unsafe.now() doesn't make it through serialization, but .now is required, so we need to cast
+      unsafe: { isReplaying: false } as UnsafeWorkflowInfo,
     };
 
     t.deepEqual(recordedCalls, [

packages/test/src/test-workflows.ts

@@ -1,3 +1,4 @@
+/* eslint-disable @typescript-eslint/no-non-null-assertion */
 import {
   ApplicationFailure,
   defaultPayloadConverter,
@@ -57,7 +58,6 @@ test.after.always(async (t) => {
 
 test.beforeEach(async (t) => {
   const { workflowCreator } = t.context;
-  // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
   const workflowType = t.title.match(/\S+$/)![0];
   const runId = t.title;
   const logs = new Array<unknown[]>();
@@ -93,7 +93,7 @@ async function createWorkflow(
       taskQueue: 'test',
       searchAttributes: {},
       historyLength: 3,
-      unsafe: { isReplaying: false },
+      unsafe: { isReplaying: false, now: Date.now },
     },
     randomnessSeed: Long.fromInt(1337).toBytes(),
     now: startTime,
@@ -338,9 +338,7 @@ function cleanWorkflowFailureStackTrace(
   req: coresdk.workflow_completion.WorkflowActivationCompletion,
   commandIndex = 0
 ) {
-  // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
   req.successful!.commands![commandIndex].failWorkflowExecution!.failure!.stackTrace = cleanStackTrace(
-    // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
     req.successful!.commands![commandIndex].failWorkflowExecution!.failure!.stackTrace!
   );
   return req;
@@ -350,9 +348,7 @@ function cleanWorkflowQueryFailureStackTrace(
   req: coresdk.workflow_completion.WorkflowActivationCompletion,
   commandIndex = 0
 ) {
-  // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
   req.successful!.commands![commandIndex].respondToQuery!.failed!.stackTrace = cleanStackTrace(
-    // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
     req.successful!.commands![commandIndex].respondToQuery!.failed!.stackTrace!
   );
   return req;
@@ -943,7 +939,6 @@ test('handleExternalWorkflowCancellationWhileActivityRunning', async (t) => {
   const url = 'https://temporal.io';
   const data = { content: 'new HTML content' };
   {
-    // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
     const completion = await activate(
       t,
       makeStartWorkflow(workflowType, toPayloads(defaultPayloadConverter, url, data) ?? [])

packages/worker/src/worker-options.ts

@@ -156,11 +156,12 @@ export interface WorkerOptions {
   enableNonLocalActivities?: boolean;
 
   /**
-   * Limits the number of activities per second that this worker will process. The worker will
-   * not poll for new activities if by doing so it might receive and execute an activity which
-   * would cause it to exceed this limit. Must be a positive floating point number.
+   * Limits the number of activities per second that this worker will process. The worker will not poll for new
+   * activities if by doing so it might receive and execute an activity which would cause it to exceed this limit. Must
+   * be a positive floating point number.
    *
-   * If unset, no rate limiting will be applied to Worker's activities.
+   * If unset, no rate limiting will be applied to Worker's activities. (`tctl task-queue describe` will display the
+   * absence of a limit as 100,000.)
    */
   maxActivitiesPerSecond?: number;
 

packages/worker/src/worker.ts

@@ -1115,6 +1115,7 @@ export class Worker {
                         cronScheduleToScheduleInterval: optionalTsToMs(cronScheduleToScheduleInterval) || undefined,
                         historyLength: activation.historyLength,
                         unsafe: {
+                          now: () => Date.now(), // re-set in initRuntime
                           isReplaying: activation.isReplaying,
                         },
                       };

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

@@ -180,6 +180,10 @@ export class VMWorkflowThreadProxy implements Workflow {
     workerThreadClient: WorkerThreadClient,
     options: WorkflowCreateOptions
   ): Promise<VMWorkflowThreadProxy> {
+    // Delete .now because functions can't be serialized / sent to thread.
+    // Cast to any to avoid type error, since .now is a required field.
+    // Safe to cast since we immediately set it inside the thread in initRuntime.
+    delete (options.info.unsafe as any).now;
     await workerThreadClient.send({ type: 'create-workflow', options });
     return new this(workerThreadClient, options.info.runId);
   }

packages/workflow/src/interfaces.ts

@@ -136,6 +136,14 @@ export interface WorkflowInfo {
  * Never rely on this information in Workflow logic as it will cause non-deterministic behavior.
  */
 export interface UnsafeWorkflowInfo {
+  /**
+   * Current system time in milliseconds
+   *
+   * The safe version of time is `new Date()` and `Date.now()`, which are set on the first invocation of a Workflow
+   * Task.
+   */
+  now(): number;
+
   isReplaying: boolean;
 }
 

packages/workflow/src/worker-interface.ts

@@ -39,8 +39,10 @@ export function setImportFuncs({ importWorkflows, importInterceptors }: ImportFu
   state.importInterceptors = importInterceptors;
 }
 
+const global = globalThis as any;
+const OriginalDate = globalThis.Date;
+
 export function overrideGlobals(): void {
-  const global = globalThis as any;
   // Mock any weak reference because GC is non-deterministic and the effect is observable from the Workflow.
   // WeakRef is implemented in V8 8.4 which is embedded in node >=14.6.0.
   // Workflow developer will get a meaningful exception if they try to use these.
@@ -53,8 +55,6 @@ export function overrideGlobals(): void {
     );
   };
 
-  const OriginalDate = globalThis.Date;
-
   global.Date = function (...args: unknown[]) {
     if (args.length > 0) {
       return new (OriginalDate as any)(...args);
@@ -120,7 +120,6 @@ export async function initRuntime({
   sourceMap,
   showStackTraceSources,
 }: WorkflowCreateOptionsWithSourceMap): Promise<void> {
-  const global = globalThis as any;
   // Set the runId globally on the context so it can be retrieved in the case
   // of an unhandled promise rejection.
   global.__TEMPORAL__.runId = info.runId;
@@ -131,6 +130,7 @@ export async function initRuntime({
   };
 
   state.info = info;
+  state.info.unsafe.now = OriginalDate.now;
   state.now = now;
   state.random = alea(randomnessSeed);
   state.showStackTraceSources = showStackTraceSources;