feat(workflow): Add support for default workflow handler and default signal handler (#1038)

mjameswh · web-flow · commit f6380398d3e2 · 2023-01-27T22:25:32.000-05:00
diff --git a/packages/test/src/test-default-workflow.ts b/packages/test/src/test-default-workflow.ts
@@ -0,0 +1,55 @@
+/**
+ * Test usage of a default workflow handler
+ */
+import test from 'ava';
+import { v4 as uuid4 } from 'uuid';
+import { TestWorkflowEnvironment } from '@temporalio/testing';
+import { Worker } from './helpers';
+import { existing } from './workflows/default-workflow-function';
+
+test('Default workflow handler is used if requested workflow does not exist', async (t) => {
+  const env = await TestWorkflowEnvironment.createLocal();
+  try {
+    const taskQueue = `${t.title}-${uuid4()}`;
+    const worker = await Worker.create({
+      connection: env.nativeConnection,
+      taskQueue,
+      workflowsPath: require.resolve('./workflows/default-workflow-function'),
+    });
+    await worker.runUntil(async () => {
+      const result = env.client.workflow.execute('non-existing', {
+        taskQueue,
+        workflowId: uuid4(),
+        args: ['test', 'foo', 'bar'],
+      });
+      t.is((await result).handler, 'default');
+      t.is((await result).workflowType, 'non-existing');
+      t.deepEqual((await result).args, ['test', 'foo', 'bar']);
+    });
+  } finally {
+    await env.teardown();
+  }
+});
+
+test('Default workflow handler is not used if requested workflow exists', async (t) => {
+  const env = await TestWorkflowEnvironment.createLocal();
+  try {
+    const taskQueue = `${t.title}-${uuid4()}`;
+    const worker = await Worker.create({
+      connection: env.nativeConnection,
+      taskQueue,
+      workflowsPath: require.resolve('./workflows/default-workflow-function'),
+    });
+    await worker.runUntil(async () => {
+      const result = env.client.workflow.execute(existing, {
+        taskQueue,
+        workflowId: uuid4(),
+        args: ['test', 'foo', 'bar'],
+      });
+      t.is((await result).handler, 'existing');
+      t.deepEqual((await result).args, ['test', 'foo', 'bar']);
+    });
+  } finally {
+    await env.teardown();
+  }
+});
diff --git a/packages/test/src/test-workflows.ts b/packages/test/src/test-workflows.ts
@@ -20,6 +20,7 @@ import { ReusableVMWorkflow, ReusableVMWorkflowCreator } from '@temporalio/worke
 import { parseWorkflowCode } from '@temporalio/worker/lib/worker';
 import * as activityFunctions from './activities';
 import { cleanStackTrace, REUSE_V8_CONTEXT, u8 } from './helpers';
+import { ProcessedSignal } from './workflows';
 
 export interface Context {
   workflow: VMWorkflow | ReusableVMWorkflow;
@@ -1967,3 +1968,86 @@ test('condition with timeout 0 in >1.5.0 - conditionTimeout0', async (t) => {
     );
   }
 });
