Skip to content

Commit

Permalink
Fix issue when failing to convert failure (#727)
Browse files Browse the repository at this point in the history
Fixes #685
  • Loading branch information
cretz authored Jan 21, 2025
1 parent fe46e1a commit 154aab9
Show file tree
Hide file tree
Showing 3 changed files with 84 additions and 0 deletions.
1 change: 1 addition & 0 deletions temporalio/worker/_activity.py
Original file line number Diff line number Diff line change
Expand Up @@ -508,6 +508,7 @@ async def _run_activity(
completion.result.failed.failure.message = (
f"Failed building exception result: {inner_err}"
)
completion.result.failed.failure.application_failure_info.SetInParent()

# Do final completion
try:
Expand Down
1 change: 1 addition & 0 deletions temporalio/worker/_workflow_instance.py
Original file line number Diff line number Diff line change
Expand Up @@ -442,6 +442,7 @@ def activate(
self._current_completion.failed.failure.message = (
f"Failed converting activation exception: {inner_err}"
)
self._current_completion.failed.failure.application_failure_info.SetInParent()

def is_completion(command):
return (
Expand Down
82 changes: 82 additions & 0 deletions tests/worker/test_workflow.py
Original file line number Diff line number Diff line change
Expand Up @@ -74,6 +74,7 @@
)
from temporalio.converter import (
DataConverter,
DefaultFailureConverter,
DefaultFailureConverterWithEncodedAttributes,
DefaultPayloadConverter,
PayloadCodec,
Expand All @@ -84,6 +85,7 @@
ApplicationError,
CancelledError,
ChildWorkflowError,
FailureError,
TemporalError,
TimeoutError,
WorkflowAlreadyStartedError,
Expand Down Expand Up @@ -6451,6 +6453,86 @@ async def test_concurrent_sleeps_use_proper_options(
await handle.query("__temporal_workflow_metadata")


class BadFailureConverterError(Exception):
pass


class BadFailureConverter(DefaultFailureConverter):
def to_failure(
self,
exception: BaseException,
payload_converter: PayloadConverter,
failure: Failure,
) -> None:
if isinstance(exception, BadFailureConverterError):
raise RuntimeError("Intentional failure conversion error")
super().to_failure(exception, payload_converter, failure)


@activity.defn
async def bad_failure_converter_activity() -> None:
raise BadFailureConverterError


@workflow.defn(sandboxed=False)
class BadFailureConverterWorkflow:
@workflow.run
async def run(self, fail_workflow_task) -> None:
if fail_workflow_task:
raise BadFailureConverterError
else:
await workflow.execute_activity(
bad_failure_converter_activity,
schedule_to_close_timeout=timedelta(seconds=30),
retry_policy=RetryPolicy(maximum_attempts=1),
)


async def test_bad_failure_converter(client: Client):
config = client.config()
config["data_converter"] = dataclasses.replace(
config["data_converter"],
failure_converter_class=BadFailureConverter,
)
client = Client(**config)
async with new_worker(
client, BadFailureConverterWorkflow, activities=[bad_failure_converter_activity]
) as worker:
# Check activity
with pytest.raises(WorkflowFailureError) as err:
await client.execute_workflow(
BadFailureConverterWorkflow.run,
False,
id=f"workflow-{uuid.uuid4()}",
task_queue=worker.task_queue,
)
assert isinstance(err.value.cause, ActivityError)
assert isinstance(err.value.cause.cause, ApplicationError)
assert (
err.value.cause.cause.message
== "Failed building exception result: Intentional failure conversion error"
)

# Check workflow
handle = await client.start_workflow(
BadFailureConverterWorkflow.run,
True,
id=f"workflow-{uuid.uuid4()}",
task_queue=worker.task_queue,
)

async def task_failed_message() -> Optional[str]:
async for e in handle.fetch_history_events():
if e.HasField("workflow_task_failed_event_attributes"):
return e.workflow_task_failed_event_attributes.failure.message
return None

await assert_eq_eventually(
"Failed converting activation exception: Intentional failure conversion error",
task_failed_message, # type: ignore
)


@workflow.defn
class SignalsActivitiesTimersUpdatesTracingWorkflow:
"""
Expand Down

0 comments on commit 154aab9

Please sign in to comment.