refactor(workflow)!: Move TaskInfo to WorkflowInfo (#761)

lorensr · web-flow · commit 2d9b7d1ef849 · 2022-07-19T20:35:16.000-07:00
diff --git a/packages/test/src/integration-tests.ts b/packages/test/src/integration-tests.ts
@@ -681,6 +681,8 @@ export function runIntegrationTests(codec?: PayloadCodec): void {
       searchAttributes: {},
       workflowType: 'returnWorkflowInfo',
       workflowId,
+      historyLength: 3,
+      unsafe: { isReplaying: false },
     });
   });
 
diff --git a/packages/test/src/test-sinks.ts b/packages/test/src/test-sinks.ts
@@ -100,6 +100,8 @@ if (RUN_INTEGRATION_TESTS) {
       memo: undefined,
       parent: undefined,
       searchAttributes: {},
+      historyLength: 3,
+      unsafe: { isReplaying: false },
     };
 
     t.deepEqual(recordedCalls, [
diff --git a/packages/test/src/test-workflows.ts b/packages/test/src/test-workflows.ts
@@ -91,12 +91,12 @@ async function createWorkflow(
       taskTimeoutMs: 1000,
       taskQueue: 'test',
       searchAttributes: {},
+      historyLength: 3,
+      unsafe: { isReplaying: false },
     },
     randomnessSeed: Long.fromInt(1337).toBytes(),
     now: startTime,
     patches: [],
-    isReplaying: false,
-    historyLength: 3,
   })) as VMWorkflow;
   return workflow;
 }
diff --git a/packages/test/src/workflows/log-sink-tester.ts b/packages/test/src/workflows/log-sink-tester.ts
@@ -13,13 +13,17 @@ const { logger } = wf.proxySinks<LoggerSinks>();
 
 export async function logSinkTester(): Promise<void> {
   logger.info(
-    `Workflow execution started, replaying: ${wf.taskInfo().unsafe.isReplaying}, hl: ${wf.taskInfo().historyLength}`
+    `Workflow execution started, replaying: ${wf.workflowInfo().unsafe.isReplaying}, hl: ${
+      wf.workflowInfo().historyLength
+    }`
   );
   // We rely on the test to run with max cached workflows of 1.
   // Executing this child will flush the current workflow from the cache
   // causing replay or the first sink call.
   await wf.executeChild(successString);
   logger.info(
-    `Workflow execution completed, replaying: ${wf.taskInfo().unsafe.isReplaying}, hl: ${wf.taskInfo().historyLength}`
+    `Workflow execution completed, replaying: ${wf.workflowInfo().unsafe.isReplaying}, hl: ${
+      wf.workflowInfo().historyLength
+    }`
   );
 }
diff --git a/packages/worker/src/worker.ts b/packages/worker/src/worker.ts
@@ -1086,6 +1086,10 @@ export class Worker {
                         cronSchedule: cronSchedule || undefined,
                         // 0 is the default, and not a valid value, since crons are at least a minute apart
                         cronScheduleToScheduleInterval: optionalTsToMs(cronScheduleToScheduleInterval) || undefined,
+                        historyLength: activation.historyLength,
+                        unsafe: {
+                          isReplaying: activation.isReplaying,
+                        },
                       };
                       this.log.trace('Creating workflow', workflowLogAttributes(workflowInfo));
                       const patchJobs = activation.jobs.filter((j): j is PatchJob => j.notifyHasPatch != null);
@@ -1103,8 +1107,6 @@ export class Worker {
                           randomnessSeed: randomnessSeed.toBytes(),
                           now: tsToMs(activation.timestamp),
                           patches,
-                          isReplaying: activation.isReplaying,
-                          historyLength: activation.historyLength,
                         });
                       });
 
