Conformance finding: Python-authored polyglot workflow stays leased under current sample smoke (#119)

Author: durable-workflow-ops

CHANGELOG.md

@@ -7,6 +7,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ## [Unreleased]
 
 ### Fixed
+- Workflow workers now report unhandled workflow-task execution errors back to
+  the server instead of leaving the leased task pending until the lease or CLI
+  wait times out. This lets the server observe and retry or fail the task
+  promptly when command serialization or interceptor code raises after a task
+  has been claimed.
 - Workflow-task completion now retries transient transport failures and server
   throttling/5xx rejections before preserving emitted commands or reporting a
   definite task failure, reducing stuck waiting runs when a signal-satisfied

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "durable-workflow"
-version = "0.4.54"
+version = "0.4.55"
 description = "Python SDK for the Durable Workflow server (language-neutral HTTP protocol)"
 readme = "README.md"
 requires-python = ">=3.10"

src/durable_workflow/worker.py

@@ -1442,13 +1442,49 @@ async def _dispatch_workflow_task(self, task: dict[str, Any]) -> None:
         try:
             commands = await self._run_workflow_task(task)
             outcome = "completed" if commands is not None else "failed"
-        except Exception:
+        except Exception as exc:
             log.exception("unhandled error in workflow task execution")
+            await self._report_unhandled_workflow_task_error(task, exc)
+            outcome = "failed"
         finally:
             self._workflow_inflight = max(0, self._workflow_inflight - 1)
             self._record_task_metrics("workflow", outcome, time.perf_counter() - task_start)
             self._wf_semaphore.release()
 
+    async def _report_unhandled_workflow_task_error(
+        self,
+        task: dict[str, Any],
+        error: Exception,
+    ) -> None:
+        task_id = _string_or_none(task.get("task_id"))
+        if task_id is None:
+            return
+
+        attempt = task.get("workflow_task_attempt", 1)
+        if not isinstance(attempt, int) or attempt < 1:
+            attempt = 1
+
+        try:
+            await self.client.fail_workflow_task(
+                task_id=task_id,
+                lease_owner=self.worker_id,
+                workflow_task_attempt=attempt,
+                message=(
+                    "unhandled workflow task execution error: "
+                    f"{str(error) or type(error).__name__}"
+                ),
+                failure_type=type(error).__name__,
+                stack_trace="".join(
+                    traceback.format_exception(type(error), error, error.__traceback__)
+                ),
+            )
+        except Exception as report_error:
+            log.warning(
+                "failed to report unhandled workflow task error for %s: %s",
+                task_id,
+                report_error,
+            )
+
     async def _poll_activity_tasks(self) -> None:
         while not self._stop.is_set():
             await self._act_semaphore.acquire()

tests/test_worker.py

@@ -55,6 +55,31 @@ def run(self, ctx):  # type: ignore[no-untyped-def]
         ]
 
 
+@workflow.defn(name="two-cross-queue-wf")
+class TwoCrossQueueWorkflow:
+    def run(self, ctx, request):  # type: ignore[no-untyped-def]
+        marker = yield ctx.schedule_activity(
+            "external.marker",
+            [request],
+            queue="external-queue",
+        )
+        description = yield ctx.schedule_activity(
+            "external.describe",
+            [marker],
+            queue="external-queue",
+        )
+        return {
+            "marker": marker,
+            "description": description,
+        }
+
+
+@workflow.defn(name="unserializable-result-wf")
+class UnserializableResultWorkflow:
+    def run(self, ctx):  # type: ignore[no-untyped-def]
+        return object()
+
+
 @workflow.defn(name="update-wf")
 class UpdateWorkflow:
     def __init__(self) -> None:
@@ -780,6 +805,124 @@ async def test_complete_on_resolved_activity(self, mock_client: AsyncMock) -> No
         assert commands[0]["result"]["codec"] == "json"
         assert serializer.decode(commands[0]["result"]["blob"], codec="json") == "done"
 
