Merge branch 'main' into andystaples/abandon-failed-workitems

Author: berndverst

.github/workflows/durabletask-azuremanaged.yml

@@ -15,10 +15,10 @@ jobs:
     runs-on: ubuntu-latest
     steps:
     - uses: actions/checkout@v4
-    - name: Set up Python 3.13
+    - name: Set up Python 3.14
       uses: actions/setup-python@v5
       with:
-        python-version: 3.13
+        python-version: 3.14
     - name: Install dependencies
       working-directory: durabletask-azuremanaged
       run: |
@@ -36,7 +36,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        python-version: ["3.9", "3.10", "3.11", "3.12", "3.13"]
+        python-version: ["3.10", "3.11", "3.12", "3.13", "3.14"]
     env:
       EMULATOR_VERSION: "latest"
     needs: lint
@@ -100,7 +100,7 @@ jobs:
       - name: Set up Python
         uses: actions/setup-python@v5
         with:
-          python-version: "3.13"  # Adjust Python version as needed
+          python-version: "3.14"  # Adjust Python version as needed
 
       - name: Install dependencies
         run: |

.github/workflows/durabletask.yml

@@ -15,10 +15,10 @@ jobs:
     runs-on: ubuntu-latest
     steps:
     - uses: actions/checkout@v4
-    - name: Set up Python 3.13
+    - name: Set up Python 3.14
       uses: actions/setup-python@v5
       with:
-        python-version: 3.13
+        python-version: 3.14
     - name: Install dependencies
       run: |
         python -m pip install --upgrade pip
@@ -38,7 +38,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        python-version: ["3.9", "3.10", "3.11", "3.12", "3.13"]
+        python-version: ["3.10", "3.11", "3.12", "3.13", "3.14"]
     needs: lint-and-unit-tests
     runs-on: ubuntu-latest
     steps:
@@ -85,7 +85,7 @@ jobs:
       - name: Set up Python
         uses: actions/setup-python@v5
         with:
-          python-version: "3.13"  # Adjust Python version as needed
+          python-version: "3.14"  # Adjust Python version as needed
 
       - name: Install dependencies
         run: |

docs/features.md

@@ -148,4 +148,21 @@ Orchestrations can be suspended using the `suspend_orchestration` client API and
 
 ### Retry policies
 
-Orchestrations can specify retry policies for activities and sub-orchestrations. These policies control how many times and how frequently an activity or sub-orchestration will be retried in the event of a transient error.
+Orchestrations can specify retry policies for activities and sub-orchestrations. These policies control how many times and how frequently an activity or sub-orchestration will be retried in the event of a transient error.
+
+### Logging configuration
+
+Both the TaskHubGrpcWorker and TaskHubGrpcClient (as well as DurableTaskSchedulerWorker and DurableTaskSchedulerClient for durabletask-azuremanaged) accept a log_handler and log_formatter object from `logging`. These can be used to customize verbosity, output location, and format of logs emitted by these sources.
+
+For example, to output logs to a file called `worker.log` at level `DEBUG`, the following syntax might apply:
+
+```python
+log_handler = logging.FileHandler('durable.log', encoding='utf-8')
+log_handler.setLevel(logging.DEBUG)
+
+with DurableTaskSchedulerWorker(host_address=endpoint, secure_channel=secure_channel,
+                                taskhub=taskhub_name, token_credential=credential, log_handler=log_handler) as w:
+```
+
+**NOTE**
+The worker and client output many logs at the `DEBUG` level that will be useful when understanding orchestration flow and diagnosing issues with Durable applications. Before submitting issues, please attempt a repro of the issue with debug logging enabled.

durabletask-azuremanaged/durabletask/azuremanaged/client.py

@@ -1,6 +1,8 @@
 # Copyright (c) Microsoft Corporation.
 # Licensed under the MIT License.
 
+import logging
+
 from typing import Optional
 
 from azure.core.credentials import TokenCredential
@@ -18,7 +20,9 @@ def __init__(self, *,
                  taskhub: str,
                  token_credential: Optional[TokenCredential],
                  secure_channel: bool = True,
-                 default_version: Optional[str] = None):
+                 default_version: Optional[str] = None,
+                 log_handler: Optional[logging.Handler] = None,
+                 log_formatter: Optional[logging.Formatter] = None):
 
         if not taskhub:
             raise ValueError("Taskhub value cannot be empty. Please provide a value for your taskhub")
@@ -31,5 +35,7 @@ def __init__(self, *,
             host_address=host_address,
             secure_channel=secure_channel,
             metadata=None,
+            log_handler=log_handler,
+            log_formatter=log_formatter,
             interceptors=interceptors,
             default_version=default_version)