+
+test('Buffered signals are dispatched to correct handler and in correct order - signalsOrdering', async (t) => {
+  const { workflowType } = t.context;
+  {
+    const completion = await activate(
+      t,
+      makeActivation(
+        undefined,
+        makeStartWorkflowJob(workflowType),
+        { signalWorkflow: { signalName: 'non-existant', input: toPayloads(defaultPayloadConverter, 1) } },
+        { signalWorkflow: { signalName: 'signalA', input: toPayloads(defaultPayloadConverter, 2) } },
+        { signalWorkflow: { signalName: 'signalA', input: toPayloads(defaultPayloadConverter, 3) } },
+        { signalWorkflow: { signalName: 'signalC', input: toPayloads(defaultPayloadConverter, 4) } },
+        { signalWorkflow: { signalName: 'signalB', input: toPayloads(defaultPayloadConverter, 5) } },
+        { signalWorkflow: { signalName: 'non-existant', input: toPayloads(defaultPayloadConverter, 6) } },
+        { signalWorkflow: { signalName: 'signalB', input: toPayloads(defaultPayloadConverter, 7) } }
+      )
+    );
+
+    // Signal handlers will be registered in the following order:
+    //
+    // Registration of handler A => Processing of signalA#2
+    // Deregistration of handler A => No more processing of signalA
+    // Registration of handler B => Processing of signalB#5, signalB#7
+    // Registration of default handler => Processing of the rest of signals, in numeric order
+    // Registration of handler C => No signal pending for handler C
+
+    compareCompletion(
+      t,
+      completion,
+      makeSuccess([
+        makeCompleteWorkflowExecution(
+          defaultPayloadConverter.toPayload([
+            { handler: 'signalA', args: [2] },
+            { handler: 'signalB', args: [5] },
+            { handler: 'signalB', args: [7] },
+            { handler: 'default', signalName: 'non-existant', args: [1] },
+            { handler: 'default', signalName: 'signalA', args: [3] },
+            { handler: 'default', signalName: 'signalC', args: [4] },
+            { handler: 'default', signalName: 'non-existant', args: [6] },
+          ] as ProcessedSignal[])
+        ),
+      ])
+    );
+  }
+});
+
+test('Buffered signals dispatch is reentrant  - signalsOrdering2', async (t) => {
+  const { workflowType } = t.context;
+  {
+    const completion = await activate(
+      t,
+      makeActivation(
+        undefined,
+        makeStartWorkflowJob(workflowType),
+        { signalWorkflow: { signalName: 'non-existant', input: toPayloads(defaultPayloadConverter, 1) } },
+        { signalWorkflow: { signalName: 'signalA', input: toPayloads(defaultPayloadConverter, 2) } },
+        { signalWorkflow: { signalName: 'signalA', input: toPayloads(defaultPayloadConverter, 3) } },
+        { signalWorkflow: { signalName: 'signalB', input: toPayloads(defaultPayloadConverter, 4) } },
+        { signalWorkflow: { signalName: 'signalB', input: toPayloads(defaultPayloadConverter, 5) } },
+        { signalWorkflow: { signalName: 'signalC', input: toPayloads(defaultPayloadConverter, 6) } },
+        { signalWorkflow: { signalName: 'signalC', input: toPayloads(defaultPayloadConverter, 7) } }
+      )
+    );
+    compareCompletion(
+      t,
+      completion,
+      makeSuccess([
+        makeCompleteWorkflowExecution(
+          defaultPayloadConverter.toPayload([
+            { handler: 'signalA', args: [2] },
+            { handler: 'signalB', args: [4] },
+            { handler: 'signalC', args: [6] },
+            { handler: 'default', signalName: 'non-existant', args: [1] },
+            { handler: 'signalA', args: [3] },
+            { handler: 'signalB', args: [5] },
+            { handler: 'signalC', args: [7] },
+          ] as ProcessedSignal[])
+        ),
+      ])
+    );
+  }
+});
diff --git a/packages/test/src/workflows/default-workflow-function.ts b/packages/test/src/workflows/default-workflow-function.ts
@@ -0,0 +1,22 @@
+import { workflowInfo } from '@temporalio/workflow';
+
+export interface WorkflowTypeAndArgs {
+  handler: string;
+  workflowType?: string;
+  args: unknown[];
+}
+
+export async function existing(...args: unknown[]): Promise<WorkflowTypeAndArgs> {
+  return {
+    handler: 'existing',
+    args,
+  };
+}
+
+export default async function (...args: unknown[]): Promise<WorkflowTypeAndArgs> {
+  return {
+    handler: 'default',
+    workflowType: workflowInfo().workflowType,
+    args,
+  };
+}
diff --git a/packages/test/src/workflows/index.ts b/packages/test/src/workflows/index.ts
@@ -72,6 +72,7 @@ export * from './shield-in-shield';
 export * from './signal-handlers-clear';
 export * from './signal-target';
 export * from './signals-are-always-processed';
+export * from './signals-ordering';
 export * from './sinks';
 export * from './sleep';
 export * from './sleep-invalid-duration';
diff --git a/packages/test/src/workflows/signals-ordering.ts b/packages/test/src/workflows/signals-ordering.ts
@@ -0,0 +1,69 @@
+/**
+ * Workflow used for testing default signal handler and signal ordering
+ *
+ * @module
+ */
+
+import { setHandler, setDefaultSignalHandler, defineSignal } from '@temporalio/workflow';
+
+const signalA = defineSignal<[number]>('signalA');
+const signalB = defineSignal<[number]>('signalB');
+const signalC = defineSignal<[number]>('signalC');
+
+export interface ProcessedSignal {
+  handler: string;
+  signalName?: string;
+  args: unknown[];
+}
+
+export async function signalsOrdering(): Promise<ProcessedSignal[]> {
+  const processedSignals: ProcessedSignal[] = [];
+
+  setHandler(signalA, (...args: unknown[]) => {
+    processedSignals.push({ handler: 'signalA', args });
+    setHandler(signalA, undefined);
+  });
+  setHandler(signalB, (...args: unknown[]) => {
+    processedSignals.push({ handler: 'signalB', args });
+  });
+  setDefaultSignalHandler((signalName: string, ...args: unknown[]) => {
+    processedSignals.push({ handler: 'default', signalName, args });
+  });
+  setHandler(signalC, (...args: unknown[]) => {
+    processedSignals.push({ handler: 'signalC', args });
+  });
+
+  return processedSignals;
+}
+
+export async function signalsOrdering2(): Promise<ProcessedSignal[]> {
+  const processedSignals: ProcessedSignal[] = [];
+
+  function handlerA(...args: unknown[]) {
+    processedSignals.push({ handler: 'signalA', args });
+    setHandler(signalA, undefined);
+    setHandler(signalB, handlerB);
+  }
+
+  function handlerB(...args: unknown[]) {
+    processedSignals.push({ handler: 'signalB', args });
+    setHandler(signalB, undefined);
+    setHandler(signalC, handlerC);
+  }
+
+  function handlerC(...args: unknown[]) {
+    processedSignals.push({ handler: 'signalC', args });
+    setHandler(signalC, undefined);
+    setDefaultSignalHandler(handlerDefault);
+  }
+
+  function handlerDefault(signalName: string, ...args: unknown[]) {
+    processedSignals.push({ handler: 'default', signalName, args });
+    setDefaultSignalHandler(undefined);
+    setHandler(signalA, handlerA);
+  }
+
+  setHandler(signalA, handlerA);
+
+  return processedSignals;
+}
diff --git a/packages/worker/src/workflow/vm-shared.ts b/packages/worker/src/workflow/vm-shared.ts
@@ -11,10 +11,6 @@ import { partition } from '../utils';
 import { Workflow } from './interface';
 import { WorkflowBundleWithSourceMapAndFilename } from './workflow-worker-thread/input';
 
-export interface ActivationContext {
-  isReplaying: boolean;
-}
-
 // Not present in @types/node for some reason
 const { promiseHooks } = v8 as any;
 
