add skipClientWorkerSetCheck to worker options

Author: chris-olszewski

packages/core-bridge/src/worker.rs

@@ -507,6 +507,7 @@ 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)]
@@ -574,6 +575,7 @@ 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/ts/native.ts

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

packages/test/src/test-bridge.ts

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

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

@@ -463,6 +463,7 @@ test('Worker requests Eager Activity Dispatch if possible', async (t) => {
     activities: {
       testActivity: () => 'workflow-and-activity-worker',
     },
+    skipClientWorkerSetCheck: true,
   });
   const handle = await startWorkflow(executeEagerActivity);
   await activityWorker.runUntil(workflowWorker.runUntil(handle.result()));
@@ -499,6 +500,7 @@ test("Worker doesn't request Eager Activity Dispatch if no activities are regist
   });
   const workflowWorker = await createWorker({
     activities: {},
+    skipClientWorkerSetCheck: true,
   });
   const handle = await startWorkflow(dontExecuteEagerActivity);
   const result = await activityWorker.runUntil(workflowWorker.runUntil(handle.result()));

packages/test/src/test-runtime.ts

@@ -87,6 +87,7 @@ if (RUN_INTEGRATION_TESTS) {
       const worker = await Worker.create({
         ...defaultOptions,
         taskQueue: 'q1',
+        skipClientWorkerSetCheck: true,
       });
       await worker.runUntil(() =>
         asyncRetry(

packages/test/src/test-sinks.ts

@@ -261,6 +261,7 @@ 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() }));
@@ -307,6 +308,7 @@ if (RUN_INTEGRATION_TESTS) {
       ...defaultOptions,
       taskQueue,
       sinks,
+      skipClientWorkerSetCheck: true,
     });
     const workflowId = uuid4();
     await worker.runUntil(client.execute(workflows.logSinkTester, { taskQueue, workflowId }));
@@ -320,6 +322,7 @@ if (RUN_INTEGRATION_TESTS) {
       {
         ...defaultOptions,
         sinks,
+        skipClientWorkerSetCheck: true,
       },
       history,
       workflowId
@@ -351,6 +354,7 @@ if (RUN_INTEGRATION_TESTS) {
       ...defaultOptions,
       taskQueue,
       sinks,
+      skipClientWorkerSetCheck: true,
     });
     const client = new WorkflowClient();
     const workflowId = uuid4();
@@ -411,6 +415,7 @@ if (RUN_INTEGRATION_TESTS) {
       ...defaultOptions,
       taskQueue,
       sinks,
+      skipClientWorkerSetCheck: true,
     });
     await worker.runUntil(
       client.execute(workflows.upsertAndReadSearchAttributes, {

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

@@ -43,6 +43,7 @@ test('Worker deployment based versioning', async (t) => {
       defaultVersioningBehavior: 'PINNED',
     },
     connection: nativeConnection,
+    skipClientWorkerSetCheck: true,
   });
   const worker1Promise = worker1.run();
   worker1Promise.catch((err) => {
@@ -58,6 +59,7 @@ test('Worker deployment based versioning', async (t) => {
       defaultVersioningBehavior: 'PINNED',
     },
     connection: nativeConnection,
+    skipClientWorkerSetCheck: true,
   });
   const worker2Promise = worker2.run();
   worker2Promise.catch((err) => {
@@ -73,6 +75,7 @@ test('Worker deployment based versioning', async (t) => {
       defaultVersioningBehavior: 'PINNED',
     },
     connection: nativeConnection,
+    skipClientWorkerSetCheck: true,
   });
   const worker3Promise = worker3.run();
   worker3Promise.catch((err) => {
@@ -160,6 +163,7 @@ test('Worker deployment based versioning with ramping', async (t) => {
       defaultVersioningBehavior: 'PINNED',
     },
     connection: nativeConnection,
+    skipClientWorkerSetCheck: true,
   });
   const worker1Promise = worker1.run();
   worker1Promise.catch((err) => {
@@ -175,6 +179,7 @@ test('Worker deployment based versioning with ramping', async (t) => {
       defaultVersioningBehavior: 'PINNED',
     },
     connection: nativeConnection,
+    skipClientWorkerSetCheck: true,
   });
   const worker2Promise = worker2.run();
   worker2Promise.catch((err) => {

packages/worker/src/worker-options.ts

@@ -464,6 +464,11 @@ export interface WorkerOptions {
    */
   defaultHeartbeatThrottleInterval?: Duration;
 
+  /**
+   * TODO: write
+   */
+  skipClientWorkerSetCheck?: boolean;
+
   /**
    * A mapping of interceptor type to a list of factories or module paths.
    *
@@ -817,6 +822,7 @@ export type WorkerOptionsWithDefaults = WorkerOptions &
       | 'debugMode'
       | 'reuseV8Context'
       | 'tuner'
+      | 'skipClientWorkerSetCheck'
     >
   > & {
     interceptors: Required<WorkerInterceptors>;
@@ -968,6 +974,7 @@ function addDefaultWorkerOptions(
     stickyQueueScheduleToStartTimeout: '10s',
     maxHeartbeatThrottleInterval: '60s',
     defaultHeartbeatThrottleInterval: '30s',
+    skipClientWorkerSetCheck: false,
     // 4294967295ms is the maximum allowed time
     isolateExecutionTimeout: debugMode ? '4294967295ms' : '5s',
     workflowThreadPoolSize: reuseV8Context ? 1 : 2,
@@ -1084,6 +1091,7 @@ export function toNativeWorkerOptions(opts: CompiledWorkerOptionsWithBuildId): n
     maxTaskQueueActivitiesPerSecond: opts.maxTaskQueueActivitiesPerSecond ?? null,
     maxActivitiesPerSecond: opts.maxActivitiesPerSecond ?? null,
     shutdownGraceTime: msToNumber(opts.shutdownGraceTime),
+    skipClientWorkerSetCheck: opts.skipClientWorkerSetCheck,
   };
 }