durabletask-azuremanaged/durabletask/azuremanaged/worker.py

@@ -1,6 +1,8 @@
 # Copyright (c) Microsoft Corporation.
 # Licensed under the MIT License.
 
+import logging
+
 from typing import Optional
 
 from azure.core.credentials import TokenCredential
@@ -28,6 +30,8 @@ class DurableTaskSchedulerWorker(TaskHubGrpcWorker):
         concurrency_options (Optional[ConcurrencyOptions], optional): Configuration
             for controlling worker concurrency limits. If None, default concurrency
             settings will be used.
+        log_handler (Optional[logging.Handler], optional): Custom logging handler for worker logs.
+        log_formatter (Optional[logging.Formatter], optional): Custom log formatter for worker logs.
 
     Raises:
         ValueError: If taskhub is empty or None.
@@ -52,12 +56,15 @@ class DurableTaskSchedulerWorker(TaskHubGrpcWorker):
         parameter is set to None since authentication is handled by the
         DTS interceptor.
     """
+
     def __init__(self, *,
                  host_address: str,
                  taskhub: str,
                  token_credential: Optional[TokenCredential],
                  secure_channel: bool = True,
-                 concurrency_options: Optional[ConcurrencyOptions] = None):
+                 concurrency_options: Optional[ConcurrencyOptions] = None,
+                 log_handler: Optional[logging.Handler] = None,
+                 log_formatter: Optional[logging.Formatter] = None):
 
         if not taskhub:
             raise ValueError("The taskhub value cannot be empty.")
@@ -70,5 +77,7 @@ def __init__(self, *,
             host_address=host_address,
             secure_channel=secure_channel,
             metadata=None,
+            log_handler=log_handler,
+            log_formatter=log_formatter,
             interceptors=interceptors,
             concurrency_options=concurrency_options)

durabletask-azuremanaged/pyproject.toml

@@ -22,7 +22,7 @@ classifiers = [
   "Programming Language :: Python :: 3",
   "License :: OSI Approved :: MIT License",
 ]
-requires-python = ">=3.9"
+requires-python = ">=3.10"
 license = {file = "LICENSE"}
 readme = "README.md"
 dependencies = [

durabletask/task.py

@@ -201,7 +201,7 @@ def lock_entities(self, entities: list[EntityInstanceId]) -> Task[EntityLock]:
         pass
 
     @abstractmethod
-    def call_sub_orchestrator(self, orchestrator: Orchestrator[TInput, TOutput], *,
+    def call_sub_orchestrator(self, orchestrator: Union[Orchestrator[TInput, TOutput], str], *,
                               input: Optional[TInput] = None,
                               instance_id: Optional[str] = None,
                               retry_policy: Optional[RetryPolicy] = None,

durabletask/worker.py

@@ -246,7 +246,7 @@ class TaskHubGrpcWorker:
             Defaults to the value from environment variables or localhost.
         metadata (Optional[list[tuple[str, str]]], optional): gRPC metadata to include with
             requests. Used for authentication and routing. Defaults to None.
-        log_handler (optional): Custom logging handler for worker logs. Defaults to None.
+        log_handler (optional[logging.Handler]): Custom logging handler for worker logs. Defaults to None.
         log_formatter (Optional[logging.Formatter], optional): Custom log formatter.
             Defaults to None.
         secure_channel (bool, optional): Whether to use a secure gRPC channel (TLS).
@@ -314,7 +314,7 @@ def __init__(
             *,
             host_address: Optional[str] = None,
             metadata: Optional[list[tuple[str, str]]] = None,
-            log_handler=None,
+            log_handler: Optional[logging.Handler] = None,
             log_formatter: Optional[logging.Formatter] = None,
             secure_channel: bool = False,
             interceptors: Optional[Sequence[shared.ClientInterceptor]] = None,
@@ -1072,15 +1072,18 @@ def lock_entities(self, entities: list[EntityInstanceId]) -> task.Task[EntityLoc
 
     def call_sub_orchestrator(
             self,
-            orchestrator: task.Orchestrator[TInput, TOutput],
+            orchestrator: Union[task.Orchestrator[TInput, TOutput], str],
             *,
             input: Optional[TInput] = None,
             instance_id: Optional[str] = None,
             retry_policy: Optional[task.RetryPolicy] = None,
             version: Optional[str] = None,
     ) -> task.Task[TOutput]:
         id = self.next_sequence_number()
-        orchestrator_name = task.get_name(orchestrator)
+        if isinstance(orchestrator, str):
+            orchestrator_name = orchestrator
+        else:
+            orchestrator_name = task.get_name(orchestrator)
         default_version = self._registry.versioning.default_version if self._registry.versioning else None
         orchestrator_version = version if version else default_version
         self.call_activity_function_helper(
@@ -1276,13 +1279,21 @@ def execute(
             old_events: Sequence[pb.HistoryEvent],
             new_events: Sequence[pb.HistoryEvent],
     ) -> ExecutionResults:
+        orchestration_name = "<unknown>"
+        orchestration_started_events = [e for e in old_events if e.HasField("executionStarted")]
+        if len(orchestration_started_events) >= 1:
+            orchestration_name = orchestration_started_events[0].executionStarted.name
+
+        self._logger.debug(
+            f"{instance_id}: Beginning replay for orchestrator {orchestration_name}..."
+        )
+
         self._entity_state = OrchestrationEntityContext(instance_id)
 
         if not new_events:
             raise task.OrchestrationStateError(
                 "The new history event list must have at least one event in it."
             )
-
         ctx = _RuntimeOrchestrationContext(instance_id, self._registry, self._entity_state)
         try:
             # Rebuild local state by replaying old history into the orchestrator function
@@ -1314,13 +1325,15 @@ def execute(
 
         except Exception as ex:
             # Unhandled exceptions fail the orchestration
+            self._logger.debug(f"{instance_id}: Orchestration {orchestration_name} failed")
             ctx.set_failed(ex)
 
         if not ctx._is_complete:
             task_count = len(ctx._pending_tasks)
             event_count = len(ctx._pending_events)
             self._logger.info(
-                f"{instance_id}: Orchestrator yielded with {task_count} task(s) and {event_count} event(s) outstanding."
+                f"{instance_id}: Orchestrator {orchestration_name} yielded with {task_count} task(s) "
+                f"and {event_count} event(s) outstanding."
             )
         elif (
                 ctx._completion_status and ctx._completion_status is not pb.ORCHESTRATION_STATUS_CONTINUED_AS_NEW
@@ -1329,7 +1342,7 @@ def execute(
                 ctx._completion_status
             )
             self._logger.info(
-                f"{instance_id}: Orchestration completed with status: {completion_status_str}"
+                f"{instance_id}: Orchestration {orchestration_name} completed with status: {completion_status_str}"
             )
 
         actions = ctx.get_actions()

pyproject.toml

@@ -21,7 +21,7 @@ classifiers = [
   "Programming Language :: Python :: 3",
   "License :: OSI Approved :: MIT License",
 ]
-requires-python = ">=3.9"
+requires-python = ">=3.10"
 license = {file = "LICENSE"}
 readme = "README.md"
 dependencies = [

tests/durabletask-azuremanaged/test_dts_orchestration_e2e.py

@@ -175,6 +175,34 @@ def parent_orchestrator(ctx: task.OrchestrationContext, count: int):
     assert activity_counter == 30
 
 
+def test_sub_orchestrator_by_name():
+    sub_orchestrator_counter = 0
+
+    def orchestrator_child(ctx: task.OrchestrationContext, _):
+        nonlocal sub_orchestrator_counter
+        sub_orchestrator_counter += 1
+
+    def parent_orchestrator(ctx: task.OrchestrationContext, _):
+        yield ctx.call_sub_orchestrator("orchestrator_child")
+
+    # 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(orchestrator_child)
+        w.add_orchestrator(parent_orchestrator)
+        w.start()
+
+        task_hub_client = DurableTaskSchedulerClient(host_address=endpoint, secure_channel=True,
+                                                     taskhub=taskhub_name, token_credential=None)
+        id = task_hub_client.schedule_new_orchestration(parent_orchestrator, input=None)
+        state = task_hub_client.wait_for_orchestration_completion(id, timeout=30)
+
+    assert state is not None
+    assert state.runtime_status == client.OrchestrationStatus.COMPLETED
+    assert state.failure_details is None
+    assert sub_orchestrator_counter == 1
+
+
 def test_wait_for_multiple_external_events():
     def orchestrator(ctx: task.OrchestrationContext, _):
         a = yield ctx.wait_for_external_event('A')
@@ -267,7 +295,7 @@ def orchestrator(ctx: task.OrchestrationContext, _):
 #         try:
 #             state = task_hub_client.wait_for_orchestration_completion(id, timeout=3)
 #             assert False, "Orchestration should not have completed"
-#         except TimeoutError:
+#         except (TimeoutError, _InactiveRpcError):
 #             pass
 
 #         # Resume the orchestration and wait for it to complete