fix(worker): Disable Eager Activities if Worker has no activities (#1326)

## What changed and why

- Force `WorkerOptions.enableNonLocalActivities` to false if no
activities are registered on a Worker, which automatically implies that
Core will not request Eager Activity Dispatch for activities scheduled
by workflows running on that Worker.

Until now, Core would request Eager Activity Dispatch for activities
scheduled on the same task queue as the Workflow, even when the Worker
was being run without any activities registered (splitting Workflow and
Activity execution to distinct Workers is a diagnostic strategy we often
suggest). In those cases, the activity execution would time out on the
first attempt, adding delay before the activity finally get executed.

- Enable several server features when running dev server for internal
integration tests (applicable to tests using the new test helper).

- Fixed an issue in bridge that was preventing passing `extraArgs` to
dev server.

- In `test-npm-init` CI tests, copy the current NPM version specifier
from the SDK repo to the created sample repo. This is required in case
packages get published to Verdaccio using a pre-release specifier (e.g.
`1.9.0-rc.0`) as NPM will not consider a pre-release version to be an
acceptable substitute for syntaxes such as `^1.8.0`, thus causing
failure to install dependencies during sample testing.

Author: mjameswh

packages/core-bridge/src/conversions.rs

@@ -37,8 +37,8 @@ impl ArrayHandleConversionsExt for Handle<'_, JsArray> {
         let len = js_vec.len();
         let mut ret_vec = Vec::<String>::with_capacity(len);
 
-        for i in 0..len - 1 {
-            ret_vec[i] = js_vec[i].downcast_or_throw::<JsString, _>(cx)?.value(cx);
+        for i in 0..len {
+            ret_vec.push(js_vec[i].downcast_or_throw::<JsString, _>(cx)?.value(cx));
         }
         Ok(ret_vec)
     }

packages/test/src/helpers-integration.ts

@@ -28,6 +28,18 @@ export interface Context {
   workflowBundle: WorkflowBundle;
 }
 
+const defaultDynamicConfigOptions = [
+  'frontend.enableUpdateWorkflowExecution=true',
+  'frontend.enableUpdateWorkflowExecutionAsyncAccepted=true',
+  'frontend.workerVersioningDataAPIs=true',
+  'frontend.workerVersioningWorkflowAPIs=true',
+  'system.enableActivityEagerExecution=true',
+  'system.enableEagerWorkflowStart=true',
+  'system.forceSearchAttributesCacheRefreshOnRead=true',
+  'worker.buildIdScavengerEnabled=true',
+  'worker.removableBuildIdDurationSinceDefault=1',
+];
+
 export function makeTestFunction(opts: {
   workflowsPath: string;
   workflowEnvironmentOpts?: LocalTestWorkflowEnvironmentOptions;
@@ -37,7 +49,16 @@ export function makeTestFunction(opts: {
   test.before(async (t) => {
     // Ignore invalid log levels
     Runtime.install({ logger: new DefaultLogger((process.env.TEST_LOG_LEVEL || 'DEBUG').toUpperCase() as LogLevel) });
-    const env = await TestWorkflowEnvironment.createLocal(opts.workflowEnvironmentOpts);
+    const env = await TestWorkflowEnvironment.createLocal({
+      ...opts.workflowEnvironmentOpts,
+      server: {
+        ...opts.workflowEnvironmentOpts?.server,
+        extraArgs: [
+          ...defaultDynamicConfigOptions.flatMap((opt) => ['--dynamic-config-value', opt]),
+          ...(opts.workflowEnvironmentOpts?.server?.extraArgs ?? []),
+        ],
+      },
+    });
     const workflowBundle = await bundleWorkflowCode({
       ...bundlerOptions,
       workflowInterceptorModules: [...defaultWorkflowInterceptorModules, ...(opts.workflowInterceptorModules ?? [])],

packages/test/src/test-integration-workflows.ts

@@ -222,3 +222,86 @@ test('Start of workflow with signal is delayed', async (t) => {
   const startDelay = workflowExecutionStartedEvent?.workflowExecutionStartedEventAttributes?.firstWorkflowTaskBackoff;
   t.is(tsToMs(startDelay), 4678000);
 });
+
+export async function executeEagerActivity(): Promise<void> {
+  const scheduleActivity = () =>
+    workflow
+      .proxyActivities({ scheduleToCloseTimeout: '5s', allowEagerDispatch: true })
+      .testActivity()
+      .then((res) => {
+        if (res !== 'workflow-and-activity-worker')
+          throw workflow.ApplicationFailure.nonRetryable('Activity was not eagerly dispatched');
+      });
+
+  for (let i = 0; i < 10; i++) {
+    // Schedule 3 activities at a time (`MAX_EAGER_ACTIVITY_RESERVATIONS_PER_WORKFLOW_TASK`)
+    await Promise.all([scheduleActivity(), scheduleActivity(), scheduleActivity()]);
+  }
+}
+
+test('Worker requests Eager Activity Dispatch if possible', async (t) => {
+  const { createWorker, startWorkflow } = helpers(t);
+
+  // If eager activity dispatch is working, then the task will always be dispatched to the workflow
+  // worker. Otherwise, chances are 50%-50% for either workers. The test workflow schedule the
+  // activity 30 times to make sure that the workflow worker is really getting the task thanks to
+  // eager activity dispatch, and not out of pure luck.
+
+  const activityWorker = await createWorker({
+    activities: {
+      testActivity: () => 'activity-only-worker',
+    },
+    // Override the default workflow bundle, to make this an activity-only worker
+    workflowBundle: undefined,
+  });
+  const workflowWorker = await createWorker({
+    activities: {
+      testActivity: () => 'workflow-and-activity-worker',
+    },
+  });
+  const handle = await startWorkflow(executeEagerActivity);
+  await activityWorker.runUntil(workflowWorker.runUntil(handle.result()));
+  const { events } = await handle.fetchHistory();
+
+  t.false(events?.some?.((ev) => ev.activityTaskTimedOutEventAttributes));
+  const activityTaskStarted = events?.filter?.((ev) => ev.activityTaskStartedEventAttributes);
+  t.is(activityTaskStarted?.length, 30);
+  t.true(activityTaskStarted?.every((ev) => ev.activityTaskStartedEventAttributes?.attempt === 1));
+});
+
+export async function dontExecuteEagerActivity(): Promise<string> {
+  return (await workflow
+    .proxyActivities({ scheduleToCloseTimeout: '5s', allowEagerDispatch: true })
+    .testActivity()
+    .catch(() => 'failed')) as string;
+}
+
+test("Worker doesn't request Eager Activity Dispatch if no activities are registered", async (t) => {
+  const { createWorker, startWorkflow } = helpers(t);
+
+  // If the activity was eagerly dispatched to the Workflow worker even though it is a Workflow-only
+  // worker, then the activity execution will timeout (because tasks are not being polled) or
+  // otherwise fail (because no activity is registered under that name). Therefore, if the history
+  // shows only one attempt for that activity and no timeout, that can only mean that the activity
+  // was not eagerly dispatched.
+
+  const activityWorker = await createWorker({
+    activities: {
+      testActivity: () => 'success',
+    },
+    // Override the default workflow bundle, to make this an activity-only worker
+    workflowBundle: undefined,
+  });
+  const workflowWorker = await createWorker({
+    activities: {},
+  });
+  const handle = await startWorkflow(dontExecuteEagerActivity);
+  const result = await activityWorker.runUntil(workflowWorker.runUntil(handle.result()));
+  const { events } = await handle.fetchHistory();
+
+  t.is(result, 'success');
+  t.false(events?.some?.((ev) => ev.activityTaskTimedOutEventAttributes));
+  const activityTaskStarted = events?.filter?.((ev) => ev.activityTaskStartedEventAttributes);
+  t.is(activityTaskStarted?.length, 1);
+  t.is(activityTaskStarted?.[0]?.activityTaskStartedEventAttributes?.attempt, 1);
+});

packages/test/src/test-worker-versioning.ts

@@ -176,7 +176,7 @@ if (RUN_INTEGRATION_TESTS) {
     await asyncRetry(
       async () => {
         const reachResp = await conn.taskQueue.getReachability({ buildIds: ['2.0', '1.0', '1.1'] });
-        assert.deepEqual(reachResp.buildIdReachability['2.0']?.taskQueueReachability[taskQueue], ['NewWorkflows']);
+        assert.deepEqual(reachResp.buildIdReachability['2.0']?.taskQueueReachability[taskQueue], ['NEW_WORKFLOWS']);
         assert.deepEqual(reachResp.buildIdReachability['1.1']?.taskQueueReachability[taskQueue], []);
         assert.deepEqual(reachResp.buildIdReachability['1.0']?.taskQueueReachability[taskQueue], []);
       },
@@ -188,7 +188,9 @@ if (RUN_INTEGRATION_TESTS) {
           buildIds: [UnversionedBuildId],
           taskQueues: [taskQueue],
         });
-        assert.deepEqual(reachResp.buildIdReachability[UnversionedBuildId]?.taskQueueReachability[taskQueue], []);
+        assert.deepEqual(reachResp.buildIdReachability[UnversionedBuildId]?.taskQueueReachability[taskQueue], [
+          'NEW_WORKFLOWS',
+        ]);
       },
       { maxTimeout: 1000 }
     );

packages/worker/src/worker-options.ts

@@ -1,7 +1,7 @@
 import * as os from 'node:os';
 import * as v8 from 'node:v8';
 import type { Configuration as WebpackConfiguration } from 'webpack';
-import { DataConverter, LoadedDataConverter } from '@temporalio/common';
+import { ActivityFunction, DataConverter, LoadedDataConverter } from '@temporalio/common';
 import { Duration, msOptionalToNumber, msToNumber } from '@temporalio/common/lib/time';
 import { loadDataConverter } from '@temporalio/common/lib/internal-non-workflow';
 import { LoggerSinks } from '@temporalio/workflow';
@@ -200,6 +200,7 @@ export interface WorkerOptions {
    * Whether or not to poll on the Activity task queue.
    *
    * If disabled and activities are registered on the Worker, it will run only local Activities.
+   * This setting is ignored if no activity is registed on the Worker.
    *
    * @default true
    */
@@ -559,7 +560,8 @@ export type WorkerOptionsWithDefaults = WorkerOptions &
  * {@link WorkerOptions} where the attributes the Worker requires are required and time units are converted from ms
  * formatted strings to numbers.
  */
-export interface CompiledWorkerOptions extends Omit<WorkerOptionsWithDefaults, 'serverOptions' | 'interceptors'> {
+export interface CompiledWorkerOptions
+  extends Omit<WorkerOptionsWithDefaults, 'serverOptions' | 'interceptors' | 'activities'> {
   interceptors: CompiledWorkerInterceptors;
   shutdownGraceTimeMs: number;
   shutdownForceTimeMs?: number;
@@ -568,6 +570,7 @@ export interface CompiledWorkerOptions extends Omit<WorkerOptionsWithDefaults, '
   maxHeartbeatThrottleIntervalMs: number;
   defaultHeartbeatThrottleIntervalMs: number;
   loadedDataConverter: LoadedDataConverter;
+  activities: Map<string, ActivityFunction>;
 }
 
 /**
@@ -752,6 +755,8 @@ export function compileWorkerOptions(opts: WorkerOptionsWithDefaults): CompiledW
     opts.maxConcurrentWorkflowTaskExecutions = 2;
   }
 
+  const activities = new Map(Object.entries(opts.activities ?? {}).filter(([_, v]) => typeof v === 'function'));
+
   return {
     ...opts,
     interceptors: compileWorkerInterceptors(opts.interceptors),
@@ -762,5 +767,7 @@ export function compileWorkerOptions(opts: WorkerOptionsWithDefaults): CompiledW
     maxHeartbeatThrottleIntervalMs: msToNumber(opts.maxHeartbeatThrottleInterval),
     defaultHeartbeatThrottleIntervalMs: msToNumber(opts.defaultHeartbeatThrottleInterval),
     loadedDataConverter: loadDataConverter(opts.dataConverter),
+    activities,
+    enableNonLocalActivities: opts.enableNonLocalActivities && activities.size > 0,
   };
 }

packages/worker/src/worker.ts

@@ -464,11 +464,7 @@ 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)
-    );
+    const registeredActivityNames = new Set(compiledOptions.activities.keys());
     // 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) {
@@ -858,16 +854,15 @@ export class Worker {
                   );
 
                   const { activityType } = info;
-                  // activities is of type "object" which does not support string indexes
-                  const fn = (this.options.activities as any)?.[activityType];
+                  const fn = this.options.activities.get(activityType);
                   if (typeof fn !== 'function') {
                     output = {
                       type: 'result',
                       result: {
                         failed: {
                           failure: {
                             message: `Activity function ${activityType} is not registered on this Worker, available activities: ${JSON.stringify(
-                              Object.keys(this.options.activities ?? {})
+                              [...this.options.activities.keys()]
                             )}`,
                             applicationFailureInfo: {
                               type: 'NotFoundError',
@@ -1158,6 +1153,7 @@ export class Worker {
     activation: coresdk.workflow_activation.WorkflowActivation,
     startWorkflow: coresdk.workflow_activation.IStartWorkflow
   ): Promise<WorkflowWithLogAttributes> {
+    // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
     const workflowCreator = this.workflowCreator!;
     if (
       !(
@@ -1531,7 +1527,7 @@ export class Worker {
 
   protected activity$(): Observable<void> {
     // This Worker did not register any activities, return early
-    if (this.options.activities === undefined || Object.keys(this.options.activities).length === 0) {
+    if (!this.options.activities?.size) {
       if (!this.isReplayWorker) this.log.warn('No activities registered, not polling for activity tasks');
       this.activityPollerStateSubject.next('SHUTDOWN');
       return EMPTY;

scripts/init-from-verdaccio.js

@@ -5,6 +5,10 @@ const { spawnNpx } = require('./utils');
 
 async function main() {
   const { registryDir, targetDir, initArgs } = await getArgs();
+  // Force samples to use the same version of @temporalio/* packages as the one
+  // we are testing. This is required when testing against a pre-release version,
+  // which would not be otherwise matched by specifier like ^1.8.0.
+  const { version } = require('../lerna.json');
 
   await withRegistry(registryDir, async () => {
     console.log('spawning npx @temporalio/create with args:', initArgs);
@@ -14,7 +18,7 @@ async function main() {
       writeFileSync(npmConfigFile, npmConfig, { encoding: 'utf-8' });
 
       await spawnNpx(
-        ['@temporalio/create', targetDir, '--no-git-init', '--temporalio-version', 'latest', ...initArgs],
+        [`@temporalio/create@${version}`, targetDir, '--no-git-init', '--sdk-version', version, ...initArgs],
         {
           stdio: 'inherit',
           stdout: 'inherit',