Reuse Azure Functions durable workers

azure-functions-durable/CHANGELOG.md

@@ -14,6 +14,11 @@ the synchronous client into synchronous functions and the asynchronous client
 into coroutine functions. Both clients support scheduled-task and history-export
 APIs without an async-to-sync bridge.
 
+CHANGED
+
+- Reused host-driven orchestration and entity workers across invocations,
+avoiding repeated allocation of unused worker resources.
+
 ## 2.0.0b1
 
 First preview (beta) release of `azure-functions-durable` 2.x — a ground-up

azure-functions-durable/azure/durable_functions/decorators/durable_app.py

@@ -233,6 +233,7 @@ def decorator(entity_func: task.Entity[Any, Any]) -> FunctionBuilder:
             if entity_name is not None:
                 entity_func.__durable_entity_name__ = entity_name  # type: ignore[union-attr]
             # Construct an orchestrator based on the end-user code
+            worker = DurableFunctionsWorker()
 
             # TODO: Because this handle method is the one actually exposed to the Functions SDK decorator,
             #       the parameter name will always be "context" here, even if the user specified a different name.
@@ -243,7 +244,7 @@ def decorator(entity_func: task.Entity[Any, Any]) -> FunctionBuilder:
             # binding converter to accept it; at runtime the host passes that
             # transport context (exposing ``.body``).
             def handle(context: func.EntityContext) -> str:
-                return DurableFunctionsWorker().execute_entity_batch_request(entity_func, context)
+                return worker.execute_entity_batch_request(entity_func, context)
 
             handle.entity_function = entity_func  # pyright: ignore[reportFunctionMemberAccess]
 

azure-functions-durable/azure/durable_functions/orchestrator.py

@@ -20,7 +20,11 @@ class Orchestrator:
     function.
     """
 
-    def __init__(self, orchestrator_func: Callable[..., Any]):
+    def __init__(
+            self,
+            orchestrator_func: Callable[..., Any],
+            worker: DurableFunctionsWorker | None = None,
+    ):
         """Create a new orchestrator wrapper for a user orchestrator function.
 
         The wrapped function may be a durabletask-native two-argument
@@ -30,6 +34,7 @@ def __init__(self, orchestrator_func: Callable[..., Any]):
         :param orchestrator_func: The user's orchestrator function to run.
         """
         self.fn: Callable[..., Any] = orchestrator_func
+        self._worker = worker if worker is not None else DurableFunctionsWorker()
 
     def handle(self, context: func.OrchestrationContext) -> str:
         """Handle the orchestration of the user defined generator function.
@@ -49,7 +54,7 @@ def handle(self, context: func.OrchestrationContext) -> str:
             Durable Functions host wire format) produced by this invocation.
         """
         self.durable_context = context
-        return DurableFunctionsWorker().execute_orchestration_request(self.fn, context)
+        return self._worker.execute_orchestration_request(self.fn, context)
 
     @classmethod
     def create(cls, fn: Callable[..., Any]) -> Callable[[Any], str]:
@@ -67,13 +72,15 @@ def create(cls, fn: Callable[..., Any]) -> Callable[[Any], str]:
             Handle function of the newly created orchestration client
         """
 
+        worker = DurableFunctionsWorker()
+
         # The generated handle is the function registered with the Azure
         # Functions host. Its ``context`` parameter must be annotated with
         # ``azure.functions.OrchestrationContext`` so the host's
         # orchestrationTrigger binding converter accepts it; at runtime the
         # host passes that transport context (exposing ``.body``).
         def handle(context: func.OrchestrationContext) -> str:
-            return Orchestrator(fn).handle(context)
+            return Orchestrator(fn, worker).handle(context)
 
         handle.orchestrator_function = fn  # pyright: ignore[reportFunctionMemberAccess]
 

azure-functions-durable/azure/durable_functions/worker.py

@@ -2,6 +2,7 @@
 # Licensed under the MIT License.
 
 import base64
+from threading import Lock
 from typing import Any, Optional
 
 from durabletask import task
@@ -41,10 +42,40 @@ def __init__(self) -> None:
         # azure-functions codec (df_dumps/df_loads) so user types round-trip in
         # the wire format the Durable Functions host extension expects.
         super().__init__(data_converter=DEFAULT_FUNCTIONS_DATA_CONVERTER)
+        self._registration_lock = Lock()
+        self._registered_orchestrator_functions: dict[
+            str, task.Orchestrator[Any, Any]
+        ] = {}
+        self._registered_entity_functions: dict[
+            str, task.Entity[Any, Any]
+        ] = {}
 
     def add_named_orchestrator(self, name: str, func: task.Orchestrator[Any, Any]) -> None:
         self._registry.add_named_orchestrator(name, func)
 
