diff --git a/CHANGELOG.md b/CHANGELOG.md index 117c33b8..ea33aa3e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,17 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +## [2.12.3] - 2026-11-02 + +### Fixed + +- Report exceptions with empty or whitespace-only messages using meaningful + fallback text, so workflow and activity failures reach callers instead of + being rejected as invalid failure reports. Preserve the exception type, + stack trace and task identity in ordinary and supervised activity workers. +- Apply the same blank-message handling to invocable activities, queries and + update validators while keeping nonblank diagnostic text unchanged. + ## [2.12.1] - 2026-10-29 ### Fixed diff --git a/pyproject.toml b/pyproject.toml index fd382a05..21776d72 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "durable-workflow" -version = "2.12.2" +version = "2.12.3" description = "Python client and worker SDK for Durable Workflow Cloud and self-hosted Server" readme = "README.md" requires-python = ">=3.10" @@ -79,8 +79,8 @@ durable-workflow-replay-conformance = "durable_workflow.replay_conformance:main" durable-workflow-workflow-updates-conformance = "durable_workflow.workflow_updates_conformance:main" [tool.durable-workflow] -product-train = "2.12.2" -registry-version = "2.12.2" +product-train = "2.12.3" +registry-version = "2.12.3" supported-server-versions = "2.8.0" worker-protocol-version = "1.20" control-plane-version = "2" diff --git a/src/durable_workflow/_activity_process.py b/src/durable_workflow/_activity_process.py index b39ee479..1df0b9fe 100644 --- a/src/durable_workflow/_activity_process.py +++ b/src/durable_workflow/_activity_process.py @@ -24,6 +24,7 @@ from pathlib import Path from typing import Any, cast +from ._failure_reporting import exception_message from .activity import ActivityContext, ActivityInfo, _set_context from .errors import ActivityCancelled, NonRetryableError from .interceptors import ActivityInterceptorContext, WorkerInterceptor @@ -139,7 +140,7 @@ async def intercepted( except BaseException as error: code = getattr(error, "code", None) _write_payload(directory, "outcome", CallbackOutcome(failure=CallbackFailure( - message=str(error), failure_type=type(error).__name__, + message=exception_message(error), failure_type=type(error).__name__, failure_class=f"{type(error).__module__}.{type(error).__qualname__}", failure_code=code if isinstance(code, int) and not isinstance(code, bool) else None, stack_trace=traceback.format_exc(), non_retryable=isinstance(error, NonRetryableError), diff --git a/src/durable_workflow/_failure_reporting.py b/src/durable_workflow/_failure_reporting.py index ae663042..e4f81580 100644 --- a/src/durable_workflow/_failure_reporting.py +++ b/src/durable_workflow/_failure_reporting.py @@ -26,6 +26,12 @@ }) +def exception_message(error: BaseException, *, fallback: str | None = None) -> str: + """Keep useful diagnostic text; blank exceptions still need a valid report.""" + message = str(error) + return message if message.strip() else fallback or type(error).__name__ + + def encoded_json(value: Any) -> bytes: """Use HTTPX's actual JSON settings, including escaping and NaN refusal.""" return httpx.Request("POST", "https://failure.invalid", json=value).content diff --git a/src/durable_workflow/invocable.py b/src/durable_workflow/invocable.py index 9ee9e289..7706665d 100644 --- a/src/durable_workflow/invocable.py +++ b/src/durable_workflow/invocable.py @@ -12,6 +12,7 @@ from typing import Any, cast from . import serializer +from ._failure_reporting import exception_message from .errors import ActivityCancelled, NonRetryableError from .external_storage import ExternalPayloadCache, ExternalStorageDriver from .external_task_input import ExternalTaskInput, parse_external_task_input @@ -105,7 +106,7 @@ async def handle(self, envelope: Mapping[str, Any]) -> dict[str, Any]: started, kind="application", classification="application_error", - message=str(exc), + message=exception_message(exc), failure_type=type(exc).__name__, stack_trace=traceback.format_exc(), retryable=False, @@ -116,7 +117,7 @@ async def handle(self, envelope: Mapping[str, Any]) -> dict[str, Any]: started, kind="timeout", classification="deadline_exceeded", - message=str(exc) or "Invocable activity handler exceeded its deadline.", + message=exception_message(exc, fallback="Invocable activity handler exceeded its deadline."), failure_type=type(exc).__name__, stack_trace=traceback.format_exc(), timeout_type="deadline_exceeded", @@ -140,7 +141,7 @@ async def handle(self, envelope: Mapping[str, Any]) -> dict[str, Any]: started, kind="application", classification="application_error", - message=str(exc), + message=exception_message(exc), failure_type=type(exc).__name__, stack_trace=traceback.format_exc(), retryable=True, diff --git a/src/durable_workflow/worker.py b/src/durable_workflow/worker.py index 10d42940..fd13a74c 100644 --- a/src/durable_workflow/worker.py +++ b/src/durable_workflow/worker.py @@ -44,7 +44,7 @@ from ._activity_process import CallbackFailure, CallbackInvocation, CallbackProcessLost, SupervisedCallback from ._cancellation_scope_history import CancellationScopeBudget, CancellationScopeDeliveryReceipt from ._cooperative_cancellation import CancellationRequest, read_cancellation_history -from ._failure_reporting import REQUEST_BYTES, normalize_local_failure +from ._failure_reporting import REQUEST_BYTES, exception_message, normalize_local_failure from ._prepared_local_activity import PreparedAttempt, PreparedCancellationObserved, PreparedLocalRunner from ._query_task_authority import QueryTaskAuthority, current_query_authority from ._sticky_workflow_cache import CacheKey, StickyWorkflowCache, complete_history @@ -3240,7 +3240,7 @@ async def _run_activity_task_body(self, task: dict[str, Any]) -> str: except NonRetryableError as e: log.exception("activity failed (non-retryable)") await self._report_activity_task_failure( - task_id, attempt_id, "activity failure", message=str(e), failure_type=type(e).__name__, + task_id, attempt_id, "activity failure", message=exception_message(e), failure_type=type(e).__name__, failure_class=_exception_class_name(e), failure_code=_exception_code(e), stack_trace=traceback.format_exc(), non_retryable=True, ) @@ -3271,7 +3271,8 @@ async def _handle_activity_exception(self, task_id: str, attempt_id: str, error: raise error log.error("activity failed", exc_info=(type(error), error, error.__traceback__)) await self._report_activity_task_failure( - task_id, attempt_id, "activity failure", message=str(error), failure_type=type(error).__name__, + task_id, attempt_id, "activity failure", message=exception_message(error), + failure_type=type(error).__name__, failure_class=_exception_class_name(error), failure_code=_exception_code(error), stack_trace="".join(traceback.format_exception(type(error), error, error.__traceback__)), ) @@ -3573,7 +3574,7 @@ async def execute() -> Any: ) return "failed" except QueryFailed as e: - message = str(e) + message = exception_message(e) if "unknown query" in message: reason = "rejected_unknown_query" elif message.startswith("workflow replay failed before query:"): @@ -3594,7 +3595,7 @@ async def execute() -> Any: await self._fail_query_task( query_task_id, attempt, - str(e) or "query execution failed", + exception_message(e, fallback="query execution failed"), reason="query_rejected", failure_type=type(e).__name__, stack_trace=traceback.format_exc(), @@ -3631,7 +3632,7 @@ async def execute() -> Any: await self._fail_query_task( query_task_id, attempt, - str(e) or "query result completion was rejected by the server", + exception_message(e, fallback="query result completion was rejected by the server"), reason=server_reason if server_reason else "query_result_completion_failed", failure_type=type(e).__name__, stack_trace=traceback.format_exc(), @@ -3667,7 +3668,7 @@ async def execute() -> Any: await self._fail_query_task( query_task_id, attempt, - str(e) or "query result completion failed", + exception_message(e, fallback="query result completion failed"), reason="query_result_completion_failed", failure_type=type(e).__name__, stack_trace=traceback.format_exc(), @@ -4075,7 +4076,7 @@ async def _run_update_validation_task(self, task: dict[str, Any]) -> str: return await self._reject_update_validation_task( task_id, attempt, - str(exc) or "update validator rejected the request", + exception_message(exc, fallback="update validator rejected the request"), reason="update_validator_rejected", failure_type=exc.exception_type or type(exc).__name__, validation_errors=exc.validation_errors, @@ -4085,7 +4086,7 @@ async def _run_update_validation_task(self, task: dict[str, Any]) -> str: return await self._reject_update_validation_task( task_id, attempt, - str(exc), + exception_message(exc), reason="update_validation_worker_failed", failure_type=type(exc).__name__, stack_trace=traceback.format_exc(), @@ -4094,7 +4095,7 @@ async def _run_update_validation_task(self, task: dict[str, Any]) -> str: return await self._reject_update_validation_task( task_id, attempt, - str(exc) or "update validator rejected the request", + exception_message(exc, fallback="update validator rejected the request"), reason="update_validator_rejected", failure_type=type(exc).__name__, stack_trace=traceback.format_exc(), diff --git a/src/durable_workflow/workflow.py b/src/durable_workflow/workflow.py index 1c90bdd0..5dbc80c2 100644 --- a/src/durable_workflow/workflow.py +++ b/src/durable_workflow/workflow.py @@ -43,6 +43,7 @@ ScopeBoundary, ) from ._cooperative_cancellation import CancellationDelivery, read_cancellation_history +from ._failure_reporting import exception_message from .cancellation import ( CancellationContext, CancellationPolicy, @@ -3292,7 +3293,7 @@ def invoke() -> Any: raise except Exception as exc: state.outcome.close() - raise QueryFailed(str(exc) or f"query {query_name!r} failed") from exc + raise QueryFailed(exception_message(exc, fallback=f"query {query_name!r} failed")) from exc except BaseException: state.outcome.close() raise @@ -3594,7 +3595,7 @@ def _invoke_update_validator( except Exception as exc: errors = exc.errors if isinstance(exc, InvalidArgument) else None raise UpdateRejected( - str(exc) or f"update validator {update_name!r} rejected the request", + exception_message(exc, fallback=f"update validator {update_name!r} rejected the request"), exception_type=type(exc).__name__, validation_errors=errors, ) from exc @@ -3614,7 +3615,7 @@ def _accepted_update_payload( def _fail_update_from_exception(update_id: str, prefix: str, exc: Exception) -> FailUpdate: - message = str(exc) or type(exc).__name__ + message = exception_message(exc) return FailUpdate( update_id=update_id, message=f"{prefix}: {message}", @@ -3628,7 +3629,7 @@ def _exception_class_name(exc: BaseException) -> str: def _fail_workflow_from_exception(exc: BaseException, *, prefix: str | None = None) -> FailWorkflow: - message = str(exc) or type(exc).__name__ + message = exception_message(exc) if prefix: message = f"{prefix}: {message}" @@ -3645,12 +3646,12 @@ def _fail_workflow_from_exception(exc: BaseException, *, prefix: str | None = No "initiating_failure": { "type": type(initiating).__name__, "class": _exception_class_name(initiating), - "message": str(initiating) or type(initiating).__name__, + "message": exception_message(initiating), }, "compensation_failure": { "type": type(compensation).__name__, "class": _exception_class_name(compensation), - "message": str(compensation) or type(compensation).__name__, + "message": exception_message(compensation), "activity_type": exc.compensation_activity_type, "registration_order": exc.compensation_registration_order, }, diff --git a/tests/fixtures/replay_regressions/blank-workflow-failure-message.json b/tests/fixtures/replay_regressions/blank-workflow-failure-message.json new file mode 100644 index 00000000..7a5da8c4 --- /dev/null +++ b/tests/fixtures/replay_regressions/blank-workflow-failure-message.json @@ -0,0 +1,15 @@ +{ + "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json", + "fixture_schema": "durable-workflow.replay-regression/v1", + "id": "blank-workflow-failure-message", + "protocol_version": "1.20", + "bindings": ["python"], + "workflow": {"type": "tests.replay.blank-failure", "input": [" \t\n"], "payload_codec": "avro"}, + "history": [{"event_type": "WorkflowStarted", "payload": {}}], + "expected": { + "type": "fail_workflow", + "message": "AssertionError", + "exception_type": "AssertionError", + "exception_class": "builtins.AssertionError" + } +} diff --git a/tests/integration/test_empty_callback_failure.py b/tests/integration/test_empty_callback_failure.py new file mode 100644 index 00000000..911749d1 --- /dev/null +++ b/tests/integration/test_empty_callback_failure.py @@ -0,0 +1,54 @@ +from __future__ import annotations + +import asyncio +import json +import uuid +from typing import Any + +import httpx +import pytest + +from durable_workflow import Client, Worker, activity +from durable_workflow.errors import WorkflowFailed +from tests.worker_observation import stop_and_drain + + +@activity.defn(name="tests.empty-failure") +async def empty_failure(message: str) -> None: + raise AssertionError(message) + + +@pytest.mark.parametrize("supervised", [False, True]) +@pytest.mark.parametrize("message", ["", " \t\n"]) +async def test_blank_callback_failure_reaches_a_terminal_caller( + server_url: str, server_token: str, supervised: bool, message: str, +) -> None: + identity = "empty-failure-" + uuid.uuid4().hex[:12] + replies: list[tuple[int, dict[str, Any], dict[str, Any]]] = [] + async with Client(server_url, worker_token=server_token, control_token=server_token) as client: + async def record(response: httpx.Response) -> None: + if response.request.url.path.endswith("/fail"): + await response.aread() + replies.append((response.status_code, json.loads(response.request.content), response.json())) + + client._http.event_hooks["response"].append(record) + worker = Worker(client, task_queue=identity, worker_id=identity, activities=[empty_failure], + capabilities=["cooperative_cancellation"] if supervised else []) + runner = asyncio.create_task(worker.run()) + try: + handle = await client.start_activity(activity_type="tests.empty-failure", task_queue=identity, + activity_id=identity, input=[message], retry_policy={"max_attempts": 1}) + with pytest.raises(WorkflowFailed): + await handle.result(timeout=15, poll_interval=.05) + assert (await handle.describe()).status == "failed" + assert len(replies) == 1 + status, submitted, accepted = replies[0] + assert status == 200 + assert accepted["recorded"] is True + assert submitted["lease_owner"] == identity + assert submitted["activity_attempt_id"] + assert submitted["failure"]["message"] == "AssertionError" + assert submitted["failure"]["type"] == "AssertionError" + assert "empty_failure" in submitted["failure"]["stack_trace"] + finally: + await stop_and_drain([worker], [runner]) diff --git a/tests/test_empty_callback_failure.py b/tests/test_empty_callback_failure.py new file mode 100644 index 00000000..ee70ea37 --- /dev/null +++ b/tests/test_empty_callback_failure.py @@ -0,0 +1,156 @@ +from __future__ import annotations + +from typing import Any +from unittest.mock import AsyncMock + +import pytest + +from durable_workflow import serializer, workflow +from durable_workflow.client import Client +from durable_workflow.errors import NonRetryableError +from durable_workflow.external_task_result import parse_external_task_result +from durable_workflow.invocable import handle_invocable_activity +from durable_workflow.worker import Worker +from durable_workflow.workflow import _fail_update_from_exception, _fail_workflow_from_exception +from tests.query_task_helpers import leased_query_task +from tests.test_cooperative_remote_worker import make_owner, task +from tests.test_invocable import activity_input +from tests.test_worker import mock_client # noqa: F401 (shared fixture) + + +async def application_failure(kind: str, message: str) -> None: + if kind == "NonRetryableError": + raise NonRetryableError(message) + raise AssertionError(message) + + +@pytest.mark.parametrize("message", ["", " \t\n", " original failure "]) +@pytest.mark.parametrize("kind", ["AssertionError", "NonRetryableError"]) +@pytest.mark.parametrize("supervised", [False, True]) +async def test_callback_failure_reports_a_nonempty_message_with_original_authority( + monkeypatch: pytest.MonkeyPatch, message: str, kind: str, supervised: bool, +) -> None: + if supervised: + worker, client = await make_owner(monkeypatch) + else: + client = AsyncMock(spec=Client) + worker = Worker(client, task_queue="queue", worker_id="owner") + worker.activities["remote"] = application_failure + try: + outcome = await worker._run_activity_task(task(kind, message)) + assert outcome == ("failed_non_retryable" if kind == "NonRetryableError" else "failed") + client.fail_activity_task.assert_awaited_once() + report = client.fail_activity_task.await_args.kwargs + assert report["message"] == (message if message.strip() else kind) + assert report["failure_type"] == kind + assert report["failure_class"].endswith("." + kind) + assert report.get("non_retryable", False) is (kind == "NonRetryableError") + assert report["task_id"] == "remote-task" + assert report["activity_attempt_id"] == "attempt" + assert report["lease_owner"] == "owner" + assert "application_failure" in report["stack_trace"] + client.complete_activity_task.assert_not_awaited() + finally: + await worker.stop() + + +@pytest.mark.parametrize("message", ["", " \t\n", " original failure "]) +@pytest.mark.parametrize("kind", ["AssertionError", "NonRetryableError"]) +async def test_invocable_failure_keeps_a_valid_typed_envelope(message: str, kind: str) -> None: + output = await handle_invocable_activity( + activity_input(kind, message), {"billing.charge-card": application_failure}, + ) + failure: Any = parse_external_task_result(output).failure + assert failure.message == (message if message.strip() else kind) + assert failure.type == kind + assert "application_failure" in failure.stack_trace + + +@pytest.mark.parametrize("message", ["", " \t\n", " original failure "]) +def test_workflow_and_update_failures_have_useful_messages(message: str) -> None: + error = AssertionError(message) + expected = message if message.strip() else "AssertionError" + failed = _fail_workflow_from_exception(error) + assert failed.message == expected + assert failed.exception_type == "AssertionError" + update = _fail_update_from_exception("update", "update rejected", error) + assert update.message == "update rejected: " + expected + assert update.exception_type == "AssertionError" + + +@workflow.defn(name="tests.blank-query-and-validator") +class BlankQueryAndValidatorWorkflow: + message = "" + + @workflow.query("status") + def status(self) -> str: + raise AssertionError(type(self).message) + + @workflow.update("approve") + def approve(self) -> None: + return None + + @approve.validator # type: ignore[attr-defined] + def validate_approve(self) -> None: + raise AssertionError(type(self).message) + + def run(self, ctx: Any) -> str: + return "waiting" + + +def blank_workflow_task(**fields: Any) -> dict[str, Any]: + return { + "workflow_type": "tests.blank-query-and-validator", + "history_events": [], + "workflow_arguments": serializer.envelope([], codec="avro"), + "payload_codec": "avro", + **fields, + } + + +@pytest.mark.parametrize("message", ["", " \t\n", " original failure "]) +async def test_query_failure_reports_a_nonempty_message( + mock_client: AsyncMock, message: str, monkeypatch: pytest.MonkeyPatch, # noqa: F811 +) -> None: + monkeypatch.setattr(BlankQueryAndValidatorWorkflow, "message", message) + worker = Worker(mock_client, task_queue="q1", workflows=[BlankQueryAndValidatorWorkflow]) + query = blank_workflow_task( + query_task_id="qt-blank", query_task_attempt=3, query_name="status", + query_arguments=serializer.envelope([], codec="avro"), + ) + + assert await worker._run_query_task(leased_query_task(worker, query)) == "failed" + + report = mock_client.fail_query_task.await_args.kwargs + assert report["message"] == (message if message.strip() else "query 'status' failed") + assert report["reason"] == "query_rejected" + assert report["query_task_id"] == "qt-blank" + assert report["query_task_attempt"] == 3 + mock_client.complete_query_task.assert_not_awaited() + + +@pytest.mark.parametrize("message", ["", " \t\n", " original failure "]) +async def test_update_validator_rejection_reports_a_nonempty_message( + mock_client: AsyncMock, message: str, monkeypatch: pytest.MonkeyPatch, # noqa: F811 +) -> None: + monkeypatch.setattr(BlankQueryAndValidatorWorkflow, "message", message) + worker = Worker( + mock_client, task_queue="q1", workflows=[BlankQueryAndValidatorWorkflow], worker_id="validator", + ) + validation = blank_workflow_task( + task_kind="update_validation", update_validation_task_id="uv-blank", update_validation_attempt=2, + update_name="approve", update_arguments=serializer.envelope([], codec="avro"), + workflow_id="wf", run_id="run", + ) + + assert await worker._run_update_validation_task(validation) == "rejected" + + rejection = mock_client.reject_update_validation_task.await_args.kwargs + expected = message if message.strip() else "update validator 'approve' rejected the request" + assert expected in rejection["message"] + assert rejection["message"].strip() + assert rejection["reason"] == "update_validator_rejected" + assert rejection["failure_type"] == "AssertionError" + assert rejection["update_validation_task_id"] == "uv-blank" + assert rejection["update_validation_attempt"] == 2 + mock_client.approve_update_validation_task.assert_not_awaited() diff --git a/tests/test_replay_regression_corpus.py b/tests/test_replay_regression_corpus.py index 52119b86..37857750 100644 --- a/tests/test_replay_regression_corpus.py +++ b/tests/test_replay_regression_corpus.py @@ -344,8 +344,15 @@ def run(self, ctx: WorkflowContext, cleanup_seconds: int = 0): # type: ignore[n return error.request_id +@workflow.defn(name="tests.replay.blank-failure") +class BlankFailureWorkflow: + def run(self, ctx: WorkflowContext, message: str) -> None: + raise AssertionError(message) + + WORKFLOWS = [ PendingFinally, + BlankFailureWorkflow, IdentityAcrossReplay, IdentityContinueAsNew, PersistedClockWorkflow,