diff --git a/CHANGELOG.md b/CHANGELOG.md index 3219ea2..cf34463 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -14,6 +14,21 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 `deprecate_patch` for the same change ID, while validating each call's range and preserving retained duplicate-marker sequences. +## [2.6.2] - 2026-10-20 + +### Fixed + +- Preserve workflow entry argument-binding refusals before application code + runs, with truthful Worker capability discovery for compatible deployment + routing. Workflow body exceptions remain ordinary workflow failures. +- Run synchronous remote activities outside the asyncio event loop so other + workflow tasks, polls and heartbeats can progress. +- Reject unsupported coroutine signal/update handlers and generator handlers + before registration or replay, including inherited and wrapped handlers. + Guard returned values without removing the Worker's existing asynchronous + query and update-validator support. Local synchronous query/validation helpers + refuse and close returned coroutines. + ## [2.6.1] - 2026-10-19 ### Fixed diff --git a/docs/reference/workflow.md b/docs/reference/workflow.md index 7811a20..841496b 100644 --- a/docs/reference/workflow.md +++ b/docs/reference/workflow.md @@ -1,5 +1,32 @@ # Workflow +## Signal, update and query handlers + +Handlers decorated with `@workflow.signal` or `@workflow.update` must be +synchronous functions. Coroutine +(`async def`), async-generator and generator handlers are rejected with +`InvalidWorkflowDefinition` when the workflow is defined, registered or replayed, +including inherited handlers. Returning an asynchronous or generator value from +a synchronous wrapper is also refused. + +Queries and update validators may be synchronous or `async def`. The Worker +awaits their results; queries retain their original deadline and cancellation +authority. The local synchronous `query_state()` and `validate_update()` helpers +refuse asynchronous results with `QueryFailed` and `UpdateRejected` respectively +and close returned coroutines. Generator and async-generator query or validator +handlers are rejected at definition or registration. + +Each replay constructs the workflow instance before executing `run()`. Signals +already present at workflow start are applied after `run()` reaches its first +yield, so constructor and pre-yield initialization happen before those handlers. +Signal and update handlers mutate that replayed state; queries inspect it and +update validators check a request before acceptance. Put durable activities, +timers and child workflow commands in `run()`, not in a handler. + +An exception from a signal handler while replaying fails the workflow run. +Update-handler exceptions fail the update, validator exceptions reject it, +and query-handler exceptions fail the query. They do not silently succeed. + Workflow retry and timeout settings are durable command budgets. Use `ActivityRetryPolicy` on `ctx.schedule_activity(...)` for activity attempts and `ChildWorkflowRetryPolicy` on `ctx.start_child_workflow(...)` for child workflow diff --git a/src/durable_workflow/worker.py b/src/durable_workflow/worker.py index 1eacfb5..b6a734e 100644 --- a/src/durable_workflow/worker.py +++ b/src/durable_workflow/worker.py @@ -109,12 +109,13 @@ StartChildWorkflow, UpsertMemo, _prepare_query_state, + _prepare_update_validation, + _UpdateValidationReplayOptions, _validate_workflow_entrypoint, _workflow_handler_registries, apply_update, commands_to_server_commands, replay, - validate_update, ) log = logging.getLogger("durable_workflow.worker") @@ -3867,21 +3868,24 @@ async def _run_update_validation_task(self, task: dict[str, Any]) -> str: codec=codec, context="workflow update validation arguments", ) - result = validate_update( + invoke = _prepare_update_validation( workflow_cls, history, start_input, update_name, update_args, - workflow_id=task.get("workflow_id"), - run_id=task.get("run_id", ""), - payload_codec=codec, - external_storage=self.external_storage, - external_storage_cache=self.external_storage_cache, - prepare_local_activities=self._prepared_local_activities_supported, - prepare_local_activity_groups=self._prepared_local_activity_groups_supported, - local_activity_cancellation_policies=self._local_activity_cancellation_policies, + replay_options=_UpdateValidationReplayOptions( + workflow_id=task.get("workflow_id"), + run_id=task.get("run_id", ""), + payload_codec=codec, + external_storage=self.external_storage, + external_storage_cache=self.external_storage_cache, + prepare_local_activities=self._prepared_local_activities_supported, + prepare_local_activity_groups=self._prepared_local_activity_groups_supported, + local_activity_cancellation_policies=self._local_activity_cancellation_policies, + ), ) + result = invoke() if inspect.isawaitable(result): result = await result if result is False: diff --git a/src/durable_workflow/workflow.py b/src/durable_workflow/workflow.py index f98ce98..d47e0e9 100644 --- a/src/durable_workflow/workflow.py +++ b/src/durable_workflow/workflow.py @@ -138,6 +138,21 @@ def _workflow_handler_registries( wire_name = getattr(member, f"__{kind}_name__", None) if not isinstance(wire_name, str) or not wire_name: continue + if not callable(member): + raise InvalidWorkflowDefinition(f"workflow {cls.__qualname__}.{attr} handler must be callable") + target = inspect.unwrap(member) + if not inspect.isroutine(target) and callable(target): + target = inspect.unwrap(target.__call__) + if ( + (inspect.iscoroutinefunction(target) and kind not in {"query", "update_validator"}) + or inspect.isasyncgenfunction(target) + or inspect.isgeneratorfunction(target) + ): + name = getattr(cls, "__workflow_name__", cls.__name__) + raise InvalidWorkflowDefinition( + f"unsupported_workflow_handler: workflow {name!r} ({cls.__qualname__}.{attr}) " + f"{kind} handler {wire_name!r} must be synchronous; yield durable commands from run()." + ) if kind == "signal" and wire_name == MESSAGE_STREAM_SIGNAL: raise ValueError(f"signal name {wire_name!r} is reserved by the workflow runtime") if wire_name in registry: @@ -149,6 +164,28 @@ def _workflow_handler_registries( return handlers["signal"], handlers["query"], handlers["update"], handlers["update_validator"] +def _invoke_workflow_handler( + handler: Callable[..., Any], args: list[Any], cls: type, method: str, *, allow_async: bool = False, +) -> Any: + return _validate_workflow_handler_result(handler(*args), cls, method, allow_async=allow_async) + + +def _validate_workflow_handler_result(result: Any, cls: type, method: str, *, allow_async: bool = False) -> Any: + if ( + (inspect.isawaitable(result) and not allow_async) + or inspect.isasyncgen(result) + or inspect.isgenerator(result) + ): + if inspect.iscoroutine(result) or inspect.isgenerator(result): + result.close() + name = getattr(cls, "__workflow_name__", cls.__name__) + raise InvalidWorkflowDefinition( + f"unsupported_workflow_handler_result: workflow {name!r} ({cls.__qualname__}.{method}) " + "returned an async or generator value; handlers must return synchronous values." + ) + return result + + def defn(*, name: str) -> Callable[[type[_WorkflowT]], type[_WorkflowT]]: """Register a class as a workflow type under a language-neutral name. @@ -3133,7 +3170,9 @@ def _prepare_query_state( def invoke() -> Any: try: - return handler(*(list(args) if args is not None else [])) + return _invoke_workflow_handler( + handler, list(args) if args is not None else [], workflow_cls, method_name, allow_async=True, + ) except QueryFailed: raise except Exception as exc: @@ -3163,7 +3202,7 @@ def query_state( Server-routed Worker execution separately checks its original query claim between replay and handler invocation. Direct local use has no Server lease. """ - return _prepare_query_state( + result = _prepare_query_state( workflow_cls, history_events, start_input, query_name, args, workflow_id=workflow_id, run_id=run_id, payload_codec=payload_codec, external_storage=external_storage, external_storage_cache=external_storage_cache, @@ -3171,6 +3210,11 @@ def query_state( prepare_local_activity_groups=prepare_local_activity_groups, local_activity_cancellation_policies=local_activity_cancellation_policies, )() + try: + registry: dict[str, str] = getattr(workflow_cls, "__workflow_queries__", {}) or {} + return _validate_workflow_handler_result(result, workflow_cls, registry.get(query_name, query_name)) + except InvalidWorkflowDefinition as exc: + raise QueryFailed(str(exc)) from exc def apply_update( @@ -3276,11 +3320,24 @@ def apply_update( ) try: - return CompleteUpdate(update_id=update_id, result=handler(*args)) + result = _invoke_workflow_handler(handler, args, workflow_cls, method_name) + return CompleteUpdate(update_id=update_id, result=result) except Exception as exc: return _fail_update_from_exception(update_id, "update handler failed", exc) +@dataclass(frozen=True) +class _UpdateValidationReplayOptions: + workflow_id: str | None = None + run_id: str = "" + payload_codec: str | None = None + external_storage: ExternalStorageDriver | None = None + external_storage_cache: ExternalPayloadCache | None = None + prepare_local_activities: bool = False + prepare_local_activity_groups: bool = False + local_activity_cancellation_policies: tuple[str, ...] = () + + def validate_update( workflow_cls: type, history_events: Iterable[dict[str, Any]], @@ -3297,13 +3354,42 @@ def validate_update( prepare_local_activity_groups: bool = False, local_activity_cancellation_policies: tuple[str, ...] = (), ) -> Any: + """Replay and invoke a validator synchronously, refusing asynchronous results.""" + from .errors import UpdateRejected + + result = _prepare_update_validation( + workflow_cls, history_events, start_input, update_name, args, + replay_options=_UpdateValidationReplayOptions( + workflow_id=workflow_id, run_id=run_id, payload_codec=payload_codec, + external_storage=external_storage, external_storage_cache=external_storage_cache, + prepare_local_activities=prepare_local_activities, + prepare_local_activity_groups=prepare_local_activity_groups, + local_activity_cancellation_policies=local_activity_cancellation_policies, + ), + )() + try: + registry: dict[str, str] = getattr(workflow_cls, "__workflow_update_validators__", {}) or {} + return _validate_workflow_handler_result(result, workflow_cls, registry.get(update_name, update_name)) + except InvalidWorkflowDefinition as exc: + raise UpdateRejected(str(exc), exception_type=type(exc).__name__) from exc + + +def _prepare_update_validation( + workflow_cls: type, + history_events: Iterable[dict[str, Any]], + start_input: list[Any], + update_name: str, + args: list[Any], + *, + replay_options: _UpdateValidationReplayOptions, +) -> Callable[[], Any]: """Replay state and invoke only the declared pre-accept update validator. The replayed instance is discarded after validation. This helper does not run the update handler and does not emit workflow commands, so the server remains the sole authority that can cross the accepted-state boundary. """ - from .errors import InvalidArgument, UpdateRejected, UpdateValidationFailed + from .errors import UpdateValidationFailed events = list(history_events) try: @@ -3311,14 +3397,14 @@ def validate_update( workflow_cls, events, start_input, - workflow_id=workflow_id, - run_id=run_id, - payload_codec=payload_codec, - external_storage=external_storage, - external_storage_cache=external_storage_cache, - prepare_local_activities=prepare_local_activities, - prepare_local_activity_groups=prepare_local_activity_groups, - local_activity_cancellation_policies=local_activity_cancellation_policies, + workflow_id=replay_options.workflow_id, + run_id=replay_options.run_id, + payload_codec=replay_options.payload_codec, + external_storage=replay_options.external_storage, + external_storage_cache=replay_options.external_storage_cache, + prepare_local_activities=replay_options.prepare_local_activities, + prepare_local_activity_groups=replay_options.prepare_local_activity_groups, + local_activity_cancellation_policies=replay_options.local_activity_cancellation_policies, ) except Exception as exc: raise UpdateValidationFailed( @@ -3349,8 +3435,20 @@ def validate_update( if validator is None: raise UpdateValidationFailed(f"update validator {update_name!r} is not available") + return lambda: _invoke_update_validator(validator, args, workflow_cls, validator_name, update_name) + + +def _invoke_update_validator( + validator: Callable[..., Any], + args: list[Any], + workflow_cls: type, + validator_name: str, + update_name: str, +) -> Any: + from .errors import InvalidArgument, UpdateRejected + try: - result = validator(*list(args)) + result = _invoke_workflow_handler(validator, list(args), workflow_cls, validator_name, allow_async=True) if result is False: raise ValueError(f"update validator {update_name!r} returned false") return result @@ -4240,6 +4338,7 @@ def _replay_state( allow_cancellation_scope_delivery: bool = False, ) -> _ReplayState: _validate_workflow_entrypoint(workflow_cls) + _workflow_handler_registries(workflow_cls) if allow_cancellation_scope_delivery and not allow_cancellation_scope_authoring: raise LocalActivityExecutionAborted("scope delivery requires canonical scope authoring") if prepare_local_activity_groups and not prepare_local_activities: @@ -5285,7 +5384,7 @@ def _apply_receiver(receiver: _PendingReceiver) -> None: return raise TypeError(f"update handler {receiver.name!r} is not available") ctx.logger._set_replaying(True) - handler(*receiver.args) + _invoke_workflow_handler(handler, receiver.args, workflow_cls, method_name) def _receiver_due(receiver: _PendingReceiver, *, before_consuming_result: bool) -> bool: if receiver.after_cancellation_delivery and not cancellation_consumed: diff --git a/tests/fixtures/replay_regressions/unsupported-async-workflow-signal.json b/tests/fixtures/replay_regressions/unsupported-async-workflow-signal.json new file mode 100644 index 0000000..91e419e --- /dev/null +++ b/tests/fixtures/replay_regressions/unsupported-async-workflow-signal.json @@ -0,0 +1,10 @@ +{ + "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json", + "fixture_schema": "durable-workflow.replay-regression/v1", + "id": "unsupported-async-workflow-signal", + "protocol_version": "1.20", + "bindings": ["python"], + "workflow": {"type": "tests.replay.unsupported-async-signal", "input": [], "payload_codec": "avro"}, + "history": [{"event_type": "WorkflowStarted", "payload": {}}], + "expected": {"error": "unsupported_workflow_handler"} +} diff --git a/tests/test_replay_regression_corpus.py b/tests/test_replay_regression_corpus.py index c51d360..74eaca4 100644 --- a/tests/test_replay_regression_corpus.py +++ b/tests/test_replay_regression_corpus.py @@ -287,6 +287,17 @@ async def run(self, ctx: WorkflowContext) -> str: return "body was never executed" +class UnsupportedAsyncSignalWorkflow: + __workflow_name__ = "tests.replay.unsupported-async-signal" + + @workflow.signal("change") + async def receive(self) -> None: + raise AssertionError("unsupported signal body must never run") + + def run(self, ctx: WorkflowContext) -> str: + return "done" + + @workflow.defn(name="tests.replay.entry-arguments-required") class RequiredEntryArgumentWorkflow: def run(self, ctx: WorkflowContext, amount: int, required: int) -> int: @@ -334,6 +345,7 @@ def run(self, ctx: WorkflowContext, cleanup_seconds: int = 0): # type: ignore[n NexusTerminalDeliveryWorkflow, NexusTargetWorkflow, UnsupportedAsyncWorkflow, + UnsupportedAsyncSignalWorkflow, ChildPolicyAuthorWorkflow, ColdReplacementSatisfiedConditionWorkflow, CooperativeReopenedConditionCleanupWorkflow, diff --git a/tests/test_worker.py b/tests/test_worker.py index daf6bd9..c4f6789 100644 --- a/tests/test_worker.py +++ b/tests/test_worker.py @@ -3805,7 +3805,7 @@ async def test_task_root_codec_rejection_precedes_decode_and_user_work( monkeypatch.setattr(worker_module, "replay", replay_spy) monkeypatch.setattr(worker_module, "_prepare_query_state", query_spy) monkeypatch.setattr(worker_module, "apply_update", update_spy) - monkeypatch.setattr(worker_module, "validate_update", validator_spy) + monkeypatch.setattr(worker_module, "_prepare_update_validation", validator_spy) monkeypatch.setattr(worker, "_execute_activity_callable", activity_spy) if task_kind == "activity": @@ -3934,13 +3934,13 @@ async def test_exact_avro_task_completes_each_path_with_handler_work( replay_spy = Mock(wraps=worker_module.replay) query_spy = Mock(wraps=worker_module._prepare_query_state) update_spy = Mock(wraps=worker_module.apply_update) - validator_spy = Mock(wraps=worker_module.validate_update) + validator_spy = Mock(wraps=worker_module._prepare_update_validation) activity_spy = AsyncMock(wraps=worker._execute_activity_callable) monkeypatch.setattr(serializer, "decode_envelope", decode_spy) monkeypatch.setattr(worker_module, "replay", replay_spy) monkeypatch.setattr(worker_module, "_prepare_query_state", query_spy) monkeypatch.setattr(worker_module, "apply_update", update_spy) - monkeypatch.setattr(worker_module, "validate_update", validator_spy) + monkeypatch.setattr(worker_module, "_prepare_update_validation", validator_spy) monkeypatch.setattr(worker, "_execute_activity_callable", activity_spy) if task_kind == "activity": diff --git a/tests/test_workflow_handler_validation.py b/tests/test_workflow_handler_validation.py new file mode 100644 index 0000000..27ad5e1 --- /dev/null +++ b/tests/test_workflow_handler_validation.py @@ -0,0 +1,150 @@ +from __future__ import annotations + +from collections.abc import AsyncIterator, Generator +from functools import partial, wraps +from typing import Any + +import pytest + +from durable_workflow import Client, Worker, workflow +from durable_workflow.errors import InvalidWorkflowDefinition +from durable_workflow.testing import WorkflowEnvironment + + +def _handler(shape: str) -> Any: + async def coroutine(self: Any) -> None: + raise AssertionError("unsupported handler must never run") + + + async def async_generator(self: Any) -> AsyncIterator[None]: + yield None + + + def generator(self: Any) -> Generator[None, None, None]: + yield None + + return {"coroutine": coroutine, "async_generator": async_generator, "generator": generator}[shape] + + +def _definition(kind: str, handler: Any, inherited: bool) -> type: + decorated = getattr(workflow, kind)("change")(handler) + + def run(self: Any, ctx: Any) -> str: + return "done" + + base = type("HandlerWorkflow", (), {"receive": decorated, "run": run}) + definition = type("InheritedHandlerWorkflow", (base,), {}) if inherited else base + definition.__workflow_name__ = "handler-validation" + return definition + + +@pytest.mark.parametrize("kind,shape", [ + (kind, shape) + for kind in ("signal", "query", "update", "update_validator") + for shape in ("coroutine", "async_generator", "generator") + if shape != "coroutine" or kind not in {"query", "update_validator"} +]) +@pytest.mark.parametrize("inherited", [False, True]) +@pytest.mark.parametrize("entry", ["definition", "worker", "environment", "replayer", "replay"]) +def test_unsupported_handlers_are_rejected_before_execution( + kind: str, shape: str, inherited: bool, entry: str, +) -> None: + definition = _definition(kind, _handler(shape), inherited) + if entry == "definition": + invoke = partial(workflow.defn(name="handler-validation"), definition) + elif entry == "worker": + client = Client("http://127.0.0.1:1") + invoke = partial(Worker, client, task_queue="unused", workflows=[definition]) + elif entry == "environment": + invoke = partial(WorkflowEnvironment().register_workflow, definition) + elif entry == "replayer": + invoke = partial(workflow.Replayer, workflows=[definition]) + else: + invoke = partial(workflow.replay, definition, [], []) + with pytest.raises(InvalidWorkflowDefinition, match="handler-validation.*receive.*synchronous"): + invoke() + + +@pytest.mark.parametrize("shape", ["coroutine", "async_generator", "generator"]) +def test_wrapped_handlers_are_validated(shape: str) -> None: + original = _handler(shape) + + @wraps(original) + def receive(self: Any) -> Any: + return original(self) + + definition = _definition("signal", receive, False) + decorate = workflow.defn(name="handler-validation") + with pytest.raises(InvalidWorkflowDefinition, match="handler-validation.*receive"): + decorate(definition) + + +@pytest.mark.parametrize("kind", ["signal", "query", "update", "update_validator"]) +@pytest.mark.parametrize("shape", ["coroutine", "async_generator", "generator"]) +def test_sync_wrapper_cannot_return_an_async_or_generator_result(kind: str, shape: str) -> None: + original = _handler(shape) + + def receive(self: Any) -> Any: + return original(self) + + definition = workflow.defn(name="handler-validation")(_definition(kind, receive, False)) + if kind == "signal": + from durable_workflow import serializer + + history = [{"event_type": "SignalReceived", "payload": { + "signal_name": "change", "value": serializer.envelope([]), "payload_codec": "avro", + }}] + + def run(self: Any, ctx: Any) -> Any: + yield ctx.schedule_activity("tick", []) + + definition.run = run + with pytest.raises(InvalidWorkflowDefinition, match="unsupported_workflow_handler_result.*receive"): + workflow.replay(definition, history, []) + elif kind == "query": + from durable_workflow.errors import QueryFailed + + with pytest.raises(QueryFailed, match="unsupported_workflow_handler_result.*receive"): + workflow.query_state(definition, [], [], "change") + elif kind == "update": + from durable_workflow import serializer + + result = workflow.apply_update(definition, [{"event_type": "UpdateAccepted", "payload": { + "update_id": "update-1", "update_name": "change", "value": serializer.envelope([]), + "payload_codec": "avro", + }}], [], "update-1") + assert isinstance(result, workflow.FailUpdate) + assert "unsupported_workflow_handler_result" in result.message + else: + from durable_workflow.errors import UpdateRejected + + definition.__workflow_updates__ = {"change": "receive"} + with pytest.raises(UpdateRejected, match="unsupported_workflow_handler_result.*receive"): + workflow.validate_update(definition, [], [], "change", []) + + +def test_valid_sync_handlers_remain_supported() -> None: + @workflow.defn(name="sync-handlers") + class SynchronousWorkflow: + @workflow.signal("change") + def receive(self) -> None: + pass + + @workflow.query("read") + def read(self) -> str: + return "value" + + @workflow.update("change") + def change(self) -> str: + return "updated" + + @workflow.update_validator("change") + def validate(self) -> bool: + return True + + def run(self, ctx: Any) -> str: + return "done" + + Worker(Client("http://127.0.0.1:1"), task_queue="unused", workflows=[SynchronousWorkflow]) + assert WorkflowEnvironment().execute_workflow(SynchronousWorkflow) == "done" + assert workflow.Replayer(workflows=[SynchronousWorkflow]).replay([], [], workflow_type="sync-handlers")