Drop signals with the wrong number of arguments

changelog/fixed/polite-signal-bouncer.md

@@ -0,0 +1 @@
+Signals sent to a workflow with too few or too many positional arguments for the handler are now logged and dropped, like signals with undecodable arguments, instead of repeatedly failing the workflow task.

temporalio/worker/_workflow_instance.py

@@ -2742,6 +2742,24 @@ def _process_signal_job(
                 extra={"temporal_workflow": self._info._logger_details()},
             )
             return
+        if defn.arg_count_range is not None:
+            minimum, maximum = defn.arg_count_range
+            if len(args) < minimum or (maximum is not None and len(args) > maximum):
+                expected = (
+                    str(minimum)
+                    if minimum == maximum
+                    else f"at least {minimum}"
+                    if maximum is None
+                    else f"{minimum} to {maximum}"
+                )
+                logger.error(
+                    f"Signal {job.signal_name} on workflow {self._info.workflow_type}"
+                    f" with ID {self._info.workflow_id} and run ID {self._info.run_id}"
+                    f" received {len(args)} arguments but accepts {expected},"
+                    " dropping the signal",
+                    extra={"temporal_workflow": self._info._logger_details()},
+                )
+                return
         input = HandleSignalInput(
             signal=job.signal_name, args=args, headers=job.headers
         )

temporalio/workflow/_handlers.py

@@ -461,6 +461,7 @@ class _SignalDefinition:
     # Types loaded on post init if None
     arg_types: list[type] | None = None
     dynamic_vararg: bool = False
+    arg_count_range: tuple[int, int | None] | None = None
 
     @staticmethod
     def from_fn(fn: Callable) -> _SignalDefinition | None:
@@ -494,11 +495,46 @@ def __post_init__(self) -> None:
                     ),
                 )
             object.__setattr__(self, "arg_types", arg_types)
+        if self.name and self.arg_count_range is None:
+            object.__setattr__(
+                self,
+                "arg_count_range",
+                _positional_arg_count_range(self.fn, self.is_method),
+            )
 
     def bind_fn(self, obj: Any) -> Callable[..., Any]:
         return _bind_method(obj, self.fn)
 
 
+def _positional_arg_count_range(
+    fn: Callable, is_method: bool
+) -> tuple[int, int | None] | None:
+    try:
+        params = list(inspect.signature(fn).parameters.values())
+    except (TypeError, ValueError):
+        return None
+    if is_method:
+        params = params[1:]
+    if any(
+        p.kind is inspect.Parameter.KEYWORD_ONLY
+        and p.default is inspect.Parameter.empty
+        for p in params
+    ):
+        return None
+    positional = [
+        p
+        for p in params
+        if p.kind
+        in (
+            inspect.Parameter.POSITIONAL_ONLY,
+            inspect.Parameter.POSITIONAL_OR_KEYWORD,
+        )
+    ]
+    minimum = sum(p.default is inspect.Parameter.empty for p in positional)
+    has_varargs = any(p.kind is inspect.Parameter.VAR_POSITIONAL for p in params)
+    return minimum, None if has_varargs else len(positional)
+
+
 @dataclass(frozen=True)
 class _QueryDefinition:
     # None if dynamic

tests/worker/test_workflow.py

@@ -742,6 +742,115 @@ async def test_workflow_bad_signal_param(client: Client):
         )
 
 
+@workflow.defn
+class BadSignalArgCountWorkflow:
+    def __init__(self) -> None:
+        self._received: list[str] = []
+
+    @workflow.run
+    async def run(self) -> list[str]:
+        await workflow.wait_condition(lambda: "finish" in self._received)
+        return self._received
+
+    @workflow.signal
+    async def required(self, value: str) -> None:
+        self._received.append(f"required:{value}")
+
+    @workflow.signal
+    async def optional(self, value: str = "default") -> None:
+        self._received.append(f"optional:{value}")
+
+    @workflow.signal
+    async def finish(self) -> None:
+        self._received.append("finish")
+
+
+async def test_workflow_bad_signal_arg_count(client: Client):
+    with LogCapturer().logs_captured(
+        temporalio.worker._workflow_instance.logger
+    ) as capturer:
+        async with new_worker(client, BadSignalArgCountWorkflow) as worker:
+            handle = await client.start_workflow(
+                BadSignalArgCountWorkflow.run,
+                id=f"workflow-{uuid.uuid4()}",
+                task_queue=worker.task_queue,
+            )
+            await handle.signal("required", args=[])
+            await handle.signal("required", args=["a", "b"])
+            await handle.signal("optional", args=["a", "b"])
+            await handle.signal("required", "good")
+            await handle.signal("optional")
+            await handle.signal("optional", "given")
+            await handle.signal(BadSignalArgCountWorkflow.finish)
+            assert [
+                "required:good",
+                "optional:default",
+                "optional:given",
+                "finish",
+            ] == await handle.result()
+
+        messages = {
+            record.message
+            for record in capturer.find_all(
+                lambda l: "dropping the signal" in l.message
+            )
+        }
+        assert len(messages) == 3
+        assert any(
+            "Signal required" in m and "received 0 arguments but accepts 1" in m
+            for m in messages
+        )
+        assert any(
+            "Signal required" in m and "received 2 arguments but accepts 1" in m
+            for m in messages
+        )
+        assert any(
+            "Signal optional" in m and "received 2 arguments but accepts 0 to 1" in m
+            for m in messages
+        )
+        assert all(
+            "BadSignalArgCountWorkflow" in m and handle.id in m for m in messages
+        )
+
+
+def test_positional_arg_count_range():
+    from temporalio.workflow._handlers import _positional_arg_count_range
+
+    def none() -> None: ...
+
+    def two(a: str, b: str) -> list[str]:
+        return [a, b]
+
+    def defaults(a: str, b: str = "", c: str = "") -> list[str]:
+        return [a, b, c]
+
+    def varargs(a: str, *rest: str) -> list[str]:
+        return [a, *rest]
+
+    def positional_only(a: str, /, b: str) -> list[str]:
+        return [a, b]
+
+    def keyword_default(a: str, *, b: str = "") -> list[str]:
+        return [a, b]
+
+    def keyword_required(a: str, *, b: str) -> list[str]:
+        return [a, b]
+
+    class Callback:
+        def __call__(self, a: str) -> str:
+            return a
+
+    assert _positional_arg_count_range(none, False) == (0, 0)
+    assert _positional_arg_count_range(two, False) == (2, 2)
+    assert _positional_arg_count_range(defaults, False) == (1, 3)
+    assert _positional_arg_count_range(varargs, False) == (1, None)
+    assert _positional_arg_count_range(positional_only, False) == (2, 2)
+    assert _positional_arg_count_range(keyword_default, False) == (1, 1)
+    assert _positional_arg_count_range(keyword_required, False) is None
+    assert _positional_arg_count_range(Callback(), False) == (1, 1)
+    assert _positional_arg_count_range(Callback.__call__, True) == (1, 1)
+
+
 @workflow.defn
 class AsyncUtilWorkflow:
     def __init__(self) -> None: