CaN USE_RAMPING_VERSION versioning behaviour

Author: THardy98

temporalio/workflow.py

@@ -5426,6 +5426,25 @@ class ContinueAsNewVersioningBehavior(IntEnum):
     effective version will be whatever is specified by the Versioning Override until the override is removed.
     """
 
+    USE_RAMPING_VERSION = int(
+        temporalio.api.enums.v1.ContinueAsNewVersioningBehavior.CONTINUE_AS_NEW_VERSIONING_BEHAVIOR_USE_RAMPING_VERSION
+    )
+    """Use the Ramping Version of the workflow's task queue at start time, regardless of the workflow's
+    Target Version. 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.
+    """
+
 
 ServiceT = TypeVar("ServiceT")
 

tests/__init__.py

@@ -1 +1 @@
-DEV_SERVER_DOWNLOAD_VERSION = "v1.6.1-server-1.31.0-151.0"
+DEV_SERVER_DOWNLOAD_VERSION = "v1.7.0"

tests/worker/test_worker.py

@@ -1456,6 +1456,40 @@ async def run(self, attempt: int) -> str:  # type:ignore[reportUnusedParameter]
         return "v2.0"
 
 
+@workflow.defn(
+    name="ContinueAsNewWithRampingVersion",
+    versioning_behavior=VersioningBehavior.PINNED,
+)
+class ContinueAsNewWithRampingVersionV1:
+    def __init__(self) -> None:
+        self._should_continue_as_new = False
+
+    @workflow.run
+    async def run(self, attempt: int) -> str:
+        if attempt > 0:
+            return "v1.0"
+
+        await workflow.wait_condition(lambda: self._should_continue_as_new)
+        workflow.continue_as_new(
+            arg=attempt + 1,
+            initial_versioning_behavior=workflow.ContinueAsNewVersioningBehavior.USE_RAMPING_VERSION,
+        )
+
+    @workflow.signal
+    def do_continue_as_new(self) -> None:
+        self._should_continue_as_new = True
+
+
+@workflow.defn(
+    name="ContinueAsNewWithRampingVersion",
+    versioning_behavior=VersioningBehavior.PINNED,
+)
+class ContinueAsNewWithRampingVersionV2:
+    @workflow.run
+    async def run(self, attempt: int) -> str:  # type:ignore[reportUnusedParameter]
+        return "v2.0"
+
+
 async def wait_for_workflow_running_on_version(
     handle: WorkflowHandle[Any, Any], expected_build_id: str
 ) -> None:
@@ -1545,6 +1579,64 @@ async def test_continue_as_new_with_version_upgrade(
         assert result == "v2.0"
 
 
+async def test_continue_as_new_with_ramping_version(
+    client: Client, env: WorkflowEnvironment
+):
+    if env.supports_time_skipping:
+        pytest.skip("Test Server doesn't support worker deployments")
+
+    deployment_name = f"deployment-can-ramping-{uuid.uuid4()}"
+    v1 = WorkerDeploymentVersion(deployment_name=deployment_name, build_id="1.0")
+    v2 = WorkerDeploymentVersion(deployment_name=deployment_name, build_id="2.0")
+
+    async with (
+        new_worker(
+            client,
+            ContinueAsNewWithRampingVersionV1,
+            deployment_config=WorkerDeploymentConfig(
+                version=v1,
+                use_worker_versioning=True,
+            ),
+        ) as w1,
+        new_worker(
+            client,
+            ContinueAsNewWithRampingVersionV2,
+            deployment_config=WorkerDeploymentConfig(
+                version=v2,
+                use_worker_versioning=True,
+            ),
+            task_queue=w1.task_queue,
+        ),
+    ):
+        describe_resp = await wait_until_worker_deployment_visible(client, v1)
+
+        resp2 = await set_current_deployment_version(
+            client, describe_resp.conflict_token, v1
+        )
+        await wait_for_worker_deployment_routing_config_propagation(
+            client, deployment_name, v1.build_id
+        )
+
+        handle = await client.start_workflow(
+            "ContinueAsNewWithRampingVersion",
+            0,
+            id=f"test-can-ramping-version-{uuid.uuid4()}",
+            task_queue=w1.task_queue,
+        )
+        await wait_for_workflow_running_on_version(handle, v1.build_id)
+
+        await wait_until_worker_deployment_visible(client, v2)
+        await set_ramping_version(client, resp2.conflict_token, v2, 0)
+        await wait_for_worker_deployment_routing_config_propagation(
+            client, deployment_name, v1.build_id, v2.build_id
+        )
+
+        await handle.signal(ContinueAsNewWithRampingVersionV1.do_continue_as_new)
+
+        result = await handle.result()
+        assert result == "v2.0"
+
+
 def test_worker_config_matches_init_params():
     """WorkerConfig TypedDict keys must match Worker.__init__ kwargs."""
     import inspect