@@ -338,7 +334,7 @@ export abstract class BaseVMWorkflow implements Workflow {
     await new Promise(setImmediate);
     if (this.unhandledRejection) {
       return {
-        runId: activation.runId,
+        runId: this.activator.info.runId,
         failed: { failure: this.activator.errorToFailure(this.unhandledRejection) },
       };
     }
diff --git a/packages/workflow/src/interfaces.ts b/packages/workflow/src/interfaces.ts
@@ -1,5 +1,12 @@
 import type { RawSourceMap } from 'source-map';
-import { RetryPolicy, TemporalFailure, CommonWorkflowOptions, SearchAttributes } from '@temporalio/common';
+import {
+  RetryPolicy,
+  TemporalFailure,
+  CommonWorkflowOptions,
+  SearchAttributes,
+  SignalDefinition,
+  QueryDefinition,
+} from '@temporalio/common';
 import { checkExtends } from '@temporalio/common/lib/type-helpers';
 import type { coresdk } from '@temporalio/proto';
 
@@ -380,3 +387,21 @@ export interface WorkflowCreateOptions {
 export interface WorkflowCreateOptionsWithSourceMap extends WorkflowCreateOptions {
   sourceMap: RawSourceMap;
 }
+
+/**
+ * A handler function capable of accepting the arguments for a given SignalDefinition or QueryDefinition.
+ */
+export type Handler<
+  Ret,
+  Args extends any[],
+  T extends SignalDefinition<Args> | QueryDefinition<Ret, Args>
+> = T extends SignalDefinition<infer A>
+  ? (...args: A) => void | Promise<void>
+  : T extends QueryDefinition<infer R, infer A>
+  ? (...args: A) => R
+  : never;
+
+/**
+ * A handler function accepting signals calls for non-registered signal names.
+ */
+export type DefaultSignalHandler = (signalName: string, ...args: unknown[]) => void | Promise<void>;
diff --git a/packages/workflow/src/internals.ts b/packages/workflow/src/internals.ts
@@ -23,6 +23,7 @@ import { DeterminismViolationError, isCancellation } from './errors';
 import { QueryInput, SignalInput, WorkflowExecuteInput, WorkflowInterceptors } from './interceptors';
 import {
   ContinueAsNew,
+  DefaultSignalHandler,
   SDKInfo,
   FileSlice,
   EnhancedStackTrace,
@@ -120,7 +121,7 @@ export class Activator implements ActivationHandler {
   /**
    * Holds buffered signal calls until a handler is registered
    */
-  readonly bufferedSignals = new Map<string, coresdk.workflow_activation.ISignalWorkflow[]>();
+  readonly bufferedSignals = Array<coresdk.workflow_activation.ISignalWorkflow>();
 
   /**
    * Holds buffered query calls until a handler is registered.
@@ -136,6 +137,11 @@ export class Activator implements ActivationHandler {
    */
   readonly signalHandlers = new Map<string, WorkflowSignalType>();
 
+  /**
+   * A signal handler that catches calls for non-registered signal names.
+   */
+  defaultSignalHandler?: DefaultSignalHandler;
+
   /**
    * Source map file for looking up the source files in response to __enhanced_stack_trace
    */
@@ -526,10 +532,13 @@ export class Activator implements ActivationHandler {
 
   public async signalWorkflowNextHandler({ signalName, args }: SignalInput): Promise<void> {
     const fn = this.signalHandlers.get(signalName);
-    if (fn === undefined) {
+    if (fn) {
+      return await fn(...args);
+    } else if (this.defaultSignalHandler) {
+      return await this.defaultSignalHandler(signalName, ...args);
+    } else {
       throw new IllegalStateError(`No registered signal handler for signal ${signalName}`);
     }
-    return await fn(...args);
   }
 
   public signalWorkflow(activation: coresdk.workflow_activation.ISignalWorkflow): void {
@@ -538,14 +547,8 @@ export class Activator implements ActivationHandler {
       throw new TypeError('Missing activation signalName');
     }
 
-    const fn = this.signalHandlers.get(signalName);
-    if (fn === undefined) {
-      let buffer = this.bufferedSignals.get(signalName);
-      if (buffer === undefined) {
-        buffer = [];
-        this.bufferedSignals.set(signalName, buffer);
-      }
-      buffer.push(activation);
+    if (!this.signalHandlers.has(signalName) && !this.defaultSignalHandler) {
+      this.bufferedSignals.push(activation);
       return;
     }
 
@@ -561,6 +564,22 @@ export class Activator implements ActivationHandler {
     }).catch(this.handleWorkflowFailure.bind(this));
   }
 
+  public dispatchBufferedSignals(): void {
+    const bufferedSignals = this.bufferedSignals;
+    while (bufferedSignals.length) {
+      if (this.defaultSignalHandler) {
+        // We have a default signal handler, so all signals are dispatchable
+        // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
+        this.signalWorkflow(bufferedSignals.shift()!);
+      } else {
+        const foundIndex = bufferedSignals.findIndex((signal) => this.signalHandlers.has(signal.signalName ?? ''));
+        if (foundIndex === -1) break;
+        const [signal] = bufferedSignals.splice(foundIndex, 1);
+        this.signalWorkflow(signal);
+      }
+    }
+  }
+
   public resolveSignalExternalWorkflow(activation: coresdk.workflow_activation.IResolveSignalExternalWorkflow): void {
     const { resolve, reject } = this.consumeCompletion('signalWorkflow', getSeq(activation));
     if (activation.failure) {
diff --git a/packages/workflow/src/worker-interface.ts b/packages/workflow/src/worker-interface.ts
diff --git a/packages/workflow/src/workflow.ts b/packages/workflow/src/workflow.ts
