Log and drop signals whose params can't be deserialized (and other error handling improvements) (#349)

Fixes #333
Fixes #318
Fixes #347

Author: cretz

poetry.lock

@@ -260,7 +260,10 @@ def _type_hints_from_func(
         if (
             index == 0
             and value.name == "self"
-            and value.annotation is inspect.Parameter.empty
+            and (
+                value.annotation is inspect.Parameter.empty
+                or str(value.annotation) == "typing.Self"
+            )
         ):
             continue
         # Stop if non-positional or not a class

temporalio/common.py

@@ -436,12 +436,7 @@ async def _run_activity(
             completion.result.completed.result.CopyFrom(
                 (await self._data_converter.encode([result]))[0]
             )
-        except (
-            Exception,
-            asyncio.CancelledError,
-            temporalio.exceptions.CancelledError,
-            temporalio.activity._CompleteAsyncError,
-        ) as err:
+        except BaseException as err:
             try:
                 if isinstance(err, temporalio.activity._CompleteAsyncError):
                     temporalio.activity.logger.debug("Completing asynchronously")

temporalio/worker/_activity.py

@@ -1231,9 +1231,19 @@ def _process_signal_job(
         defn: temporalio.workflow._SignalDefinition,
         job: temporalio.bridge.proto.workflow_activation.SignalWorkflow,
     ) -> None:
-        args = self._process_handler_args(
-            job.signal_name, job.input, defn.name, defn.arg_types, defn.dynamic_vararg
-        )
+        try:
+            args = self._process_handler_args(
+                job.signal_name,
+                job.input,
+                defn.name,
+                defn.arg_types,
+                defn.dynamic_vararg,
+            )
+        except Exception:
+            logger.exception(
+                f"Failed deserializing signal input for {job.signal_name}, dropping the signal"
+            )
+            return
         input = HandleSignalInput(
             signal=job.signal_name, args=args, headers=job.headers
         )

temporalio/worker/_workflow_instance.py

@@ -7,8 +7,8 @@
 import queue
 import sys
 import threading
+import typing
 import uuid
-import warnings
 from abc import ABC, abstractmethod
 from dataclasses import dataclass
 from datetime import datetime, timedelta, timezone
@@ -493,6 +493,46 @@ async def test_workflow_signal_qnd_query_handlers_old_dynamic_style(client: Clie
         )
 
 
+@dataclass
+class BadSignalParam:
+    some_str: str
+
+
+@workflow.defn
+class BadSignalParamWorkflow:
+    def __init__(self) -> None:
+        self._signals: List[BadSignalParam] = []
+
+    @workflow.run
+    async def run(self) -> List[BadSignalParam]:
+        await workflow.wait_condition(
+            lambda: bool(self._signals) and self._signals[-1].some_str == "finish"
+        )
+        return self._signals
+
+    @workflow.signal
+    async def some_signal(self, param: BadSignalParam) -> None:
+        self._signals.append(param)
+
+
+async def test_workflow_bad_signal_param(client: Client):
+    async with new_worker(client, BadSignalParamWorkflow) as worker:
+        handle = await client.start_workflow(
+            BadSignalParamWorkflow.run,
+            id=f"workflow-{uuid.uuid4()}",
+            task_queue=worker.task_queue,
+        )
+        # Send 4 signals, first and third are bad
+        await handle.signal("some_signal", "bad")
+        await handle.signal("some_signal", BadSignalParam(some_str="good"))
+        await handle.signal("some_signal", 123)
+        await handle.signal("some_signal", BadSignalParam(some_str="finish"))
+        assert [
+            BadSignalParam(some_str="good"),
+            BadSignalParam(some_str="finish"),
+        ] == await handle.result()
+
+
 @workflow.defn
 class AsyncUtilWorkflow:
     def __init__(self) -> None:
@@ -3055,3 +3095,27 @@ async def test_workflow_dynamic(client: Client):
         )
         assert isinstance(result, DynamicWorkflowValue)
         assert result == DynamicWorkflowValue("some-workflow - val1 - val2")
+
+
+# typing.Self only in 3.11+
+if sys.version_info >= (3, 11):
+
+    @dataclass
+    class AnnotatedWithSelfParam:
+        some_str: str
+
+    @workflow.defn
+    class WorkflowAnnotatedWithSelf:
+        @workflow.run
+        async def run(self: typing.Self, some_arg: AnnotatedWithSelfParam) -> str:
+            assert isinstance(some_arg, AnnotatedWithSelfParam)
+            return some_arg.some_str
+
+    async def test_workflow_annotated_with_self(client: Client):
+        async with new_worker(client, WorkflowAnnotatedWithSelf) as worker:
+            assert "foo" == await client.execute_workflow(
+                WorkflowAnnotatedWithSelf.run,
+                AnnotatedWithSelfParam(some_str="foo"),
+                id=f"wf-{uuid.uuid4()}",
+                task_queue=worker.task_queue,
+            )