+    def _register_orchestrator_once(
+            self,
+            name: str,
+            func: task.Orchestrator[Any, Any],
+    ) -> None:
+        with self._registration_lock:
+            if self._registered_orchestrator_functions.get(name) is func:
+                return
+            if self._registry.get_orchestrator(name) is not None:
+                raise ValueError(f"A '{name}' orchestrator already exists.")
+            self.add_named_orchestrator(name, wrap_orchestrator(func))
+            self._registered_orchestrator_functions[name] = func
+
+    def _register_entity_once(self, func: task.Entity[Any, Any]) -> None:
+        name = task.get_entity_name(func).lower()
+        with self._registration_lock:
+            if self._registered_entity_functions.get(name) is func:
+                return
+            if self._registry.get_entity(name) is not None:
+                raise ValueError(f"A '{name}' entity already exists.")
+            self.add_entity(wrap_entity(func), name=name)
+            self._registered_entity_functions[name] = func
+
     def execute_orchestration_request(self, func: task.Orchestrator[Any, Any], context: Any) -> str:
         context_body = getattr(context, "body", None)
         if context_body is None:
@@ -70,7 +101,7 @@ def stub_complete(stub_response: OrchestratorResponse) -> None:
             raise ValueError("No ExecutionStarted event found in orchestration request.")
 
         function_name = execution_started_events[-1].executionStarted.name
-        self.add_named_orchestrator(function_name, wrap_orchestrator(func))
+        self._register_orchestrator_once(function_name, func)
         super()._execute_orchestrator(request, stub, None)
 
         if response is None:
@@ -94,7 +125,7 @@ def stub_complete(stub_response: EntityBatchResult) -> None:
             response = stub_response
         stub.CompleteEntityTask = stub_complete
 
-        self.add_entity(wrap_entity(func))
+        self._register_entity_once(func)
         super()._execute_entity_batch(request, stub, None)
 
         if response is None:

tests/azure-functions-durable/test_decorator_compat.py

@@ -4,6 +4,7 @@
 import azure.durable_functions as df
 import inspect
 import pytest
