Worker Deployment Versioning (#1679)

Co-authored-by: James Watkins-Harvey <[email protected]>

Author: Sushisource

package-lock.json

@@ -1227,6 +1227,7 @@ export class WorkflowClient extends BaseClient {
       cronSchedule: options.cronSchedule,
       header: { fields: headers },
       priority: options.priority ? compilePriority(options.priority) : undefined,
+      versioningOverride: options.versioningOverride ?? undefined,
     };
     try {
       return (await this.workflowService.signalWithStartWorkflowExecution(req)).runId;
@@ -1296,6 +1297,7 @@ export class WorkflowClient extends BaseClient {
       cronSchedule: opts.cronSchedule,
       header: { fields: headers },
       priority: opts.priority ? compilePriority(opts.priority) : undefined,
+      versioningOverride: opts.versioningOverride ?? undefined,
     };
   }
 

packages/client/src/workflow-client.ts

@@ -1,7 +1,14 @@
-import { CommonWorkflowOptions, SignalDefinition, WithWorkflowArgs, Workflow } from '@temporalio/common';
+import {
+  CommonWorkflowOptions,
+  SignalDefinition,
+  WithWorkflowArgs,
+  Workflow,
+  VersioningOverride,
+  toCanonicalString,
+} from '@temporalio/common';
 import { Duration, msOptionalToTs } from '@temporalio/common/lib/time';
 import { Replace } from '@temporalio/common/lib/type-helpers';
-import { google } from '@temporalio/proto';
+import { google, temporal } from '@temporalio/proto';
 
 export * from '@temporalio/common/lib/workflow-options';
 
@@ -38,9 +45,15 @@ export interface WorkflowOptions extends CommonWorkflowOptions {
 
   /**
    * Amount of time to wait before starting the workflow.
-   *
    */
   startDelay?: Duration;
+
+  /**
+   * Override the versioning behavior of the Workflow that is about to be started.
+   *
+   * @experimental Deployment based versioning is experimental and may change in the future.
+   */
+  versioningOverride?: VersioningOverride;
 }
 
 export type WithCompiledWorkflowOptions<T extends WorkflowOptions> = Replace<
@@ -50,18 +63,21 @@ export type WithCompiledWorkflowOptions<T extends WorkflowOptions> = Replace<
     workflowRunTimeout?: google.protobuf.IDuration;
     workflowTaskTimeout?: google.protobuf.IDuration;
     startDelay?: google.protobuf.IDuration;
+    versioningOverride?: temporal.api.workflow.v1.IVersioningOverride;
   }
 >;
 
 export function compileWorkflowOptions<T extends WorkflowOptions>(options: T): WithCompiledWorkflowOptions<T> {
-  const { workflowExecutionTimeout, workflowRunTimeout, workflowTaskTimeout, startDelay, ...rest } = options;
+  const { workflowExecutionTimeout, workflowRunTimeout, workflowTaskTimeout, startDelay, versioningOverride, ...rest } =
+    options;
 
   return {
     ...rest,
     workflowExecutionTimeout: msOptionalToTs(workflowExecutionTimeout),
     workflowRunTimeout: msOptionalToTs(workflowRunTimeout),
     workflowTaskTimeout: msOptionalToTs(workflowTaskTimeout),
     startDelay: msOptionalToTs(startDelay),
+    versioningOverride: versioningOverrideToProto(versioningOverride),
   };
 }
 
@@ -109,3 +125,25 @@ export interface WorkflowSignalWithStartOptionsWithArgs<SignalArgs extends any[]
  * Options for starting a Workflow
  */
 export type WorkflowStartOptions<T extends Workflow = Workflow> = WithWorkflowArgs<T, WorkflowOptions>;
+
+function versioningOverrideToProto(
+  vo: VersioningOverride | undefined
+): temporal.api.workflow.v1.IVersioningOverride | undefined {
+  if (!vo) return undefined;
+
+  // TODO: Remove deprecated field assignments when versioning is non-experimental
+  if (vo === 'AUTO_UPGRADE') {
+    return {
+      autoUpgrade: true,
+      behavior: temporal.api.enums.v1.VersioningBehavior.VERSIONING_BEHAVIOR_AUTO_UPGRADE,
+    };
+  }
+
+  return {
+    pinned: {
+      version: vo.pinnedTo,
+    },
+    behavior: temporal.api.enums.v1.VersioningBehavior.VERSIONING_BEHAVIOR_PINNED,
+    pinnedVersion: toCanonicalString(vo.pinnedTo),
+  };
+}

packages/client/src/workflow-options.ts

@@ -22,6 +22,8 @@ export * from './logger';
 export * from './priority';
 export * from './retry-policy';
 export type { Timestamp, Duration, StringValue } from './time';
+export * from './worker-deployments';
+export * from './workflow-definition-options';
 export * from './workflow-handle';
 export * from './workflow-options';
 export * from './versioning-intent';

packages/common/src/index.ts

@@ -0,0 +1,70 @@
+import type { temporal } from '@temporalio/proto';
+import { makeProtoEnumConverters } from './internal-workflow';
+
+/**
+ * Represents the version of a specific worker deployment.
+ *
+ * @experimental Deployment based versioning is experimental and may change in the future.
+ */
+export interface WorkerDeploymentVersion {
+  readonly buildId: string;
+  readonly deploymentName: string;
+}
+
+/**
+ * @returns The canonical representation of a deployment version, which is a string in the format
+ * `deploymentName.buildId`.
+ */
+export function toCanonicalString(version: WorkerDeploymentVersion): string {
+  return `${version.deploymentName}.${version.buildId}`;
+}
+
+/**
+ * Specifies when a workflow might move from a worker of one Build Id to another.
+ *
+ * * 'PINNED' - The workflow will be pinned to the current Build ID unless manually moved.
+ * * 'AUTO_UPGRADE' - The workflow will automatically move to the latest version (default Build ID
+ *    of the task queue) when the next task is dispatched.
+ *
+ * @experimental Deployment based versioning is experimental and may change in the future.
+ */
+export const VersioningBehavior = {
+  PINNED: 'PINNED',
+  AUTO_UPGRADE: 'AUTO_UPGRADE',
+} as const;
+export type VersioningBehavior = (typeof VersioningBehavior)[keyof typeof VersioningBehavior];
+
+export const [encodeVersioningBehavior, decodeVersioningBehavior] = makeProtoEnumConverters<
+  temporal.api.enums.v1.VersioningBehavior,
+  typeof temporal.api.enums.v1.VersioningBehavior,
+  keyof typeof temporal.api.enums.v1.VersioningBehavior,
+  typeof VersioningBehavior,
+  'VERSIONING_BEHAVIOR_'
+>(
+  {
+    [VersioningBehavior.PINNED]: 1,
+    [VersioningBehavior.AUTO_UPGRADE]: 2,
+    UNSPECIFIED: 0,
+  } as const,
+  'VERSIONING_BEHAVIOR_'
+);
+
+/**
+ * Represents versioning overrides. For example, when starting workflows.
+ */
+export type VersioningOverride = PinnedVersioningOverride | 'AUTO_UPGRADE';
+
+/**
+ * Workflow will be pinned to a specific deployment version.
+ */
+export interface PinnedVersioningOverride {
+  /**
+   * The worker deployment version to pin the workflow to.
+   */
+  pinnedTo: WorkerDeploymentVersion;
+}
+
+/**
+ * The workflow will auto-upgrade to the current deployment version on the next workflow task.
+ */
+export type AutoUpgradeVersioningOverride = 'AUTO_UPGRADE';

packages/common/src/worker-deployments.ts

@@ -0,0 +1,20 @@
+import { VersioningBehavior } from './worker-deployments';
+
+/**
+ * Options that can be used when defining a workflow via {@link setWorkflowOptions}.
+ */
+export interface WorkflowDefinitionOptions {
+  versioningBehavior?: VersioningBehavior;
+}
+
+type AsyncFunction<Args extends any[], ReturnType> = (...args: Args) => Promise<ReturnType>;
+export type WorkflowDefinitionOptionsOrGetter = WorkflowDefinitionOptions | (() => WorkflowDefinitionOptions);
+
+/**
+ * @internal
+ * @hidden
+ * A workflow function that has been defined with options from {@link WorkflowDefinitionOptions}.
+ */
+export interface WorkflowFunctionWithOptions<Args extends any[], ReturnType> extends AsyncFunction<Args, ReturnType> {
+  workflowDefinitionOptions: WorkflowDefinitionOptionsOrGetter;
+}

packages/common/src/workflow-definition-options.ts

@@ -5,6 +5,7 @@ import { Duration } from './time';
 import { makeProtoEnumConverters } from './internal-workflow';
 import { SearchAttributePair, SearchAttributes, TypedSearchAttributes } from './search-attributes';
 import { Priority } from './priority';
+import { WorkflowFunctionWithOptions } from './workflow-definition-options';
 
 /**
  * Defines what happens when trying to start a Workflow with the same ID as a *Closed* Workflow.
@@ -243,7 +244,9 @@ export interface WorkflowDurationOptions {
 
 export type CommonWorkflowOptions = BaseWorkflowOptions & WorkflowDurationOptions;
 
-export function extractWorkflowType<T extends Workflow>(workflowTypeOrFunc: string | T): string {
+export function extractWorkflowType<T extends Workflow>(
+  workflowTypeOrFunc: string | T | WorkflowFunctionWithOptions<any[], any>
+): string {
   if (typeof workflowTypeOrFunc === 'string') return workflowTypeOrFunc as string;
   if (typeof workflowTypeOrFunc === 'function') {
     if (workflowTypeOrFunc?.name) return workflowTypeOrFunc.name;