Dynamic workflow support

Author: Sushisource

packages/common/src/workflow-options.ts

@@ -252,5 +252,6 @@ export function extractWorkflowType<T extends Workflow>(
 }
 
 export function isWorkflowFunctionWithOptions(obj: any): obj is WorkflowFunctionWithOptions<any[], any> {
+  if (obj === undefined || obj === null) return false;
   return obj.__temporal_is_workflow_function_with_options === true;
 }

packages/test/src/deployment-versioning-v1/index.ts

@@ -9,3 +9,9 @@ export async function deploymentVersioning(): Promise<string> {
   await condition(() => doFinish);
   return 'version-v1';
 }
+
+// Dynamic/default workflow handler
+export default defineWorkflowWithOptions({ versioningBehavior: 'pinned' }, _default);
+async function _default(): Promise<string> {
+  return 'dynamic';
+}

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

@@ -12,6 +12,7 @@ import * as activities from './activities';
 import { toCanonicalString, WorkerDeploymentVersion } from '@temporalio/common';
 import { makeTestFunction } from './helpers-integration';
 import { unblockSignal, versionQuery } from './workflows/';
+import { temporal } from '@temporalio/proto';
 
 const test = makeTestFunction({ workflowsPath: __filename });
 
@@ -40,6 +41,7 @@ test('Worker deployment based versioning', async (t) => {
     workerDeploymentOptions: {
       useWorkerVersioning: true,
       version: w1DeploymentVersion,
+      defaultVersioningBehavior: 'pinned',
     },
   });
   const worker1Promise = worker1.run();
@@ -54,6 +56,7 @@ test('Worker deployment based versioning', async (t) => {
     workerDeploymentOptions: {
       useWorkerVersioning: true,
       version: w2DeploymentVersion,
+      defaultVersioningBehavior: 'pinned',
     },
   });
   const worker2Promise = worker2.run();
@@ -68,6 +71,7 @@ test('Worker deployment based versioning', async (t) => {
     workerDeploymentOptions: {
       useWorkerVersioning: true,
       version: w3DeploymentVersion,
+      defaultVersioningBehavior: 'pinned',
     },
   });
   const worker3Promise = worker3.run();
@@ -154,6 +158,7 @@ test('Worker deployment based versioning with ramping', async (t) => {
     workerDeploymentOptions: {
       useWorkerVersioning: true,
       version: v1,
+      defaultVersioningBehavior: 'pinned',
     },
   });
   const worker1Promise = worker1.run();
@@ -168,6 +173,7 @@ test('Worker deployment based versioning with ramping', async (t) => {
     workerDeploymentOptions: {
       useWorkerVersioning: true,
       version: v2,
+      defaultVersioningBehavior: 'pinned',
     },
   });
   const worker2Promise = worker2.run();
@@ -237,6 +243,116 @@ test('Worker deployment based versioning with ramping', async (t) => {
   t.pass();
 });
 
+test('Worker deployment with dynamic workflow on run', async (t) => {
+  if (t.context.env.supportsTimeSkipping) {
+    t.pass("Test Server doesn't support worker deployments");
+    return;
+  }
+
+  const taskQueue = 'worker-deployment-dynamic-' + randomUUID();
+  const deploymentName = 'deployment-dynamic-' + randomUUID();
+  const client = t.context.env.client;
+
+  const version = {
+    buildId: '1.0',
+    deploymentName: deploymentName,
+  };
+
+  const worker = await Worker.create({
+    workflowsPath: require.resolve('./deployment-versioning-v1'),
+    activities,
+    taskQueue,
+    workerDeploymentOptions: {
+      useWorkerVersioning: true,
+      version: version,
+      defaultVersioningBehavior: 'auto-upgrade',
+    },
+  });
+
+  const workerPromise = worker.run();
+  workerPromise.catch((err) => {
+    t.fail('Worker run error: ' + JSON.stringify(err));
+  });
+
+  const describeResp = await waitUntilWorkerDeploymentVisible(client, version);
+  await setCurrentDeploymentVersion(client, describeResp.conflictToken, version);
+
+  const wf = await client.workflow.start('cooldynamicworkflow', {
+    taskQueue,
+    workflowId: 'dynamic-workflow-versioning-' + randomUUID(),
+  });
+
+  const result = await wf.result();
+  assert.equal(result, 'dynamic');
+
+  // Check history for versioning behavior
+  const history = await wf.fetchHistory();
+
+  const hasPinnedVersioningBehavior = history.events!.some(
+    (event) =>
+      event.workflowTaskCompletedEventAttributes &&
+      event.workflowTaskCompletedEventAttributes.versioningBehavior ===
+        temporal.api.enums.v1.VersioningBehavior.VERSIONING_BEHAVIOR_PINNED
+  );
+  assert.ok(hasPinnedVersioningBehavior, 'Expected workflow to use pinned versioning behavior');
+
+  worker.shutdown();
+  await workerPromise;
+  t.pass();
+});
+
+test('Workflows can use default versioning behavior', async (t) => {
+  const taskQueue = 'task-queue-default-versioning-' + randomUUID();
+  const deploymentName = 'deployment-default-versioning-' + randomUUID();
+  const client = t.context.env.client;
+
+  const workerV1 = {
+    buildId: '1.0',
+    deploymentName: deploymentName,
+  };
+
+  const worker = await Worker.create({
+    workflowsPath: require.resolve('./deployment-versioning-no-annotations'),
+    activities,
+    taskQueue,
+    workerDeploymentOptions: {
+      useWorkerVersioning: true,
+      version: workerV1,
+      defaultVersioningBehavior: 'pinned',
+    },
+  });
+
+  const workerPromise = worker.run();
+  workerPromise.catch((err) => {
+    t.fail('Worker run error: ' + JSON.stringify(err));
+  });
+
+  const describeResp = await waitUntilWorkerDeploymentVisible(client, workerV1);
+  await setCurrentDeploymentVersion(client, describeResp.conflictToken, workerV1);
+
+  const wf = await client.workflow.start('noVersioningAnnotationWorkflow', {
+    taskQueue,
+    workflowId: 'default-versioning-behavior-' + randomUUID(),
+  });
+
+  await wf.result();
+
+  // Check history for versioning behavior
+  const history = await wf.fetchHistory();
+
+  const hasPinnedVersioningBehavior = history.events!.some(
+    (event) =>
+      event.workflowTaskCompletedEventAttributes &&
+      event.workflowTaskCompletedEventAttributes.versioningBehavior ===
+        temporal.api.enums.v1.VersioningBehavior.VERSIONING_BEHAVIOR_PINNED
+  );
+  assert.ok(hasPinnedVersioningBehavior, 'Expected workflow to use pinned versioning behavior');
+
+  worker.shutdown();
+  await workerPromise;
+  t.pass();
+});
+
 async function setRampingVersion(
   client: Client,
   conflictToken: Uint8Array,

packages/worker/src/worker-options.ts

@@ -138,11 +138,10 @@ export interface WorkerOptions {
     useWorkerVersioning: boolean;
 
     /**
-     * If specified, the default versioning behavior to use for all workflows on this worker.
-     * If not specified, and `useWorkerVersioning` is true, workflows that do not specify a
-     * versioning behavior via {@link TODO} will cause an error to be thrown on startup.
+     * The default versioning behavior to use for all workflows on this worker. Specifying a default
+     * behavior is required,
      */
-    defaultVersioningBehavior?: VersioningBehavior;
+    defaultVersioningBehavior: VersioningBehavior;
   };
 
   /**

packages/workflow/src/worker-interface.ts

@@ -82,6 +82,9 @@ export function initRuntime(options: WorkflowCreateOptionsInternal): void {
     if (isWorkflowFunctionWithOptions(workflowFn)) {
       activator.workflow = workflowFn;
       activator.versioningBehavior = workflowFn.options.versioningBehavior;
+    } else if (isWorkflowFunctionWithOptions(defaultWorkflowFn)) {
+      activator.workflow = defaultWorkflowFn;
+      activator.versioningBehavior = defaultWorkflowFn.options.versioningBehavior;
     } else if (typeof workflowFn === 'function') {
       activator.workflow = workflowFn;
     } else if (typeof defaultWorkflowFn === 'function') {