Add orchestration version to orchestration context

Author: andystaples

durabletask/task.py

@@ -35,6 +35,21 @@ def instance_id(self) -> str:
         """
         pass
 
+    @property
+    @abstractmethod
+    def version(self) -> Optional[str]:
+        """Get the version of the orchestration instance.
+
+        This version is set when the orchestration is scheduled and can be used
+        to determine which version of the orchestrator function is being executed.
+
+        Returns
+        -------
+        Optional[str]
+            The version of the orchestration instance, or None if not set.
+        """
+        pass
+
     @property
     @abstractmethod
     def current_utc_datetime(self) -> datetime:

durabletask/worker.py

@@ -651,6 +651,7 @@ def __init__(self, instance_id: str, registry: _Registry):
         self._current_utc_datetime = datetime(1000, 1, 1)
         self._instance_id = instance_id
         self._registry = registry
+        self._version: Optional[str] = None
         self._completion_status: Optional[pb.OrchestrationStatus] = None
         self._received_events: dict[str, list[Any]] = {}
         self._pending_events: dict[str, list[task.CompletableTask]] = {}
@@ -776,6 +777,10 @@ def next_sequence_number(self) -> int:
     def instance_id(self) -> str:
         return self._instance_id
 
+    @property
+    def version(self) -> Optional[str]:
+        return self._version
+
     @property
     def current_utc_datetime(self) -> datetime:
         return self._current_utc_datetime
@@ -977,11 +982,12 @@ def execute(
 
             # Process versioning if applicable
             execution_started_events = [e.executionStarted for e in old_events if e.HasField("executionStarted")]
+            # We only check versioning if there are executionStarted events - otherwise, on the first replay when
+            # ctx.version will be Null, we may invalidate orchestrations early depending on the versioning strategy.
             if self._registry.versioning and len(execution_started_events) > 0:
-                execution_started_event = execution_started_events[-1]
                 version_failure = self.evaluate_orchestration_versioning(
                     self._registry.versioning,
-                    execution_started_event.version.value if execution_started_event.version else None,
+                    ctx.version
                 )
                 if version_failure:
                     self._logger.warning(
@@ -1059,6 +1065,9 @@ def process_event(
                         f"A '{event.executionStarted.name}' orchestrator was not registered."
                     )
 
+                if event.executionStarted.version:
+                    ctx._version = event.executionStarted.version.value
+
                 # deserialize the input, if any
                 input = None
                 if (

tests/durabletask-azuremanaged/test_dts_orchestration_versioning_e2e.py

@@ -23,6 +23,10 @@ def plus_one(_: task.ActivityContext, input: int) -> int:
     return input + 1
 
 
+def plus_two(_: task.ActivityContext, input: int) -> int:
+    return input + 2
+
+
 def single_activity(ctx: task.OrchestrationContext, start_val: int):
     yield ctx.call_activity(plus_one, input=start_val)
     return "Success"
@@ -153,6 +157,35 @@ def test_upper_version_worker_succeeds():
 
 
 def test_upper_version_worker_strict_fails():
+    # Start a worker, which will connect to the sidecar in a background thread
+    with DurableTaskSchedulerWorker(host_address=endpoint, secure_channel=True,
+                                    taskhub=taskhub_name, token_credential=None) as w:
+        w.add_orchestrator(single_activity)
+        w.add_activity(plus_one)
+        w.use_versioning(worker.VersioningOptions(
+            version="1.1.0",
+            default_version="1.1.0",
+            match_strategy=worker.VersionMatchStrategy.STRICT,
+            failure_strategy=worker.VersionFailureStrategy.FAIL
+        ))
+        w.start()
+
+        task_hub_client = DurableTaskSchedulerClient(host_address=endpoint, secure_channel=True,
+                                                     taskhub=taskhub_name, token_credential=None,
+                                                     default_version="1.1.0")
+        id = task_hub_client.schedule_new_orchestration(single_activity, input=1, version="1.0.0")
+        state = task_hub_client.wait_for_orchestration_completion(
+            id, timeout=30)
+
+    assert state is not None
+    assert state.name == task.get_name(single_activity)
+    assert state.instance_id == id
+    assert state.runtime_status == client.OrchestrationStatus.FAILED
+    assert state.failure_details is not None
+    assert state.failure_details.message.find("The orchestration version '1.0.0' does not match the worker version '1.1.0'.") >= 0
+
+
+def test_reject_abandons_and_reprocess():
     # Start a worker, which will connect to the sidecar in a background thread
     instance_id: str = ''
     thrown = False
@@ -206,36 +239,58 @@ def test_upper_version_worker_strict_fails():
     assert state.failure_details is None
 
 
-def test_reject_abandons_and_reprocess():
+def multiversion_sequence(ctx: task.OrchestrationContext, start_val: int):
+    if ctx.version == "1.0.0":
+        result = yield ctx.call_activity(plus_one, input=start_val)
+    elif ctx.version == "1.1.0":
+        result = yield ctx.call_activity(plus_two, input=start_val)
+    else:
+        raise ValueError(f"Unsupported version: {ctx.version}")
+    return result
+
+
+def test_multiversion_orchestration_succeeds():
     # Start a worker, which will connect to the sidecar in a background thread
     with DurableTaskSchedulerWorker(host_address=endpoint, secure_channel=True,
                                     taskhub=taskhub_name, token_credential=None) as w:
-        w.add_orchestrator(single_activity)
+        w.add_orchestrator(multiversion_sequence)
         w.add_activity(plus_one)
+        w.add_activity(plus_two)
         w.use_versioning(worker.VersioningOptions(
             version="1.1.0",
             default_version="1.1.0",
-            match_strategy=worker.VersionMatchStrategy.STRICT,
+            match_strategy=worker.VersionMatchStrategy.CURRENT_OR_OLDER,
             failure_strategy=worker.VersionFailureStrategy.FAIL
         ))
         w.start()
 
         task_hub_client = DurableTaskSchedulerClient(host_address=endpoint, secure_channel=True,
                                                      taskhub=taskhub_name, token_credential=None,
                                                      default_version="1.1.0")
-        id = task_hub_client.schedule_new_orchestration(single_activity, input=1, version="1.0.0")
-        state = task_hub_client.wait_for_orchestration_completion(
-            id, timeout=30)
+        id = task_hub_client.schedule_new_orchestration(multiversion_sequence, input=1, version="1.0.0")
+        state = task_hub_client.wait_for_orchestration_completion(id, timeout=30)
 
-    assert state is not None
-    assert state.name == task.get_name(single_activity)
-    assert state.instance_id == id
-    assert state.runtime_status == client.OrchestrationStatus.FAILED
-    assert state.failure_details is not None
-    assert state.failure_details.message.find("The orchestration version '1.0.0' does not match the worker version '1.1.0'.") >= 0
+        id_2 = task_hub_client.schedule_new_orchestration(multiversion_sequence, input=1, version="1.1.0")
+        state_2 = task_hub_client.wait_for_orchestration_completion(id_2, timeout=30)
 
+    print(state.failure_details.message if state and state.failure_details else "State is None")
+    print(state_2.failure_details.message if state_2 and state_2.failure_details else "State is None")
 
-# Sub-orchestration tests
+    assert state is not None
+    assert state.name == task.get_name(multiversion_sequence)
+    assert state.instance_id == id
+    assert state.runtime_status == client.OrchestrationStatus.COMPLETED
+    assert state.failure_details is None
+    assert state.serialized_input == json.dumps(1)
+    assert state.serialized_output == json.dumps(2)
+
+    assert state_2 is not None
+    assert state_2.name == task.get_name(multiversion_sequence)
+    assert state_2.instance_id == id_2
+    assert state_2.runtime_status == client.OrchestrationStatus.COMPLETED
+    assert state_2.failure_details is None
+    assert state_2.serialized_input == json.dumps(1)
+    assert state_2.serialized_output == json.dumps(3)
 
 
 def sequence_suborchestator(ctx: task.OrchestrationContext, start_val: int):