+from unittest.mock import MagicMock, patch
 from azure.durable_functions.constants import (
     ACTIVITY_TRIGGER,
     DURABLE_CLIENT,
@@ -104,6 +105,32 @@ def my_entity(context):
     assert trigger.entity_name == "MyEntity"
 
 
+def test_entity_trigger_reuses_worker_across_invocations():
+    app = df.DFApp()
+
+    def my_entity(context):
+        return None
+
+    with patch(
+        "azure.durable_functions.decorators.durable_app.DurableFunctionsWorker"
+    ) as worker_cls:
+        worker = worker_cls.return_value
+        worker.execute_entity_batch_request.return_value = "encoded"
+        fb = app.entity_trigger(context_name="context")(my_entity)
+        handle = fb._function._func
+        first_context = MagicMock()
+        second_context = MagicMock()
+        first_result = handle(first_context)
+        second_result = handle(second_context)
+
+    assert first_result == second_result == "encoded"
+    worker_cls.assert_called_once_with()
+    assert worker.execute_entity_batch_request.call_args_list == [
+        ((my_entity, first_context),),
+        ((my_entity, second_context),),
+    ]
+
+
 # ---------------------------------------------------------------------------
 # durable_client_input
 # ---------------------------------------------------------------------------

tests/azure-functions-durable/test_orchestrator_compat.py

@@ -5,7 +5,7 @@
 
 ``Orchestrator`` is the thin adapter registered with the Azure Functions host:
 its generated ``handle`` receives the host's transport context and delegates to
-a fresh :class:`DurableFunctionsWorker` per invocation.
+a shared :class:`DurableFunctionsWorker`.
 """
 
 from unittest.mock import MagicMock, patch
@@ -21,11 +21,12 @@ def user_orchestrator(context):
     with patch(
         "azure.durable_functions.orchestrator.DurableFunctionsWorker"
     ) as worker_cls:
-        worker_cls.return_value.execute_orchestration_request.return_value = "encoded"
+        worker = worker_cls.return_value
+        worker.execute_orchestration_request.return_value = "encoded"
         result = Orchestrator(user_orchestrator).handle(context)
 
     assert result == "encoded"
-    worker_cls.return_value.execute_orchestration_request.assert_called_once_with(
+    worker.execute_orchestration_request.assert_called_once_with(
         user_orchestrator, context)
 
 
@@ -34,8 +35,8 @@ def user_orchestrator(context):
         return None
 
     context = MagicMock()
-    orchestrator = Orchestrator(user_orchestrator)
     with patch("azure.durable_functions.orchestrator.DurableFunctionsWorker"):
+        orchestrator = Orchestrator(user_orchestrator)
         orchestrator.handle(context)
     assert orchestrator.durable_context is context
 
@@ -53,14 +54,20 @@ def test_created_handle_delegates_to_worker():
     def user_orchestrator(context):
         return None
 
-    handle = Orchestrator.create(user_orchestrator)
-    context = MagicMock()
     with patch(
         "azure.durable_functions.orchestrator.DurableFunctionsWorker"
     ) as worker_cls:
-        worker_cls.return_value.execute_orchestration_request.return_value = "encoded"
-        result = handle(context)
-
-    assert result == "encoded"
-    worker_cls.return_value.execute_orchestration_request.assert_called_once_with(
-        user_orchestrator, context)
+        worker = worker_cls.return_value
+        worker.execute_orchestration_request.return_value = "encoded"
+        handle = Orchestrator.create(user_orchestrator)
+        first_context = MagicMock()
+        second_context = MagicMock()
+        first_result = handle(first_context)
+        second_result = handle(second_context)
+
+    assert first_result == second_result == "encoded"
+    worker_cls.assert_called_once_with()
+    assert worker.execute_orchestration_request.call_args_list == [
+        ((user_orchestrator, first_context),),
+        ((user_orchestrator, second_context),),
+    ]

tests/azure-functions-durable/test_worker_compat.py

@@ -12,6 +12,7 @@
 
 import base64
 import json
+from concurrent.futures import ThreadPoolExecutor
 from types import SimpleNamespace
 
 import pytest
@@ -131,6 +132,40 @@ def orchestrator(context):
     assert "boom" in completion.failureDetails.errorMessage
 
 
+def test_execute_orchestration_request_supports_concurrent_reinvocation():
+    def orchestrator(context):
+        return context.instance_id
+
+    worker = DurableFunctionsWorker()
+    encoded = _encode_orchestrator_request("concurrent-orch")
+    with ThreadPoolExecutor(max_workers=4) as executor:
+        results = list(executor.map(
+            lambda _: worker.execute_orchestration_request(orchestrator, encoded),
+            range(8),
+        ))
+
+    assert all(
+        json.loads(_get_completion_action(
+            _decode_orchestrator_response(result)).result.value) == TEST_INSTANCE_ID
+        for result in results
+    )
+
+
+def test_execute_orchestration_request_rejects_different_function_with_same_name():
+    def first_orchestrator(context):
+        return "first"
+
+    def second_orchestrator(context):
+        return "second"
+
+    worker = DurableFunctionsWorker()
+    encoded = _encode_orchestrator_request("same-name")
+    worker.execute_orchestration_request(first_orchestrator, encoded)
+
+    with pytest.raises(ValueError, match="A 'same-name' orchestrator already exists"):
+        worker.execute_orchestration_request(second_orchestrator, encoded)
+
+
 # ---------------------------------------------------------------------------
 # execute_entity_batch_request
 # ---------------------------------------------------------------------------
@@ -197,3 +232,40 @@ def entity(context):
     response = _decode_entity_response(result)
     assert response.results[0].HasField("failure")
     assert "entity failed" in response.results[0].failure.failureDetails.errorMessage
+
+
+def test_execute_entity_batch_request_supports_concurrent_reinvocation():
+    def entity(context):
+        context.set_result("handled")
+
+    entity.__durable_entity_name__ = "Counter"
+    worker = DurableFunctionsWorker()
+    encoded = _encode_entity_batch_request("@counter@key1", "op")
+    with ThreadPoolExecutor(max_workers=4) as executor:
+        results = list(executor.map(
+            lambda _: worker.execute_entity_batch_request(entity, encoded),
+            range(8),
+        ))
+
+    assert all(
+        json.loads(_decode_entity_response(
+            result).results[0].success.result.value) == "handled"
+        for result in results
+    )
+
+
+def test_execute_entity_batch_request_rejects_different_function_with_same_name():
+    def first_entity(context):
+        context.set_result("first")
+
+    def second_entity(context):
+        context.set_result("second")
+
+    first_entity.__durable_entity_name__ = "Counter"
+    second_entity.__durable_entity_name__ = "Counter"
+    worker = DurableFunctionsWorker()
+    encoded = _encode_entity_batch_request("@counter@key1", "op")
+    worker.execute_entity_batch_request(first_entity, encoded)
+
+    with pytest.raises(ValueError, match="A 'counter' entity already exists"):
+        worker.execute_entity_batch_request(second_entity, encoded)