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: