diff --git a/changelog/breaking-changes/vigilant-validator-vole.md b/changelog/breaking-changes/vigilant-validator-vole.md new file mode 100644 index 0000000000..f1a9121ff7 --- /dev/null +++ b/changelog/breaking-changes/vigilant-validator-vole.md @@ -0,0 +1 @@ +Decorating an `async def` function with `@update_method.validator` or passing one as `validator` to `workflow.set_update_handler`/`set_dynamic_update_handler` now raises `ValueError`. Previously the validator was silently never run, so every update was accepted. diff --git a/temporalio/workflow/_handlers.py b/temporalio/workflow/_handlers.py index afa0bb6e4e..99fafa2051 100644 --- a/temporalio/workflow/_handlers.py +++ b/temporalio/workflow/_handlers.py @@ -267,7 +267,11 @@ def __call__( def validator( self, vfunc: Callable[MultiParamSpec, None] ) -> Callable[MultiParamSpec, None]: - """Use to decorate a function to validate the arguments passed to the update handler.""" + """Use to decorate a function to validate the arguments passed to the update handler. + + The validator must be a regular (non-``async``) function. Decorating an + ``async def`` function raises :py:class:`ValueError`. + """ ... @@ -568,6 +572,10 @@ def bind_validator(self, obj: Any) -> Callable[..., Any]: return lambda *args, **kwargs: None def set_validator(self, validator: Callable[..., None]) -> None: + # Validators are called without being awaited, so an async one would + # never run and every update would be accepted. + if inspect.iscoroutinefunction(validator): + raise ValueError("Update validator must not be an async function") if self.validator: raise RuntimeError(f"Validator already set for update {self.name}") object.__setattr__(self, "validator", validator) diff --git a/temporalio/workflow/_workflow_ops.py b/temporalio/workflow/_workflow_ops.py index 75871bc5ba..bbf95f6c2e 100644 --- a/temporalio/workflow/_workflow_ops.py +++ b/temporalio/workflow/_workflow_ops.py @@ -1049,6 +1049,7 @@ def set_update_handler( name: Name of the update. handler: Callable to set or None to unset. validator: Callable to set or None to unset as the update validator. + This must not be an ``async def`` function. """ _Runtime.current().workflow_set_update_handler(name, handler, validator) @@ -1076,6 +1077,7 @@ def set_dynamic_update_handler( Args: handler: Callable to set or None to unset. validator: Callable to set or None to unset as the update validator. + This must not be an ``async def`` function. """ _Runtime.current().workflow_set_update_handler(None, handler, validator) diff --git a/tests/test_workflow.py b/tests/test_workflow.py index 5618c34f59..5221395715 100644 --- a/tests/test_workflow.py +++ b/tests/test_workflow.py @@ -491,6 +491,42 @@ def test_workflow_update_validator_not_update(): ) +def test_workflow_update_validator_async(): + # The validator decorator itself raises, so the class body fails to execute + with pytest.raises(ValueError) as err: + + class AsyncUpdateValidator: # type:ignore[reportUnusedClass] + @workflow.update + async def my_update(self, _a: str): + pass + + # assert-type-error-pyright: "Argument of type .+ cannot be assigned to parameter" + @my_update.validator # type: ignore + async def my_validator(self, _a: str): + pass + + assert "Update validator must not be an async function" in str(err.value) + + +class SyncUpdateValidator: + @workflow.update + async def my_update(self, _a: str): + pass + + @my_update.validator + def my_validator(self, _a: str) -> None: + pass + + @workflow.run + async def run(self): + pass + + +def test_workflow_update_validator_sync(): + defn = workflow._Definition.must_from_class(workflow.defn(SyncUpdateValidator)) + assert defn.updates["my_update"].validator is SyncUpdateValidator.my_validator + + def _assert_config_function_parity( function_obj: Callable[..., Any], config_class: type[Any], diff --git a/tests/worker/test_workflow.py b/tests/worker/test_workflow.py index 4186bb6676..ee583abdec 100644 --- a/tests/worker/test_workflow.py +++ b/tests/worker/test_workflow.py @@ -6086,6 +6086,57 @@ async def test_workflow_update_task_fails(client: Client): assert bad_validator_fail_ct == 2 +@workflow.defn +class AsyncUpdateValidatorDynamicWorkflow: + @workflow.run + async def run(self) -> list[str]: + def handler(_arg: int) -> int: + return 1 + + async def async_validator(_arg: int) -> None: + pass + + def sync_validator(_arg: int) -> None: + pass + + def dynamic_handler(_name: str, _args: Sequence[RawValue]) -> int: + return 3 + + # A rejected registration must not replace the existing handler + workflow.set_update_handler("named", handler, validator=sync_validator) + results = [] + for set_handler in ( + lambda: workflow.set_update_handler( + "named", lambda _arg: 2, validator=async_validator + ), + lambda: workflow.set_dynamic_update_handler( + dynamic_handler, validator=async_validator + ), + ): + try: + set_handler() + except ValueError as err: + results.append(str(err)) + results.append(str(workflow.get_update_handler("named") is handler)) + results.append(str(workflow.get_dynamic_update_handler() is None)) + return results + + +async def test_workflow_update_async_validator_dynamic_rejected(client: Client): + async with new_worker(client, AsyncUpdateValidatorDynamicWorkflow) as worker: + result = await client.execute_workflow( + AsyncUpdateValidatorDynamicWorkflow.run, + id=f"async-update-validator-dynamic-{uuid.uuid4()}", + task_queue=worker.task_queue, + ) + assert result == [ + "Update validator must not be an async function", + "Update validator must not be an async function", + "True", + "True", + ] + + async def test_workflow_update_continue_as_new_fails_task(client: Client): # Continue-as-new is a control-flow BaseException. From an update handler # it must fail the workflow task, then retry the update rather than letting