remove public config setting

Author: chris-olszewski

packages/core-bridge/sdk-core

@@ -507,7 +507,6 @@ mod config {
         max_activities_per_second: Option<f64>,
         max_task_queue_activities_per_second: Option<f64>,
         shutdown_grace_time: Option<Duration>,
-        skip_client_worker_set_check: bool,
     }
 
     #[derive(TryFromJs)]
@@ -575,7 +574,6 @@ mod config {
                 .max_task_queue_activities_per_second(self.max_task_queue_activities_per_second)
                 .max_worker_activities_per_second(self.max_activities_per_second)
                 .graceful_shutdown_period(self.shutdown_grace_time)
-                .skip_client_worker_set_check(self.skip_client_worker_set_check)
                 .build()
         }
     }

packages/core-bridge/src/worker.rs

@@ -221,7 +221,6 @@ export interface WorkerOptions {
   maxTaskQueueActivitiesPerSecond: Option<number>;
   maxActivitiesPerSecond: Option<number>;
   shutdownGraceTime: number;
-  skipClientWorkerSetCheck: boolean;
 }
 
 export type PollerBehavior =

packages/core-bridge/ts/native.ts

@@ -17,6 +17,8 @@ import {
   DefaultLogger,
   LogEntry,
   LogLevel,
+  NativeConnection,
+  NativeConnectionOptions,
   ReplayWorkerOptions,
   Runtime,
   RuntimeOptions,
@@ -184,6 +186,7 @@ export async function createTestWorkflowEnvironment(
 export interface Helpers {
   taskQueue: string;
   createWorker(opts?: Partial<WorkerOptions>): Promise<Worker>;
+  createNativeConnection(opts?: Partial<NativeConnectionOptions>): Promise<NativeConnection>;
   runReplayHistory(opts: Partial<ReplayWorkerOptions>, history: temporal.api.history.v1.IHistory): Promise<void>;
   executeWorkflow<T extends () => Promise<any>>(workflowType: T): Promise<workflow.WorkflowResultType<T>>;
   executeWorkflow<T extends workflow.Workflow>(
@@ -218,6 +221,9 @@ export function configurableHelpers<T>(
         ...opts,
       });
     },
+    async createNativeConnection(opts?: Partial<NativeConnectionOptions>): Promise<NativeConnection> {
+      return await NativeConnection.connect({ address: testEnv.address, ...opts });
+    },
     async runReplayHistory(
       opts: Partial<ReplayWorkerOptions>,
       history: temporal.api.history.v1.IHistory

packages/test/src/helpers-integration.ts

@@ -306,7 +306,6 @@ const GenericConfigs = {
       maxTaskQueueActivitiesPerSecond: null,
       maxActivitiesPerSecond: null,
       shutdownGraceTime: 1000,
-      skipClientWorkerSetCheck: true,
     } satisfies native.WorkerOptions,
   },
   ephemeralServer: {

packages/test/src/test-bridge.ts

@@ -445,7 +445,7 @@ export async function executeEagerActivity(): Promise<void> {
 }
 
 test('Worker requests Eager Activity Dispatch if possible', async (t) => {
-  const { createWorker, startWorkflow } = helpers(t);
+  const { createWorker, startWorkflow, createNativeConnection } = 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
@@ -459,11 +459,15 @@ test('Worker requests Eager Activity Dispatch if possible', async (t) => {
     // Override the default workflow bundle, to make this an activity-only worker
     workflowBundle: undefined,
   });
+  const workflowWorkerConnection = await createNativeConnection();
+  t.teardown(() => {
+    workflowWorkerConnection.close();
+  });
   const workflowWorker = await createWorker({
+    connection: workflowWorkerConnection,
     activities: {
       testActivity: () => 'workflow-and-activity-worker',
     },
-    skipClientWorkerSetCheck: true,
   });
   const handle = await startWorkflow(executeEagerActivity);
   await activityWorker.runUntil(workflowWorker.runUntil(handle.result()));
@@ -483,7 +487,7 @@ export async function dontExecuteEagerActivity(): Promise<string> {
 }
 
 test("Worker doesn't request Eager Activity Dispatch if no activities are registered", async (t) => {
-  const { createWorker, startWorkflow } = helpers(t);
+  const { createNativeConnection, 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
@@ -498,9 +502,11 @@ test("Worker doesn't request Eager Activity Dispatch if no activities are regist
     // Override the default workflow bundle, to make this an activity-only worker
     workflowBundle: undefined,
   });
+  const workflowWorkerConnection = await createNativeConnection();
+  t.teardown(() => workflowWorkerConnection.close());
   const workflowWorker = await createWorker({
+    connection: workflowWorkerConnection,
     activities: {},
-    skipClientWorkerSetCheck: true,
   });
   const handle = await startWorkflow(dontExecuteEagerActivity);
   const result = await activityWorker.runUntil(workflowWorker.runUntil(handle.result()));

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

@@ -261,7 +261,6 @@ if (RUN_INTEGRATION_TESTS) {
       sinks,
       maxCachedWorkflows: 0,
       maxConcurrentWorkflowTaskExecutions: 2,
-      skipClientWorkerSetCheck: true,
     });
     const client = new WorkflowClient();
     await worker.runUntil(client.execute(workflows.logSinkTester, { taskQueue, workflowId: uuid4() }));
@@ -308,7 +307,6 @@ if (RUN_INTEGRATION_TESTS) {
       ...defaultOptions,
       taskQueue,
       sinks,
-      skipClientWorkerSetCheck: true,
     });
     const workflowId = uuid4();
     await worker.runUntil(client.execute(workflows.logSinkTester, { taskQueue, workflowId }));
@@ -322,7 +320,6 @@ if (RUN_INTEGRATION_TESTS) {
       {
         ...defaultOptions,
         sinks,
-        skipClientWorkerSetCheck: true,
       },
       history,
       workflowId
@@ -354,7 +351,6 @@ if (RUN_INTEGRATION_TESTS) {
       ...defaultOptions,
       taskQueue,
       sinks,
-      skipClientWorkerSetCheck: true,
     });
     const client = new WorkflowClient();
     const workflowId = uuid4();
@@ -371,7 +367,6 @@ if (RUN_INTEGRATION_TESTS) {
       {
         ...defaultOptions,
         sinks,
-        skipClientWorkerSetCheck: true,
       },
       history,
       workflowId
@@ -416,7 +411,6 @@ if (RUN_INTEGRATION_TESTS) {
       ...defaultOptions,
       taskQueue,
       sinks,
-      skipClientWorkerSetCheck: true,
     });
     await worker.runUntil(
       client.execute(workflows.upsertAndReadSearchAttributes, {

packages/test/src/test-sinks.ts

@@ -11,7 +11,7 @@ import { Client } from '@temporalio/client';
 import { toCanonicalString, WorkerDeploymentVersion } from '@temporalio/common';
 import { temporal } from '@temporalio/proto';
 import { Worker } from './helpers';
-import { Context, makeTestFunction } from './helpers-integration';
+import { Context, helpers, makeTestFunction } from './helpers-integration';
 import { unblockSignal, versionQuery } from './workflows';
 
 const test = makeTestFunction({ workflowsPath: __filename });
@@ -20,6 +20,7 @@ test('Worker deployment based versioning', async (t) => {
   const taskQueue = 'worker-deployment-based-versioning-' + randomUUID();
   const deploymentName = 'deployment-' + randomUUID();
   const { client, nativeConnection } = t.context.env;
+  const { createNativeConnection } = helpers(t);
 
   const w1DeploymentVersion = {
     buildId: '1.0',
@@ -43,13 +44,14 @@ test('Worker deployment based versioning', async (t) => {
       defaultVersioningBehavior: 'PINNED',
     },
     connection: nativeConnection,
-    skipClientWorkerSetCheck: true,
   });
   const worker1Promise = worker1.run();
   worker1Promise.catch((err) => {
     t.fail('Worker 1.0 run error: ' + err);
   });
 
+  const worker2Connection = await createNativeConnection();
+  t.teardown(() => worker2Connection.close());
   const worker2 = await Worker.create({
     workflowsPath: require.resolve('./deployment-versioning-v2'),
     taskQueue,
@@ -58,14 +60,15 @@ test('Worker deployment based versioning', async (t) => {
       version: w2DeploymentVersion,
       defaultVersioningBehavior: 'PINNED',
     },
-    connection: nativeConnection,
-    skipClientWorkerSetCheck: true,
+    connection: worker2Connection,
   });
   const worker2Promise = worker2.run();
   worker2Promise.catch((err) => {
     t.fail('Worker 2.0 run error: ' + err);
   });
 
+  const worker3Connection = await createNativeConnection();
+  t.teardown(() => worker3Connection.close());
   const worker3 = await Worker.create({
     workflowsPath: require.resolve('./deployment-versioning-v3'),
     taskQueue,
@@ -74,8 +77,7 @@ test('Worker deployment based versioning', async (t) => {
       version: w3DeploymentVersion,
       defaultVersioningBehavior: 'PINNED',
     },
-    connection: nativeConnection,
-    skipClientWorkerSetCheck: true,
+    connection: worker3Connection,
   });
   const worker3Promise = worker3.run();
   worker3Promise.catch((err) => {
@@ -144,6 +146,7 @@ test('Worker deployment based versioning with ramping', async (t) => {
   const taskQueue = 'worker-deployment-based-ramping-' + randomUUID();
   const deploymentName = 'deployment-ramping-' + randomUUID();
   const { client, nativeConnection } = t.context.env;
+  const { createNativeConnection } = helpers(t);
 
   const v1 = {
     buildId: '1.0',
@@ -163,13 +166,14 @@ test('Worker deployment based versioning with ramping', async (t) => {
       defaultVersioningBehavior: 'PINNED',
     },
     connection: nativeConnection,
-    skipClientWorkerSetCheck: true,
   });
   const worker1Promise = worker1.run();
   worker1Promise.catch((err) => {
     t.fail('Worker 1.0 run error: ' + err);
   });
 
+  const worker2Connection = await createNativeConnection();
+  t.teardown(() => worker2Connection.close());
   const worker2 = await Worker.create({
     workflowsPath: require.resolve('./deployment-versioning-v2'),
     taskQueue,
@@ -178,8 +182,7 @@ test('Worker deployment based versioning with ramping', async (t) => {
       version: v2,
       defaultVersioningBehavior: 'PINNED',
     },
-    connection: nativeConnection,
-    skipClientWorkerSetCheck: true,
+    connection: worker2Connection,
   });
   const worker2Promise = worker2.run();
   worker2Promise.catch((err) => {

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

@@ -90,7 +90,8 @@ export class TestWorkflowEnvironment {
     protected readonly server: native.EphemeralServer | 'existing',
     connection: Connection,
     nativeConnection: NativeConnection,
-    namespace: string | undefined
+    namespace: string | undefined,
+    public readonly address: string
   ) {
     this.connection = connection;
     this.nativeConnection = nativeConnection;
@@ -238,7 +239,16 @@ export class TestWorkflowEnvironment {
       [InternalConnectionOptionsSymbol]: { supportsTestService: supportsTimeSkipping },
     });
 
-    return new this(runtime, optsWithDefaults, supportsTimeSkipping, server, connection, nativeConnection, namespace);
+    return new this(
+      runtime,
+      optsWithDefaults,
+      supportsTimeSkipping,
+      server,
+      connection,
+      nativeConnection,
+      namespace,
+      address
+    );
   }
 
   /**

packages/testing/src/testing-workflow-environment.ts

@@ -464,14 +464,6 @@ export interface WorkerOptions {
    */
   defaultHeartbeatThrottleInterval?: Duration;
 
-  /**
-   * Skip the runtime validation that ensures the client is registered with the worker set.
-   * This should only be used in tests.
-   *
-   * @default false
-   */
-  skipClientWorkerSetCheck?: boolean;
-
   /**
    * A mapping of interceptor type to a list of factories or module paths.
    *
@@ -825,7 +817,6 @@ export type WorkerOptionsWithDefaults = WorkerOptions &
       | 'debugMode'
       | 'reuseV8Context'
       | 'tuner'
-      | 'skipClientWorkerSetCheck'
     >
   > & {
     interceptors: Required<WorkerInterceptors>;
@@ -977,7 +968,6 @@ function addDefaultWorkerOptions(
     stickyQueueScheduleToStartTimeout: '10s',
     maxHeartbeatThrottleInterval: '60s',
     defaultHeartbeatThrottleInterval: '30s',
-    skipClientWorkerSetCheck: false,
     // 4294967295ms is the maximum allowed time
     isolateExecutionTimeout: debugMode ? '4294967295ms' : '5s',
     workflowThreadPoolSize: reuseV8Context ? 1 : 2,
@@ -1094,7 +1084,6 @@ export function toNativeWorkerOptions(opts: CompiledWorkerOptionsWithBuildId): n
     maxTaskQueueActivitiesPerSecond: opts.maxTaskQueueActivitiesPerSecond ?? null,
     maxActivitiesPerSecond: opts.maxActivitiesPerSecond ?? null,
     shutdownGraceTime: msToNumber(opts.shutdownGraceTime),
-    skipClientWorkerSetCheck: opts.skipClientWorkerSetCheck,
   };
 }