@@ -1521,7 +1523,7 @@ export class Worker {
   }
 
   /**
-   * {@link run | Run} the Worker until `fnOrPromise` completes.
+   * Run the Worker until `fnOrPromise` completes. Then {@link shutdown} and wait for {@link run} to complete.
    *
    * @returns the result of `fnOrPromise`
    *
@@ -1544,11 +1546,14 @@ export class Worker {
   }
 
   /**
-   * Start polling on tasks, completes after graceful shutdown.
+   * Start polling on the Task Queue for tasks. Completes after graceful {@link shutdown}, once the Worker reaches the
+   * `'STOPPED'` state.
+   *
    * Throws on a fatal error or failure to shutdown gracefully.
+   *
    * @see {@link errors}
    *
-   * To stop polling call {@link shutdown} or send one of {@link Runtime.options.shutdownSignals}.
+   * To stop polling, call {@link shutdown} or send one of {@link Runtime.options.shutdownSignals}.
    */
   async run(): Promise<void> {
     if (this.state !== 'INITIALIZED') {
diff --git a/packages/worker/src/workflow/vm.ts b/packages/worker/src/workflow/vm.ts
@@ -76,7 +76,7 @@ export class VMWorkflowCreator implements WorkflowCreator {
    */
   async createWorkflow(options: WorkflowCreateOptions): Promise<Workflow> {
     const context = await this.getContext();
-    const activationContext = { isReplaying: options.isReplaying };
+    const activationContext = { isReplaying: options.info.unsafe.isReplaying };
     this.injectConsole(context, options.info, activationContext);
     const { hasSeparateMicrotaskQueue, isolateExecutionTimeoutMs } = this;
     const workflowModule: WorkflowModule = new Proxy(
diff --git a/packages/workflow/src/interfaces.ts b/packages/workflow/src/interfaces.ts
@@ -24,6 +24,8 @@ export interface WorkflowInfo {
 
   /**
    * Indexed information attached to the Workflow Execution
+   *
+   * This value may change during the lifetime of an Execution.
    */
   searchAttributes: SearchAttributes;
 
@@ -49,6 +51,15 @@ export interface WorkflowInfo {
    */
   lastFailure?: TemporalFailure;
 
+  /**
+   * Length of Workflow history up until the current Workflow Task.
+   *
+   * This value changes during the lifetime of an Execution.
+   *
+   * You may safely use this information to decide when to {@link continueAsNew}.
+   */
+  historyLength: number;
+
   /**
    * Task queue this Workflow is executing on
    */
@@ -114,6 +125,17 @@ export interface WorkflowInfo {
    * Milliseconds between Cron Runs
    */
   cronScheduleToScheduleInterval?: number;
+
+  unsafe: UnsafeWorkflowInfo;
+}
+
+/**
+ * Unsafe information about the current Workflow Execution.
+ *
+ * Never rely on this information in Workflow logic as it will cause non-deterministic behavior.
+ */
+export interface UnsafeWorkflowInfo {
+  isReplaying: boolean;
 }
 
 export interface ParentWorkflowInfo {
diff --git a/packages/workflow/src/internals.ts b/packages/workflow/src/internals.ts
@@ -476,16 +476,6 @@ export class State {
    */
   public info?: WorkflowInfo;
 
-  /**
-   * Whether a Workflow is replaying history or processing new events
-   */
-  isReplaying?: boolean;
-
-  /**
-   * ID of last WorkflowTaskStarted event
-   */
-  historyLength?: number;
-
   /**
    * A deterministic RNG, used by the isolate's overridden Math.random
    */
diff --git a/packages/workflow/src/worker-interface.ts b/packages/workflow/src/worker-interface.ts
@@ -22,8 +22,6 @@ export interface WorkflowCreateOptions {
   randomnessSeed: number[];
   now: number;
   patches: string[];
-  isReplaying: boolean;
-  historyLength: number;
 }
 
 export interface ImportFunctions {
@@ -108,14 +106,7 @@ export function overrideGlobals(): void {
  *
  * Sets required internal state and instantiates the workflow and interceptors.
  */
-export async function initRuntime({
-  info,
-  randomnessSeed,
-  now,
-  patches,
-  isReplaying,
-  historyLength,
-}: WorkflowCreateOptions): Promise<void> {
+export async function initRuntime({ info, randomnessSeed, now, patches }: WorkflowCreateOptions): 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.
@@ -129,9 +120,8 @@ export async function initRuntime({
   state.info = info;
   state.now = now;
   state.random = alea(randomnessSeed);
-  state.historyLength = historyLength;
 
-  if (isReplaying) {
+  if (info.unsafe.isReplaying) {
     for (const patch of patches) {
       state.knownPresentPatches.add(patch);
     }
@@ -193,8 +183,8 @@ export function activate(activation: coresdk.workflow_activation.WorkflowActivat
       if (activation.historyLength == null) {
         throw new TypeError('Got activation with no historyLength');
       }
-      state.isReplaying = activation.isReplaying ?? false;
-      state.historyLength = activation.historyLength;
+      state.info.unsafe.isReplaying = activation.isReplaying ?? false;
+      state.info.historyLength = activation.historyLength;
     }
 
     // Cast from the interface to the class which has the `variant` attribute.
diff --git a/packages/workflow/src/workflow.ts b/packages/workflow/src/workflow.ts
@@ -817,7 +817,29 @@ export async function executeChild<T extends Workflow>(
 }
 
 /**
- * Get information about the current Workflow
+ * Get information about the current Workflow.
+ *
+ * ⚠️ We recommend calling `workflowInfo()` whenever accessing {@link WorkflowInfo} fields. Some WorkflowInfo fields
+ * change during the lifetime of an Execution—like {@link WorkflowInfo.historyLength} and
+ * {@link WorkflowInfo.searchAttributes}—and some may be changeable in the future—like {@link WorkflowInfo.taskQueue}.
+ *
+ * ```ts
+ * // GOOD
+ * function myWorkflow() {
+ *   doSomething(workflowInfo().searchAttributes)
+ *   ...
+ *   doSomethingElse(workflowInfo().searchAttributes)
+ * }
+ * ```
+ *
+ * ```ts
+ * // BAD
+ * function myWorkflow() {
+ *   const attributes = workflowInfo().searchAttributes
+ *   doSomething(attributes)
+ *   ...
+ *   doSomethingElse(attributes)
+ * }
  */
 export function workflowInfo(): WorkflowInfo {
   if (state.info === undefined) {
@@ -1037,7 +1059,10 @@ function patchInternal(patchId: string, deprecated: boolean): boolean {
   if (state.workflow === undefined) {
     throw new IllegalStateError('Patches cannot be used before Workflow starts');
   }
-  const usePatch = !state.isReplaying || state.knownPresentPatches.has(patchId);
+  if (state.info === undefined) {
+    throw new IllegalStateError('Workflow uninitialized');
+  }
+  const usePatch = !state.info.unsafe.isReplaying || state.knownPresentPatches.has(patchId);
   // Avoid sending commands for patches core already knows about.
   // This optimization enables development of automatic patching tools.
   if (usePatch && !state.sentPatches.has(patchId)) {
@@ -1219,47 +1244,4 @@ export function upsertSearchAttributes(searchAttributes: SearchAttributes): void
   state.info.searchAttributes = mergedSearchAttributes;
 }
 
-/**
- * Unsafe information about the currently executing Workflow Task.
- *
- * Never rely on this information in Workflow logic as it will cause non-deterministic behavior.
- */
-export interface UnsafeTaskInfo {
-  isReplaying: boolean;
-}
-
-/**
- * Information about the currently executing Workflow Task.
- *
- * Meant for advanced usage.
- */
-export interface TaskInfo {
-  /**
-   * Length of Workflow history up until the current Workflow Task.
-   *
-   * You may safely use this information to decide when to {@link continueAsNew}.
-   */
-  historyLength: number;
-  unsafe: UnsafeTaskInfo;
-}
-
-/**
- * Get information about the currently executing Workflow Task.
- *
- * See {@link TaskInfo}
- */
-export function taskInfo(): TaskInfo {
-  const { isReplaying, historyLength } = state;
-  if (isReplaying == null || historyLength == null) {
-    throw new IllegalStateError('Workflow uninitialized');
-  }
-
-  return {
-    historyLength,
-    unsafe: {
-      isReplaying,
-    },
-  };
-}
-
 export const stackTraceQuery = defineQuery<string>('__stack_trace');
