fix: Fail workflow task if local activity not registered with worker (#1152)

Previously the worker would needlessly retry these local activities.
With this change, the worker fails the workflow task allowing a user to
deploy a fix.

Replaces #1150 with
better behavior.
Matches Java behavior temporalio/sdk-java#1628.

Author: bergundy

packages/common/src/converter/payload-converter.ts

@@ -1,4 +1,3 @@
-import type { temporal } from '@temporalio/proto';
 import { decode, encode } from '../encoding';
 import { PayloadConverterError, ValueError } from '../errors';
 import { Payload } from '../interfaces';

packages/test/src/test-local-activities.ts

@@ -141,7 +141,7 @@ export async function runMyLocalActivityWithOption(
   return await workflow.proxyLocalActivities(opts).myLocalActivity();
 }
 
-test.serial('Local activity with ', async (t) => {
+test.serial('Local activity with various timeouts', async (t) => {
   const { executeWorkflow, createWorker } = helpers(t);
   const worker = await createWorker({
     activities: {
@@ -523,34 +523,43 @@ test.serial('Local activity can be intercepted', async (t) => {
   });
 });
 
-export async function runANonExisitingLocalActivity(): Promise<void> {
-  // TODO: default behavior should be to not retry activities that are not found
+export async function runNonExisitingLocalActivity(): Promise<void> {
   const { activityNotFound } = workflow.proxyLocalActivities({
     startToCloseTimeout: '1m',
-    retry: { maximumAttempts: 1 },
   });
 
-  await activityNotFound();
+  try {
+    await activityNotFound();
+  } catch (err) {
+    if (err instanceof ReferenceError) {
+      return;
+    }
+    throw err;
+  }
+  throw ApplicationFailure.nonRetryable('Unreachable');
 }
 
-test.serial('Local activity fails if not registered on Worker', async (t) => {
+test.serial('Local activity not registered on Worker throws ReferenceError in workflow context', async (t) => {
   const { executeWorkflow, createWorker } = helpers(t);
+  const worker = await createWorker();
+  await worker.runUntil(executeWorkflow(runNonExisitingLocalActivity));
+  t.pass();
+});
+
+test.serial('Local activity not registered on replay Worker does not throw', async (t) => {
+  const { startWorkflow, createWorker } = helpers(t);
   const worker = await createWorker({
     activities: {
-      // We need at least one activity, so that the Worker start its activity poller
-      async dummy(): Promise<string> {
-        return 'dummy';
+      async echo(input: string) {
+        return input;
       },
     },
   });
-  await worker.runUntil(async () => {
-    const err: WorkflowFailedError | undefined = await t.throwsAsync(
-      executeWorkflow(runANonExisitingLocalActivity, {}),
-      { instanceOf: WorkflowFailedError }
-    );
-    t.true(err?.cause instanceof ApplicationFailure && !err.cause.nonRetryable);
-    t.truthy(err?.cause?.message?.startsWith('Activity function activityNotFound is not registered on this Worker'));
-  });
+  const handle = await startWorkflow(runOneLocalActivity, { args: ['hello'] });
+  await worker.runUntil(() => handle.result());
+  const history = await handle.fetchHistory();
+  await Worker.runReplayHistory({ workflowBundle: t.context.workflowBundle }, history, handle.workflowId);
+  t.pass();
 });
 
 /**

packages/test/src/test-workflows.ts

@@ -64,8 +64,8 @@ test.before(async (t) => {
   const bundler = new WorkflowCodeBundler({ workflowsPath });
   const workflowBundle = parseWorkflowCode((await bundler.createBundle()).code);
   t.context.workflowCreator = REUSE_V8_CONTEXT
-    ? await TestReusableVMWorkflowCreator.create(workflowBundle, 200)
-    : await TestVMWorkflowCreator.create(workflowBundle, 200);
+    ? await TestReusableVMWorkflowCreator.create(workflowBundle, 200, new Set())
+    : await TestVMWorkflowCreator.create(workflowBundle, 200, new Set());
 });
 
 test.after.always(async (t) => {

packages/worker/src/runtime.ts

@@ -3,7 +3,8 @@ import * as v8 from 'node:v8';
 import * as fs from 'node:fs';
 import * as os from 'node:os';
 import { Heap } from 'heap-js';
-import { Subject, firstValueFrom } from 'rxjs';
+import { BehaviorSubject, lastValueFrom, of } from 'rxjs';
+import { concatMap, delay, map, repeat, takeWhile } from 'rxjs/operators';
 import * as native from '@temporalio/core-bridge';
 import {
   pollLogs,
@@ -134,7 +135,7 @@ export class Runtime {
   protected pendingCreations = 0;
   /** Track the registered native objects to automatically shutdown when all have been deregistered */
   protected readonly backRefs = new Set<TrackedNativeObject>();
-  protected readonly stopPollingForLogs = new Subject<void>();
+  protected readonly shouldPollForLogs = new BehaviorSubject<boolean>(false);
   protected readonly logPollPromise: Promise<void>;
   public readonly logger: Logger;
   protected readonly shutdownSignalCallbacks = new Set<() => void>();
@@ -265,30 +266,30 @@ export class Runtime {
   }
 
   protected async initLogPolling(logger: CoreLogger): Promise<void> {
+    this.shouldPollForLogs.next(true);
+
     if (!this.isForwardingLogs()) {
       return;
     }
-
-    const stopPollingForLogs = firstValueFrom(this.stopPollingForLogs);
-
     const poll = promisify(pollLogs);
     try {
-      for (;;) {
-        const logs = await poll(this.native);
-        for (const log of logs) {
-          logger.log(log.level, log.message, {
-            [LogTimestamp]: timeOfDayToBigint(log.timestamp),
-          });
-        }
-        logger.flush();
-        const stop = await Promise.race([
-          stopPollingForLogs.then(() => true),
-          new Promise<boolean>((resolve) => setTimeout(() => resolve(false), 3)),
-        ]);
-        if (stop) {
-          break;
-        }
-      }
+      await lastValueFrom(
+        of(this.shouldPollForLogs).pipe(
+          map((subject) => subject.getValue()),
+          takeWhile((shouldPoll) => shouldPoll),
+          concatMap(() => poll(this.native)),
+          map((logs) => {
+            for (const log of logs) {
+              logger.log(log.level, log.message, {
+                [LogTimestamp]: timeOfDayToBigint(log.timestamp),
+              });
+            }
+            logger.flush();
+          }),
+          delay(3), // Don't go wild polling as fast as possible
+          repeat()
+        )
+      );
     } catch (error) {
       // Log using the original logger instead of buffering
       this.options.logger.warn('Error gathering forwarded logs from core', { error });
@@ -473,7 +474,7 @@ export class Runtime {
   public async shutdown(): Promise<void> {
     delete Runtime._instance;
     this.teardownShutdownHook();
-    this.stopPollingForLogs.next();
+    this.shouldPollForLogs.next(false);
     // This will effectively drain all logs
     await this.logPollPromise;
     await promisify(runtimeShutdown)(this.native);

packages/worker/src/worker.ts

@@ -472,19 +472,33 @@ export class Worker {
     workflowBundle: WorkflowBundleWithSourceMapAndFilename,
     compiledOptions: CompiledWorkerOptions
   ): Promise<WorkflowCreator> {
+    const registeredActivityNames = new Set(
+      Object.entries(compiledOptions.activities ?? {})
+        .filter(([_, v]) => typeof v === 'function')
+        .map(([k]) => k)
+    );
     // This isn't required for vscode, only for Chrome Dev Tools which doesn't support debugging worker threads.
     // We also rely on this in debug-replayer where we inject a global variable to be read from workflow context.
     if (compiledOptions.debugMode) {
       if (compiledOptions.reuseV8Context) {
-        return await ReusableVMWorkflowCreator.create(workflowBundle, compiledOptions.isolateExecutionTimeoutMs);
+        return await ReusableVMWorkflowCreator.create(
+          workflowBundle,
+          compiledOptions.isolateExecutionTimeoutMs,
+          registeredActivityNames
+        );
       }
-      return await VMWorkflowCreator.create(workflowBundle, compiledOptions.isolateExecutionTimeoutMs);
+      return await VMWorkflowCreator.create(
+        workflowBundle,
+        compiledOptions.isolateExecutionTimeoutMs,
+        registeredActivityNames
+      );
     } else {
       return await ThreadedVMWorkflowCreator.create({
         workflowBundle,
         threadPoolSize: compiledOptions.workflowThreadPoolSize,
         isolateExecutionTimeoutMs: compiledOptions.isolateExecutionTimeoutMs,
         reuseV8Context: compiledOptions.reuseV8Context ?? false,
+        registeredActivityNames,
       });
     }
   }

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

@@ -55,8 +55,10 @@ export class ReusableVMWorkflowCreator implements WorkflowCreator {
 
   constructor(
     script: vm.Script,
-    public readonly workflowBundle: WorkflowBundleWithSourceMapAndFilename,
-    public readonly isolateExecutionTimeoutMs: number
+    protected readonly workflowBundle: WorkflowBundleWithSourceMapAndFilename,
+    protected readonly isolateExecutionTimeoutMs: number,
+    /** Known activity names registered on the executing worker */
+    protected readonly registeredActivityNames: Set<string>
   ) {
     if (!ReusableVMWorkflowCreator.unhandledRejectionHandlerHasBeenSet) {
       setUnhandledRejectionHandler((runId) => ReusableVMWorkflowCreator.workflowByRunId.get(runId));
@@ -166,6 +168,7 @@ export class ReusableVMWorkflowCreator implements WorkflowCreator {
       ...options,
       sourceMap: this.workflowBundle.sourceMap,
       getTimeOfDay: () => timeOfDayToBigint(getTimeOfDay()),
+      registeredActivityNames: this.registeredActivityNames,
     });
     const activator = bag.__TEMPORAL_ACTIVATOR__ as any;
 
@@ -182,12 +185,13 @@ export class ReusableVMWorkflowCreator implements WorkflowCreator {
   public static async create<T extends typeof ReusableVMWorkflowCreator>(
     this: T,
     workflowBundle: WorkflowBundleWithSourceMapAndFilename,
-    isolateExecutionTimeoutMs: number
+    isolateExecutionTimeoutMs: number,
+    registeredActivityNames: Set<string>
   ): Promise<InstanceType<T>> {
     globalHandlers.install(); // Call is idempotent
     await globalHandlers.addWorkflowBundle(workflowBundle);
     const script = new vm.Script(workflowBundle.code, { filename: workflowBundle.filename });
-    return new this(script, workflowBundle, isolateExecutionTimeoutMs) as InstanceType<T>;
+    return new this(script, workflowBundle, isolateExecutionTimeoutMs, registeredActivityNames) as InstanceType<T>;
   }
 
   /**

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

@@ -136,6 +136,7 @@ export interface ThreadedVMWorkflowCreatorOptions {
   threadPoolSize: number;
   isolateExecutionTimeoutMs: number;
   reuseV8Context: boolean;
+  registeredActivityNames: Set<string>;
 }
 
 /**
@@ -152,13 +153,20 @@ export class ThreadedVMWorkflowCreator implements WorkflowCreator {
     workflowBundle,
     isolateExecutionTimeoutMs,
     reuseV8Context,
+    registeredActivityNames,
   }: ThreadedVMWorkflowCreatorOptions): Promise<ThreadedVMWorkflowCreator> {
     const workerThreadClients = Array(threadPoolSize)
       .fill(0)
       .map(() => new WorkerThreadClient(new NodeWorker(require.resolve('./workflow-worker-thread'))));
     await Promise.all(
       workerThreadClients.map((client) =>
-        client.send({ type: 'init', workflowBundle, isolateExecutionTimeoutMs, reuseV8Context })
+        client.send({
+          type: 'init',
+          workflowBundle,
+          isolateExecutionTimeoutMs,
+          reuseV8Context,
+          registeredActivityNames,
+        })
       )
     );
     return new this(workerThreadClients);

packages/worker/src/workflow/vm.ts

@@ -25,8 +25,9 @@ export class VMWorkflowCreator implements WorkflowCreator {
 
   constructor(
     script: vm.Script,
-    public readonly workflowBundle: WorkflowBundleWithSourceMapAndFilename,
-    public readonly isolateExecutionTimeoutMs: number
+    protected readonly workflowBundle: WorkflowBundleWithSourceMapAndFilename,
+    protected readonly isolateExecutionTimeoutMs: number,
+    protected readonly registeredActivityNames: Set<string>
   ) {
     if (!VMWorkflowCreator.unhandledRejectionHandlerHasBeenSet) {
       setUnhandledRejectionHandler((runId) => VMWorkflowCreator.workflowByRunId.get(runId));
@@ -62,6 +63,7 @@ export class VMWorkflowCreator implements WorkflowCreator {
       ...options,
       sourceMap: this.workflowBundle.sourceMap,
       getTimeOfDay: () => timeOfDayToBigint(getTimeOfDay()),
+      registeredActivityNames: this.registeredActivityNames,
     });
     const activator = context.__TEMPORAL_ACTIVATOR__ as any;
 
@@ -97,12 +99,13 @@ export class VMWorkflowCreator implements WorkflowCreator {
   public static async create<T extends typeof VMWorkflowCreator>(
     this: T,
     workflowBundle: WorkflowBundleWithSourceMapAndFilename,
-    isolateExecutionTimeoutMs: number
+    isolateExecutionTimeoutMs: number,
+    registeredActivityNames: Set<string>
   ): Promise<InstanceType<T>> {
     globalHandlers.install();
     await globalHandlers.addWorkflowBundle(workflowBundle);
     const script = new vm.Script(workflowBundle.code, { filename: workflowBundle.filename });
-    return new this(script, workflowBundle, isolateExecutionTimeoutMs) as InstanceType<T>;
+    return new this(script, workflowBundle, isolateExecutionTimeoutMs, registeredActivityNames) as InstanceType<T>;
   }
 
   /**

packages/worker/src/workflow/workflow-worker-thread.ts

@@ -31,10 +31,18 @@ async function handleRequest({ requestId, input }: WorkerThreadRequest): Promise
   switch (input.type) {
     case 'init':
       if (input.reuseV8Context) {
-        workflowCreator = await ReusableVMWorkflowCreator.create(input.workflowBundle, input.isolateExecutionTimeoutMs);
+        workflowCreator = await ReusableVMWorkflowCreator.create(
+          input.workflowBundle,
+          input.isolateExecutionTimeoutMs,
+          input.registeredActivityNames
+        );
         workflowGetter = (runId) => ReusableVMWorkflowCreator.workflowByRunId.get(runId);
       } else {
-        workflowCreator = await VMWorkflowCreator.create(input.workflowBundle, input.isolateExecutionTimeoutMs);
+        workflowCreator = await VMWorkflowCreator.create(
+          input.workflowBundle,
+          input.isolateExecutionTimeoutMs,
+          input.registeredActivityNames
+        );
         workflowGetter = (runId) => VMWorkflowCreator.workflowByRunId.get(runId);
       }
       return ok(requestId);

packages/worker/src/workflow/workflow-worker-thread/input.ts

@@ -15,6 +15,7 @@ export interface Init {
   type: 'init';
   isolateExecutionTimeoutMs: number;
   workflowBundle: WorkflowBundleWithSourceMapAndFilename;
+  registeredActivityNames: Set<string>;
   reuseV8Context: boolean;
 }
 

packages/workflow/src/interfaces.ts

@@ -401,6 +401,7 @@ export interface WorkflowCreateOptions {
 
 export interface WorkflowCreateOptionsInternal extends WorkflowCreateOptions {
   sourceMap: RawSourceMap;
+  registeredActivityNames: Set<string>;
   getTimeOfDay(): bigint;
 }
 

packages/workflow/src/internals.ts

@@ -281,6 +281,8 @@ export class Activator implements ActivationHandler {
    */
   public readonly getTimeOfDay: () => bigint;
 
+  public readonly registeredActivityNames: Set<string>;
+
   constructor({
     info,
     now,
@@ -289,13 +291,15 @@ export class Activator implements ActivationHandler {
     getTimeOfDay,
     randomnessSeed,
     patches,
+    registeredActivityNames,
   }: WorkflowCreateOptionsInternal) {
     this.getTimeOfDay = getTimeOfDay;
     this.info = info;
     this.now = now;
     this.showStackTraceSources = showStackTraceSources;
     this.sourceMap = sourceMap;
     this.random = alea(randomnessSeed);
+    this.registeredActivityNames = registeredActivityNames;
 
     if (info.unsafe.isReplaying) {
       for (const patchId of patches) {

packages/workflow/src/workflow.ts

@@ -199,6 +199,11 @@ async function scheduleLocalActivityNextHandler({
   originalScheduleTime,
 }: LocalActivityInput): Promise<unknown> {
   const activator = getActivator();
+  // Eagerly fail the local activity (which will in turn fail the workflow task.
+  // Do not fail on replay where the local activities may not be registered on the replay worker.
+  if (!workflowInfo().unsafe.isReplaying && !activator.registeredActivityNames.has(activityType)) {
+    throw new ReferenceError(`Local activity of type '${activityType}' not registered on worker`);
+  }
   validateLocalActivityOptions(options);
 
   return new Promise((resolve, reject) => {