+    @pytest.mark.asyncio
+    async def test_cross_queue_second_activity_uses_completed_first_result(
+        self, mock_client: AsyncMock
+    ) -> None:
+        worker = Worker(mock_client, task_queue="workflow-queue", workflows=[TwoCrossQueueWorkflow], activities=[])
+        marker = {"runtime": "external", "name": "Grace", "message": "hello"}
+        task = {
+            "task_id": "t-cross-queue-second",
+            "workflow_type": "two-cross-queue-wf",
+            "workflow_task_attempt": 2,
+            "history_events": [
+                {
+                    "event_type": "ActivityCompleted",
+                    "payload": {
+                        "sequence": 1,
+                        "activity_type": "external.marker",
+                        "payload_codec": "json",
+                        "result": serializer.envelope(marker, codec="json"),
+                    },
+                },
+            ],
+            "arguments": serializer.encode([{"name": "Grace"}], codec="json"),
+            "payload_codec": "json",
+        }
+
+        await worker._run_workflow_task(task)
+
+        mock_client.complete_workflow_task.assert_called_once()
+        commands = mock_client.complete_workflow_task.call_args.kwargs["commands"]
+        assert commands == [
+            {
+                "type": "schedule_activity",
+                "activity_type": "external.describe",
+                "queue": "external-queue",
+                "arguments": commands[0]["arguments"],
+            }
+        ]
+        assert commands[0]["arguments"]["codec"] == "json"
+        assert serializer.decode(commands[0]["arguments"]["blob"], codec="json") == [marker]
+
+    @pytest.mark.asyncio
+    async def test_cross_queue_workflow_completes_after_second_activity(
+        self, mock_client: AsyncMock
+    ) -> None:
+        worker = Worker(mock_client, task_queue="workflow-queue", workflows=[TwoCrossQueueWorkflow], activities=[])
+        marker = {"runtime": "external", "name": "Grace", "message": "hello"}
+        description = {"runtime": "external", "description": "Grace handled by external activity"}
+        task = {
+            "task_id": "t-cross-queue-complete",
+            "workflow_type": "two-cross-queue-wf",
+            "workflow_task_attempt": 3,
+            "history_events": [
+                {
+                    "event_type": "ActivityCompleted",
+                    "payload": {
+                        "sequence": 1,
+                        "activity_type": "external.marker",
+                        "payload_codec": "json",
+                        "result": serializer.envelope(marker, codec="json"),
+                    },
+                },
+                {
+                    "event_type": "ActivityCompleted",
+                    "payload": {
+                        "sequence": 2,
+                        "activity_type": "external.describe",
+                        "payload_codec": "json",
+                        "result": serializer.envelope(description, codec="json"),
+                    },
+                },
+            ],
+            "arguments": serializer.encode([{"name": "Grace"}], codec="json"),
+            "payload_codec": "json",
+        }
+
+        await worker._run_workflow_task(task)
+
+        mock_client.complete_workflow_task.assert_called_once()
+        commands = mock_client.complete_workflow_task.call_args.kwargs["commands"]
+        assert commands[0]["type"] == "complete_workflow"
+        assert commands[0]["result"]["codec"] == "json"
+        assert serializer.decode(commands[0]["result"]["blob"], codec="json") == {
+            "marker": marker,
+            "description": description,
+        }
+
+    @pytest.mark.asyncio
+    async def test_dispatch_reports_unhandled_workflow_task_error(
+        self, mock_client: AsyncMock
+    ) -> None:
+        worker = Worker(
+            mock_client,
+            task_queue="q1",
+            workflows=[UnserializableResultWorkflow],
+            activities=[],
+            worker_id="w-unserializable",
+        )
+        task = {
+            "task_id": "t-unserializable",
+            "workflow_type": "unserializable-result-wf",
+            "workflow_task_attempt": 4,
+            "history_events": [],
+            "arguments": "[]",
+            "payload_codec": "json",
+        }
+
+        await worker._dispatch_workflow_task(task)
+
+        mock_client.complete_workflow_task.assert_not_called()
+        mock_client.fail_workflow_task.assert_awaited_once()
+        call_kwargs = mock_client.fail_workflow_task.await_args.kwargs
+        assert call_kwargs["task_id"] == "t-unserializable"
+        assert call_kwargs["workflow_task_attempt"] == 4
+        assert call_kwargs["lease_owner"] == worker.worker_id
+        assert call_kwargs["failure_type"] == "TypeError"
+        assert "unhandled workflow task execution error" in call_kwargs["message"]
+        assert "Object of type object" in call_kwargs["message"]
+
     @pytest.mark.asyncio
     async def test_unknown_workflow_type_fails_task(self, mock_client: AsyncMock) -> None:
         worker = Worker(mock_client, task_queue="q1", workflows=[], activities=[])