Warn on unfinished handlers and provide workflow.allHandlersFinished() API (#1459)

Author: dandavison

packages/common/src/interfaces.ts

@@ -8,11 +8,16 @@ export type WorkflowUpdateType = (...args: any[]) => Promise<any> | any;
 export type WorkflowUpdateValidatorType = (...args: any[]) => void;
 export type WorkflowUpdateAnnotatedType = {
   handler: WorkflowUpdateType;
+  unfinishedPolicy: HandlerUnfinishedPolicy;
   validator?: WorkflowUpdateValidatorType;
   description?: string;
 };
 export type WorkflowSignalType = (...args: any[]) => Promise<void> | void;
-export type WorkflowSignalAnnotatedType = { handler: WorkflowSignalType; description?: string };
+export type WorkflowSignalAnnotatedType = {
+  handler: WorkflowSignalType;
+  unfinishedPolicy: HandlerUnfinishedPolicy;
+  description?: string;
+};
 export type WorkflowQueryType = (...args: any[]) => any;
 export type WorkflowQueryAnnotatedType = { handler: WorkflowQueryType; description?: string };
 
@@ -118,3 +123,22 @@ export interface HistoryAndWorkflowId {
   workflowId: string;
   history: temporal.api.history.v1.History | unknown | undefined;
 }
+
+/**
+ * Policy defining actions taken when a workflow exits while update or signal handlers are running.
+ * The workflow exit may be due to successful return, failure, cancellation, or continue-as-new.
+ */
+export enum HandlerUnfinishedPolicy {
+  /**
+   * Issue a warning in addition to abandoning the handler execution. The warning will not be issued if the workflow fails.
+   */
+  WARN_AND_ABANDON = 1,
+
+  /**
+   * Abandon the handler execution.
+   *
+   * In the case of an update handler this means that the client will receive an error rather than
+   * the update result.
+   */
+  ABANDON = 2,
+}

packages/test/src/helpers-integration.ts

@@ -1,6 +1,8 @@
 import { randomUUID } from 'crypto';
+import { status as grpcStatus } from '@grpc/grpc-js';
 import { ErrorConstructor, ExecutionContext, TestFn } from 'ava';
 import {
+  isGrpcServiceError,
   WorkflowFailedError,
   WorkflowHandle,
   WorkflowStartOptions,
@@ -12,6 +14,7 @@ import {
 } from '@temporalio/testing';
 import {
   DefaultLogger,
+  LogEntry,
   LogLevel,
   Runtime,
   WorkerOptions,
@@ -50,6 +53,7 @@ export function makeTestFunction(opts: {
   workflowsPath: string;
   workflowEnvironmentOpts?: LocalTestWorkflowEnvironmentOptions;
   workflowInterceptorModules?: string[];
+  recordedLogs?: { [workflowId: string]: LogEntry[] };
 }): TestFn<Context> {
   const test = anyTest as TestFn<Context>;
   test.before(async (t) => {
@@ -59,9 +63,15 @@ export function makeTestFunction(opts: {
       workflowsPath: opts.workflowsPath,
       logger: new DefaultLogger('WARN'),
     });
-    // Ignore invalid log levels
+    const logger = opts.recordedLogs
+      ? new DefaultLogger('DEBUG', (entry) => {
+          const workflowId = (entry.meta as any)?.workflowInfo?.workflowId ?? (entry.meta as any)?.workflowId;
+          opts.recordedLogs![workflowId] ??= [];
+          opts.recordedLogs![workflowId].push(entry);
+        })
+      : new DefaultLogger((process.env.TEST_LOG_LEVEL || 'DEBUG').toUpperCase() as LogLevel);
     Runtime.install({
-      logger: new DefaultLogger((process.env.TEST_LOG_LEVEL || 'DEBUG').toUpperCase() as LogLevel),
+      logger,
       telemetryOptions: {
         logging: {
           filter: makeTelemetryFilterString({
@@ -107,6 +117,7 @@ export interface Helpers {
   ): Promise<WorkflowHandle<T>>;
   assertWorkflowUpdateFailed(p: Promise<any>, causeConstructor: ErrorConstructor, message?: string): Promise<void>;
   assertWorkflowFailedError(p: Promise<any>, causeConstructor: ErrorConstructor, message?: string): Promise<void>;
+  updateHasBeenAdmitted(handle: WorkflowHandle<workflow.Workflow>, updateId: string): Promise<boolean>;
 }
 
 export function helpers(t: ExecutionContext<Context>, testEnv: TestWorkflowEnvironment = t.context.env): Helpers {
@@ -172,5 +183,22 @@ export function helpers(t: ExecutionContext<Context>, testEnv: TestWorkflowEnvir
         t.is(err.cause?.message, message);
       }
     },
+    async updateHasBeenAdmitted(handle: WorkflowHandle<workflow.Workflow>, updateId: string): Promise<boolean> {
+      try {
+        await testEnv.client.workflowService.pollWorkflowExecutionUpdate({
+          namespace: testEnv.client.options.namespace,
+          updateRef: {
+            workflowExecution: { workflowId: handle.workflowId },
+            updateId,
+          },
+        });
+        return true;
+      } catch (err) {
+        if (isGrpcServiceError(err) && err.code === grpcStatus.NOT_FOUND) {
+          return false;
+        }
+        throw err;
+      }
+    },
   };
 }

packages/test/src/helpers.ts

@@ -45,6 +45,23 @@ export async function sleep(ms: number): Promise<void> {
   return new Promise((resolve) => setTimeout(resolve, ms));
 }
 
+export async function waitUntil(
+  condition: () => Promise<boolean>,
+  timeoutMs: number,
+  intervalMs: number = 100
+): Promise<void> {
+  const endTime = Date.now() + timeoutMs;
+  for (;;) {
+    if (await condition()) {
+      return;
+    } else if (Date.now() >= endTime) {
+      throw new Error('timed out waiting for condition');
+    } else {
+      await sleep(intervalMs);
+    }
+  }
+}
+
 export function cleanOptionalStackTrace(stackTrace: string | undefined | null): string | undefined {
   return stackTrace ? cleanStackTrace(stackTrace) : undefined;
 }