From a2cc0244c913999a8396af59c1a7d7579a1ff01f Mon Sep 17 00:00:00 2001 From: Ayesha Waseem Date: Sun, 25 Oct 2026 16:45:23 +0500 Subject: [PATCH] Drop signals with the wrong number of arguments A signal whose argument count does not match the handler signature raised a TypeError inside the handler and failed the workflow task on every retry. The count is now checked before the handler runs, and the signal is logged and dropped, as already happens for arguments that fail to decode. --- changelog/fixed/polite-signal-bouncer.md | 1 + temporalio/worker/_workflow_instance.py | 18 ++++ temporalio/workflow/_handlers.py | 36 ++++++++ tests/worker/test_workflow.py | 109 +++++++++++++++++++++++ 4 files changed, 164 insertions(+) create mode 100644 changelog/fixed/polite-signal-bouncer.md diff --git a/changelog/fixed/polite-signal-bouncer.md b/changelog/fixed/polite-signal-bouncer.md new file mode 100644 index 000000000..048d447cb --- /dev/null +++ b/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. diff --git a/temporalio/worker/_workflow_instance.py b/temporalio/worker/_workflow_instance.py index 5bcda83f9..26cb662db 100644 --- a/temporalio/worker/_workflow_instance.py +++ b/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 ) diff --git a/temporalio/workflow/_handlers.py b/temporalio/workflow/_handlers.py index 99fafa205..eeeb0a16a 100644 --- a/temporalio/workflow/_handlers.py +++ b/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 diff --git a/tests/worker/test_workflow.py b/tests/worker/test_workflow.py index ec6d0bf83..2f163139c 100644 --- a/tests/worker/test_workflow.py +++ b/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: