CaN USE_RAMPING_VERSION versioning behaviour

Author: THardy98

packages/common/src/worker-deployments.ts

@@ -78,10 +78,25 @@ export type AutoUpgradeVersioningOverride = 'AUTO_UPGRADE';
  *   behavior, the base version of the new run will be the Target Version as described above, but the
  *   effective version will be whatever is specified by the Versioning Override until the override is removed.
  *
+ * USE_RAMPING_VERSION - Use the Ramping Version of the workflow's task queue at start time, regardless of the workflow's
+ *   Target Version (according to f(workflow_id, ramp_percentage)). After the first workflow task completes,
+ *   the workflow will use whatever Versioning Behavior it is annotated with. If there is no Ramping
+ *   Version by the time that the first workflow task is dispatched, it will be sent to the Current Version.
+ *
+ *   It is highly discouraged to use this if the workflow is annotated with AutoUpgrade behavior, because
+ *   this setting ONLY applies to the first task of the workflow. If, after the first task, the workflow
+ *   is AutoUpgrade, it will behave like a normal AutoUpgrade workflow and go to the Target Version, which
+ *   may be the Current Version instead of the Ramping Version.
+ *
+ *   Note that if the workflow being continued has a Pinned override, that override will be inherited by the
+ *   new workflow run regardless of the ContinueAsNewVersioningBehavior specified in the continue-as-new
+ *   command. Versioning Override always takes precedence until it's removed manually via UpdateWorkflowExecutionOptions.
+ *
  * @experimental Versioning semantics with continue-as-new are experimental and may change in the future.
  */
 export const InitialVersioningBehavior = {
   AUTO_UPGRADE: 'AUTO_UPGRADE',
+  USE_RAMPING_VERSION: 'USE_RAMPING_VERSION',
 } as const;
 export type InitialVersioningBehavior = (typeof InitialVersioningBehavior)[keyof typeof InitialVersioningBehavior];
 
@@ -94,6 +109,7 @@ export const [encodeInitialVersioningBehavior, decodeInitialVersioningBehavior]
 >(
   {
     [InitialVersioningBehavior.AUTO_UPGRADE]: 1,
+    [InitialVersioningBehavior.USE_RAMPING_VERSION]: 2,
     UNSPECIFIED: 0,
   } as const,
   'CONTINUE_AS_NEW_VERSIONING_BEHAVIOR_'

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

@@ -1,4 +1,12 @@
-import { setWorkflowOptions, workflowInfo, sleep, makeContinueAsNewFunc } from '@temporalio/workflow';
+import {
+  condition,
+  setHandler,
+  setWorkflowOptions,
+  workflowInfo,
+  sleep,
+  makeContinueAsNewFunc,
+} from '@temporalio/workflow';
+import { unblockSignal } from '../workflows';
 
 setWorkflowOptions({ versioningBehavior: 'PINNED' }, continueAsNewWithVersionUpgrade);
 export async function continueAsNewWithVersionUpgrade(attempt: number): Promise<string> {
@@ -17,3 +25,19 @@ export async function continueAsNewWithVersionUpgrade(attempt: number): Promise<
     }
   }
 }
+
+setWorkflowOptions({ versioningBehavior: 'PINNED' }, continueAsNewWithRampingVersion);
+export async function continueAsNewWithRampingVersion(attempt: number): Promise<string> {
+  if (attempt > 0) {
+    return 'v1.0';
+  }
+
+  let shouldContinue = false;
+  setHandler(unblockSignal, () => void (shouldContinue = true));
+  await condition(() => shouldContinue);
+
+  const canWithRampingVersion = makeContinueAsNewFunc<typeof continueAsNewWithRampingVersion>({
+    initialVersioningBehavior: 'USE_RAMPING_VERSION',
+  });
+  return await canWithRampingVersion(attempt + 1);
+}

packages/test/src/deployment-versioning-can-v2/index.ts

@@ -5,3 +5,9 @@ setWorkflowOptions({ versioningBehavior: 'PINNED' }, continueAsNewWithVersionUpg
 export async function continueAsNewWithVersionUpgrade(attempt: number): Promise<string> {
   return 'v2.0';
 }
+
+setWorkflowOptions({ versioningBehavior: 'PINNED' }, continueAsNewWithRampingVersion);
+// eslint-disable-next-line @typescript-eslint/no-unused-vars
+export async function continueAsNewWithRampingVersion(attempt: number): Promise<string> {
+  return 'v2.0';
+}

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

@@ -604,6 +604,69 @@ test('ContinueAsNew with version upgrade', async (t) => {
   await worker2Promise;
 });
 
+test('ContinueAsNew with ramping version', async (t) => {
+  const taskQueue = 'can-ramping-version-' + randomUUID();
+  const deploymentName = 'deployment-can-ramping-' + randomUUID();
+  const { client, nativeConnection } = t.context.env;
+  const { createNativeConnection } = helpers(t);
+
+  const v1 = { buildId: '1.0', deploymentName };
+  const v2 = { buildId: '2.0', deploymentName };
+
+  const worker1 = await Worker.create({
+    workflowsPath: require.resolve('./deployment-versioning-can-v1'),
+    taskQueue,
+    workerDeploymentOptions: {
+      useWorkerVersioning: true,
+      version: v1,
+      defaultVersioningBehavior: 'PINNED',
+    },
+    connection: nativeConnection,
+  });
+  const worker1Promise = worker1.run();
+  worker1Promise.catch((err) => t.fail('Worker 1.0 error: ' + err));
+
+  const worker2Connection = await createNativeConnection();
+  t.teardown(() => worker2Connection.close());
+  const worker2 = await Worker.create({
+    workflowsPath: require.resolve('./deployment-versioning-can-v2'),
+    taskQueue,
+    workerDeploymentOptions: {
+      useWorkerVersioning: true,
+      version: v2,
+      defaultVersioningBehavior: 'PINNED',
+    },
+    connection: worker2Connection,
+  });
+  const worker2Promise = worker2.run();
+  worker2Promise.catch((err) => t.fail('Worker 2.0 error: ' + err));
+
+  const describeResp1 = await waitUntilWorkerDeploymentVisible(client, v1);
+  const setResp1 = await setCurrentDeploymentVersion(client, describeResp1.conflictToken, v1);
+  await waitForRoutingConfigPropagation(client, deploymentName, v1.buildId);
+
+  const handle = await client.workflow.start('continueAsNewWithRampingVersion', {
+    args: [0],
+    taskQueue,
+    workflowId: 'can-ramping-version-' + randomUUID(),
+  });
+  await waitForWorkflowRunningOnVersion(client, handle, v1.buildId);
+
+  await waitUntilWorkerDeploymentVisible(client, v2);
+  await setRampingVersion(client, setResp1.conflictToken, v2, 0);
+  await waitForRoutingConfigPropagation(client, deploymentName, v1.buildId, v2.buildId);
+
+  await handle.signal(unblockSignal);
+
+  const result = await handle.result();
+  t.is(result, 'v2.0');
+
+  worker1.shutdown();
+  worker2.shutdown();
+  await worker1Promise;
+  await worker2Promise;
+});
+
 ///////////////////////////////
 
 /**