Dynamic workflows

Author: Sushisource

README.md

@@ -510,7 +510,7 @@ class GreetingWorkflow:
                 start_to_close_timeout=timedelta(seconds=5),
             )
             workflow.logger.debug("Greeting set to %s", self._current_greeting)
-            
+
             # Wait for salutation update or complete signal (this can be
             # cancelled)
             await asyncio.wait(
@@ -536,7 +536,7 @@ class GreetingWorkflow:
     @workflow.query
     def current_greeting(self) -> str:
         return self._current_greeting
-    
+
     @workflow.update
     def set_and_get_greeting(self, greeting: str) -> str:
       old = self._current_greeting
@@ -622,7 +622,7 @@ Here are the decorators that can be applied:
   * May mutate workflow state, and make calls to other workflow APIs like starting activities, etc.
   * Also accepts the `name` and `dynamic` parameters like signal, with the same semantics.
   * Update handlers may optionally define a validator method by decorating it with `@update_handler_method.validator`.
-    To reject an update before any events are written to history, throw an exception in a validator. Validators cannot 
+    To reject an update before any events are written to history, throw an exception in a validator. Validators cannot
     be `async`, cannot mutate workflow state, and return nothing.
   * See [Signal and update handlers](#signal-and-update-handlers) below
 * `@workflow.query` - Defines a method as a query
@@ -994,7 +994,7 @@ To run an entire workflow outside of a sandbox, set `sandboxed=False` on the `@w
 it. This will run the entire workflow outside of the workflow which means it can share global state and other bad
 things.
 
-To disable the sandbox entirely for a worker, set the `Worker` init's `workflow_runner` keyword argument to 
+To disable the sandbox entirely for a worker, set the `Worker` init's `workflow_runner` keyword argument to
 `temporalio.worker.UnsandboxedWorkflowRunner()`. This value is defaulted to
 `temporalio.worker.workflow_sandbox.SandboxedWorkflowRunner()` so by changing it to the unsandboxed runner, the sandbox
 will not be used at all.

temporalio/worker/_workflow_instance.py

@@ -414,6 +414,20 @@ def activate(
                     )
                 activation_err = None
 
+        # If this workflow is dynamic, try calling the dynamic versioning behavior
+        if self._defn.name is None:
+            dvb = self.workflow_get_dynamic_versioning_behavior()
+            if dvb:
+                with self._as_read_only():
+                    vb = dvb()
+                    if (
+                        vb
+                        != temporalio.api.enums.v1.VersioningBehavior.VERSIONING_BEHAVIOR_UNSPECIFIED
+                    ):
+                        self._current_completion.successful.versioning_behavior = (
+                            vb._to_proto()
+                        )
+
         # If we're deleting, there better be no more tasks. It is important for
         # the integrity of the system that we check this. If there are tasks
         # remaining, they and any associated coroutines will get garbage
@@ -434,6 +448,7 @@ def activate(
             )
             # Set completion failure
             self._current_completion.failed.failure.SetInParent()
+            # TODO: Review - odd that we don't un-set success here?
             try:
                 self._failure_converter.to_failure(
                     activation_err,
@@ -1025,6 +1040,14 @@ def workflow_get_update_validator(self, name: Optional[str]) -> Optional[Callabl
         # Bind if a method
         return defn.bind_validator(self._object) if defn.is_method else defn.validator
 
+    def workflow_get_dynamic_versioning_behavior(
+        self,
+    ) -> Optional[Callable[[], temporalio.common.VersioningBehavior]]:
+        defn = self._defn.dynamic_versioning_behavior
+        if not defn:
+            return None
+        return temporalio.workflow._bind_method(self._object, defn)
+
     def workflow_info(self) -> temporalio.workflow.Info:
         return self._outbound.info()
 

temporalio/workflow.py

@@ -130,8 +130,6 @@ def defn(
             applied in addition to ones set on the worker constructor. If
             ``Exception`` is set, it effectively will fail a workflow/update in
             all user exception cases. WARNING: This setting is experimental.
-        versioning_behavior: Specifies when this workflow might move from a worker
-            of one Build Id to another. WARNING: This setting is experimental.
     """
 
     def decorator(cls: ClassType) -> ClassType:
@@ -205,6 +203,7 @@ def run(
     Args:
         fn: The function to decorate.
         versioning_behavior: Specifies the versioning behavior to use for this workflow.
+            WARNING: This setting is experimental.
     """
 
     def decorator(
@@ -259,6 +258,25 @@ class UnfinishedSignalHandlersWarning(RuntimeWarning):
     """The workflow exited before all signal handlers had finished executing."""
 
 
+def dynamic_versioning_behavior(
+    fn: MethodSyncNoParam[SelfType, temporalio.common.VersioningBehavior],
+) -> MethodSyncNoParam[SelfType, temporalio.common.VersioningBehavior]:
+    """Decorator for specifying versioning behavior in dynamic workflows.
+
+    This function must be read-only and not mutate workflow state. Any mutation could lead to
+    nondeterministic behavior. If the workflow function specifies a `versioning_behavior` on
+    both ``@workflow.run`` and with a getter, the getter's value will be used whenever it is not
+    ``VersioningBehavior.UNSPECIFIED``.
+
+    WARNING: This setting is experimental.
+
+    Args:
+        fn: The method to decorate
+    """
+    setattr(fn, "__temporal_dynamic_versioning_behavior", True)
+    return fn
+
+
 @overload
 def signal(
     fn: CallableSyncOrAsyncReturnNoneType,
@@ -1464,6 +1482,7 @@ class _Definition:
     arg_types: Optional[List[Type]] = None
     ret_type: Optional[Type] = None
     versioning_behavior: Optional[temporalio.common.VersioningBehavior] = None
+    dynamic_versioning_behavior: Optional[CallableSyncNoParam] = None
 
     @staticmethod
     def from_class(cls: Type) -> Optional[_Definition]:
@@ -1531,6 +1550,7 @@ def _apply_to_class(
         signals: Dict[Optional[str], _SignalDefinition] = {}
         queries: Dict[Optional[str], _QueryDefinition] = {}
         updates: Dict[Optional[str], _UpdateDefinition] = {}
+        dynamic_versioning_behavior: Optional[CallableSyncNoParam] = None
         for name, member in inspect.getmembers(cls):
             if hasattr(member, "__temporal_workflow_run"):
                 seen_run_attr = getattr(member, "__temporal_workflow_run")
@@ -1574,6 +1594,15 @@ def _apply_to_class(
                     queries[query_defn.name] = query_defn
             elif name == "__init__" and hasattr(member, "__temporal_workflow_init"):
                 init_fn = member
+            elif hasattr(member, "__temporal_dynamic_versioning_behavior"):
+                if workflow_name:
+                    issues.append(
+                        "Non-dynamic workflows should not specify "
+                        "@workflow.dynamic_versioning_behavior, which was found on "
+                        f"{cls.__qualname__}.{name}. Use the versioning_behavior "
+                        "argument to @workflow.run instead."
+                    )
+                dynamic_versioning_behavior = member
             elif isinstance(member, UpdateMethodMultiParam):
                 update_defn = member._defn
                 if update_defn.name in updates:
@@ -1658,6 +1687,7 @@ def _apply_to_class(
             sandboxed=sandboxed,
             failure_exception_types=failure_exception_types,
             versioning_behavior=seen_run_attr.versioning_behavior,
+            dynamic_versioning_behavior=dynamic_versioning_behavior,
         )
         setattr(cls, "__temporal_workflow_definition", defn)
         setattr(run_fn, "__temporal_workflow_definition", defn)

tests/test_workflow.py

@@ -225,12 +225,16 @@ def update2(self, arg1: str):
     def base_update(self):
         pass
 
+    @workflow.dynamic_versioning_behavior
+    def i_shouldnt_exist(self) -> VersioningBehavior:
+        return VersioningBehavior.PINNED
+
 
 def test_workflow_defn_bad():
     with pytest.raises(ValueError) as err:
         workflow.defn(BadDefn)
 
-    assert "Invalid workflow class for 9 reasons" in str(err.value)
+    assert "Invalid workflow class for 10 reasons" in str(err.value)
     assert "Missing @workflow.run method" in str(err.value)
     assert (
         "Multiple signal methods found for signal1 (at least on signal2 and signal1)"
@@ -264,6 +268,10 @@ def test_workflow_defn_bad():
         "@workflow.update defined on BadDefnBase.base_update but not on the override"
         in str(err.value)
     )
+    assert (
+        "Non-dynamic workflows should not specify @workflow.dynamic_versioning_behavior, which "
+        "was found on BadDefn.i_shouldnt_exist" in str(err.value)
+    )
 
 
 def test_workflow_defn_local_class():
@@ -422,9 +430,9 @@ def test_parameters_identical_up_to_naming():
     for f1, f2 in itertools.combinations(fns, 2):
         name1, name2 = f1.__name__, f2.__name__
         expect_equal = name1[0] == name2[0]
-        assert (
-            workflow._parameters_identical_up_to_naming(f1, f2) == (expect_equal)
-        ), f"expected {name1} and {name2} parameters{' ' if expect_equal else ' not '}to compare equal"
+        assert workflow._parameters_identical_up_to_naming(f1, f2) == (expect_equal), (
+            f"expected {name1} and {name2} parameters{' ' if expect_equal else ' not '}to compare equal"
+        )
 
 
 @workflow.defn

tests/worker/test_worker.py

@@ -4,10 +4,11 @@
 import concurrent.futures
 import uuid
 from datetime import timedelta
-from typing import Any, Awaitable, Callable, Optional
+from typing import Any, Awaitable, Callable, Optional, Sequence
 
 import pytest
 
+import temporalio.api.enums.v1
 import temporalio.worker._worker
 from temporalio import activity, workflow
 from temporalio.api.workflowservice.v1 import (
@@ -19,7 +20,7 @@
     SetWorkerDeploymentRampingVersionResponse,
 )
 from temporalio.client import BuildIdOpAddNewDefault, Client, TaskReachabilityType
-from temporalio.common import VersioningBehavior
+from temporalio.common import RawValue, VersioningBehavior
 from temporalio.service import RPCError
 from temporalio.testing import WorkflowEnvironment
 from temporalio.worker import (
@@ -729,13 +730,13 @@ async def test_worker_deployment_ramp(client: Client, env: WorkflowEnvironment):
             await set_ramping_version(client, conflict_token, v2, 0)
         ).conflict_token
         for i in range(3):
-            wf = await client.start_workflow(
+            wfa = await client.start_workflow(
                 DeploymentVersioningWorkflowV1AutoUpgrade.run,
                 id=f"versioning-ramp-0-{i}-{uuid.uuid4()}",
                 task_queue=w1.task_queue,
             )
-            await wf.signal(DeploymentVersioningWorkflowV1AutoUpgrade.do_finish)
-            res = await wf.result()
+            await wfa.signal(DeploymentVersioningWorkflowV1AutoUpgrade.do_finish)
+            res = await wfa.result()
             assert res == "version-v1"
 
         # Set ramp to 50 and eventually verify workflows run on both versions
@@ -759,6 +760,92 @@ async def check_results():
         await assert_eventually(check_results)
 
 
+@workflow.defn(dynamic=True)
+class DynamicWorkflowVersioningOnRun:
+    @workflow.run(versioning_behavior=VersioningBehavior.PINNED)
+    async def run(self, args: Sequence[RawValue]) -> str:
+        return "dynamic"
+
+
+@workflow.defn(dynamic=True)
+class DynamicWorkflowVersioningWithGetter:
+    @workflow.run
+    async def run(self, args: Sequence[RawValue]) -> str:
+        return "dynamic"
+
+    @workflow.dynamic_versioning_behavior
+    def huh(self) -> VersioningBehavior:
+        return VersioningBehavior.PINNED
+
+
+async def _test_dynamic_workflow_versioning(
+    client: Client, workflow_class, expected_versioning_behavior
+):
+    deployment_name = f"deployment-dynamic-{uuid.uuid4()}"
+    worker_v1 = WorkerDeploymentVersion(deployment_name=deployment_name, build_id="1.0")
+
+    async with new_worker(
+        client,
+        workflow_class,
+        deployment_options=WorkerDeploymentOptions(
+            version=worker_v1,
+            use_worker_versioning=True,
+        ),
+    ) as w:
+        describe_resp = await wait_until_worker_deployment_visible(
+            client,
+            worker_v1,
+        )
+        await set_current_deployment_version(
+            client, describe_resp.conflict_token, worker_v1
+        )
+
+        wf = await client.start_workflow(
+            "cooldynamicworkflow",
+            id=f"dynamic-workflow-versioning-{uuid.uuid4()}",
+            task_queue=w.task_queue,
+        )
+        result = await wf.result()
+        assert result == "dynamic"
+
+        history = await wf.fetch_history()
+        assert any(
+            event.HasField("workflow_task_completed_event_attributes")
+            and event.workflow_task_completed_event_attributes.versioning_behavior
+            == expected_versioning_behavior
+            for event in history.events
+        )
+
+
+async def test_worker_deployment_dynamic_workflow_on_run(
+    client: Client, env: WorkflowEnvironment
+):
+    if env.supports_time_skipping:
+        pytest.skip("Test Server doesn't support worker deployments")
+
+    await _test_dynamic_workflow_versioning(
+        client,
+        DynamicWorkflowVersioningOnRun,
+        temporalio.api.enums.v1.VersioningBehavior.VERSIONING_BEHAVIOR_PINNED,
+    )
+
+
+async def test_worker_deployment_dynamic_workflow_getter(
+    client: Client, env: WorkflowEnvironment
+):
+    if env.supports_time_skipping:
+        pytest.skip("Test Server doesn't support worker deployments")
+
+    await _test_dynamic_workflow_versioning(
+        client,
+        DynamicWorkflowVersioningWithGetter,
+        temporalio.api.enums.v1.VersioningBehavior.VERSIONING_BEHAVIOR_PINNED,
+    )
+
+
+# TODO: Test for fail at registration time if deployment versioning on, no default, no behavior
+
+
 async def wait_until_worker_deployment_visible(
     client: Client, version: WorkerDeploymentVersion
 ) -> DescribeWorkerDeploymentResponse: