feat(worker): Worker.runReplayHistories returns an AsyncIterableIterator (#1067)

Author: bergundy

packages/test/src/test-replay.ts

@@ -4,13 +4,21 @@ import * as fs from 'fs';
 import path from 'path';
 import anyTest, { TestFn } from 'ava';
 import { temporal } from '@temporalio/proto';
-import { DefaultLogger, ReplayError, Runtime } from '@temporalio/worker';
+import { bundleWorkflowCode, ReplayError, WorkflowBundle } from '@temporalio/worker';
 import { DeterminismViolationError } from '@temporalio/workflow';
 import { Worker } from './helpers';
 import History = temporal.api.history.v1.History;
 
+async function gen2array<T>(gen: AsyncIterable<T>): Promise<T[]> {
+  const out: T[] = [];
+  for await (const x of gen) {
+    out.push(x);
+  }
+  return out;
+}
+
 export interface Context {
-  runtime: Runtime;
+  bundle: WorkflowBundle;
 }
 
 async function getHistories(fname: string): Promise<History> {
@@ -38,18 +46,18 @@ const test = anyTest as TestFn<Context>;
 test.before(async (t) => {
   // We don't want AVA to whine about unhandled rejections thrown by workflows
   process.removeAllListeners('unhandledRejection');
-  const logger = new DefaultLogger('DEBUG');
-  const runtime = Runtime.install({ logger });
+  const bundle = await bundleWorkflowCode({ workflowsPath: require.resolve('./workflows') });
+
   t.context = {
-    runtime,
+    bundle,
   };
 });
 
 test('cancel-fake-progress-replay', async (t) => {
   const hist = await getHistories('cancel_fake_progress_history.bin');
   await Worker.runReplayHistory(
     {
-      workflowsPath: require.resolve('./workflows'),
+      workflowBundle: t.context.bundle,
     },
     hist
   );
@@ -60,7 +68,7 @@ test('cancel-fake-progress-replay from JSON', async (t) => {
   const hist = await getHistories('cancel_fake_progress_history.json');
   await Worker.runReplayHistory(
     {
-      workflowsPath: require.resolve('./workflows'),
+      workflowBundle: t.context.bundle,
     },
     hist
   );
@@ -75,8 +83,7 @@ test('cancel-fake-progress-replay-nondeterministic', async (t) => {
   await t.throwsAsync(
     Worker.runReplayHistory(
       {
-        workflowsPath: require.resolve('./workflows'),
-        failFast: false, // Verify this flag is ignored for single replay
+        workflowBundle: t.context.bundle,
       },
       hist
     ),
@@ -91,67 +98,69 @@ test('workflow-task-failure-fails-replay', async (t) => {
   // Manually alter the workflow type to point to our workflow which will fail workflow tasks
   hist.events[0].workflowExecutionStartedEventAttributes!.workflowType!.name = 'failsWorkflowTask';
 
-  const err: ReplayError | undefined = await t.throwsAsync(
+  await t.throwsAsync(
     Worker.runReplayHistory(
       {
-        workflowsPath: require.resolve('./workflows'),
+        workflowBundle: t.context.bundle,
         replayName: t.title,
       },
       hist
     ),
     { instanceOf: ReplayError }
   );
-  t.false(err?.isNonDeterminism);
 });
 
 test('multiple-histories-replay', async (t) => {
   const hist1 = await getHistories('cancel_fake_progress_history.bin');
   const hist2 = await getHistories('cancel_fake_progress_history.json');
   const histories = historator([hist1, hist2]);
 
-  const res = await Worker.runReplayHistories(
-    {
-      workflowsPath: require.resolve('./workflows'),
-      replayName: t.title,
-    },
-    histories
+  const res = await gen2array(
+    Worker.runReplayHistories(
+      {
+        workflowBundle: t.context.bundle,
+        replayName: t.title,
+      },
+      histories
+    )
+  );
+  t.deepEqual(
+    res.map(({ error }) => error),
+    [undefined, undefined]
   );
-  t.is(res.errors.length, 0);
 });
 
-test('multiple-histories-replay-fails-fast', async (t) => {
+test('multiple-histories-replay-returns-errors', async (t) => {
   const hist1 = await getHistories('cancel_fake_progress_history.bin');
   const hist2 = await getHistories('cancel_fake_progress_history.json');
   // change workflow type to break determinism
   hist1.events[0].workflowExecutionStartedEventAttributes!.workflowType!.name = 'http';
   hist2.events[0].workflowExecutionStartedEventAttributes!.workflowType!.name = 'http';
   const histories = historator([hist1, hist2]);
 
-  await t.throwsAsync(
+  const results = await gen2array(
     Worker.runReplayHistories(
       {
-        workflowsPath: require.resolve('./workflows'),
+        workflowBundle: t.context.bundle,
         replayName: t.title,
       },
       histories
     )
   );
+
+  t.is(results.filter(({ error }) => error instanceof DeterminismViolationError).length, 2);
 });
 
-test('multiple-histories-replay-fails-slow', async (t) => {
-  const hist1 = await getHistories('cancel_fake_progress_history.bin');
-  const hist2 = await getHistories('cancel_fake_progress_history.json');
-  // change workflow type to break determinism
-  hist1.events[0].workflowExecutionStartedEventAttributes!.workflowType!.name = 'http';
-  const histories = historator([hist1, hist2]);
+test('empty-histories-replay-returns-empty-result', async (t) => {
+  const histories = historator([]);
 
-  const res = await Worker.runReplayHistories(
-    {
-      workflowsPath: require.resolve('./workflows'),
-      replayName: t.title,
-      failFast: false,
-    },
-    histories
+  const res = await gen2array(
+    Worker.runReplayHistories(
+      {
+        workflowBundle: t.context.bundle,
+      },
+      histories
+    )
   );
-  t.is(res.errors.length, 1);
+  t.is(res.length, 0);
 });

packages/worker/src/index.ts

@@ -49,7 +49,7 @@ export {
   WorkflowBundlePath,
   WorkflowBundlePathWithSourceMap, // eslint-disable-line deprecation/deprecation
 } from './worker-options';
-export { ReplayError, ReplayHistoriesIterable, ReplayResults } from './replay';
+export { ReplayError, ReplayHistoriesIterable, ReplayResult } from './replay';
 export { WorkflowInboundLogInterceptor, workflowLogAttributes } from './workflow-log-interceptor';
 export {
   BundleOptions,

packages/worker/src/replay.ts

@@ -1,5 +1,6 @@
 import { HistoryAndWorkflowId } from '@temporalio/client';
 import { coresdk } from '@temporalio/proto';
+import { DeterminismViolationError } from '@temporalio/workflow';
 
 export type EvictionReason = coresdk.workflow_activation.RemoveFromCache.EvictionReason;
 export const EvictionReason = coresdk.workflow_activation.RemoveFromCache.EvictionReason;
@@ -10,40 +11,15 @@ export type RemoveFromCache = coresdk.workflow_activation.IRemoveFromCache;
  */
 export class ReplayError extends Error {
   public readonly name = 'ReplayError';
-
-  constructor(
-    /**
-     * Workflow ID of the Workflow that failed to replay
-     */
-    public readonly workflowId: string,
-    /**
-     * Run ID of the Workflow that failed to replay
-     */
-    public runId: string,
-    /**
-     * Whether or not this error is caused by non-determinism
-     */
-    public readonly isNonDeterminism: boolean,
-    /**
-     * Why replay failed
-     */
-    message: string
-  ) {
-    super(message);
-  }
-}
-
-export interface ReplayResults {
-  readonly hasErrors: boolean;
-  /** Maps run id to information about the replay failure */
-  readonly errors: ReplayError[];
 }
 
 /**
- * @internal
+ * Result of a single workflow replay
  */
-export interface ReplayRunOptions {
-  failFast?: boolean;
+export interface ReplayResult {
+  readonly workflowId: string;
+  readonly runId: string;
+  readonly error?: ReplayError | DeterminismViolationError;
 }
 
 /**
@@ -60,28 +36,22 @@ export type ReplayHistoriesIterable = AsyncIterable<HistoryAndWorkflowId> | Iter
  *
  * @internal
  */
-export function handleReplayEviction(evictJob: RemoveFromCache, workflowId: string, runId: string): ReplayError | null {
+export function evictionReasonToReplayError(
+  evictJob: RemoveFromCache
+): ReplayError | DeterminismViolationError | undefined {
   switch (evictJob.reason) {
     case EvictionReason.NONDETERMINISM:
-      return new ReplayError(
-        workflowId,
-        runId,
-        true,
+      return new DeterminismViolationError(
         'Replay failed with a nondeterminism error. This means that the workflow code as written ' +
           `is not compatible with the history that was fed in. Details: ${evictJob.message}`
       );
     case EvictionReason.LANG_FAIL:
-      return new ReplayError(
-        workflowId,
-        runId,
-        false,
-        `Replay failed due workflow task failure. Details: ${evictJob.message}`
-      );
+      return new ReplayError(`Replay failed due workflow task failure. Details: ${evictJob.message}`);
     // Both of these reasons are not considered errors.
     // LANG_REQUESTED is used internally by Core to support duplicate runIds during replay.
     case EvictionReason.LANG_REQUESTED:
     case EvictionReason.CACHE_FULL:
-      return null;
+      return undefined;
     case undefined:
     case null:
     case EvictionReason.UNSPECIFIED:
@@ -91,9 +61,6 @@ export function handleReplayEviction(evictJob: RemoveFromCache, workflowId: stri
     case EvictionReason.PAGINATION_OR_HISTORY_FETCH:
     case EvictionReason.FATAL:
       return new ReplayError(
-        workflowId,
-        runId,
-        false,
         `Replay failed due to internal SDK issue. Code: ${
           evictJob.reason ? EvictionReason[evictJob.reason] : 'absent'
         }, Details: ${evictJob.message}`

packages/worker/src/worker-options.ts

@@ -402,20 +402,12 @@ export interface ReplayWorkerOptions
     | 'maxCachedWorkflows'
   > {
   /**
-   *  A name for this replay worker. It will be combined with a short random ID to form a unique
+   *  A optional name for this replay worker. It will be combined with an incremental ID to form a unique
    *  task queue for the replay worker.
    *
-   *  @default workflow name from given history
+   *  @default "fake_replay_queue"
    */
   replayName?: string;
-  /**
-   * If set to false, and replaying with multiple histories, do not throw an exception upon
-   * encountering a problem with replaying a workflow. Instead, finish replaying all workflows
-   * and return a results object containing the failures (if any).
-   *
-   * @default true
-   */
-  failFast?: boolean;
 }
 
 /**