From 4e135a29c71be04d05b7bd23258ed058cbddc6a3 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Tue, 20 Oct 2026 17:12:30 +0000 Subject: [PATCH 1/4] Reproduce unsupported Python workflow handler definitions and results --- tests/test_workflow_handler_validation.py | 142 ++++++++++++++++++++++ 1 file changed, 142 insertions(+) create mode 100644 tests/test_workflow_handler_validation.py diff --git a/tests/test_workflow_handler_validation.py b/tests/test_workflow_handler_validation.py new file mode 100644 index 0000000..77d776d --- /dev/null +++ b/tests/test_workflow_handler_validation.py @@ -0,0 +1,142 @@ +from __future__ import annotations + +from collections.abc import AsyncIterator, Generator +from functools import 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", ["signal", "query", "update", "update_validator"]) +@pytest.mark.parametrize("shape", ["coroutine", "async_generator", "generator"]) +@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) + with pytest.raises(InvalidWorkflowDefinition, match="handler-validation.*receive.*synchronous"): + if entry == "definition": + workflow.defn(name="handler-validation")(definition) + elif entry == "worker": + Worker(Client("http://127.0.0.1:1"), task_queue="unused", workflows=[definition]) + elif entry == "environment": + WorkflowEnvironment().register_workflow(definition) + elif entry == "replayer": + workflow.Replayer(workflows=[definition]) + else: + workflow.replay(definition, [], []) + + +@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) + + with pytest.raises(InvalidWorkflowDefinition, match="handler-validation.*receive"): + workflow.defn(name="handler-validation")(_definition("signal", receive, False)) + + +@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") From f5769bb8124d231cf47b2aba1efd206edfd8193d Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Tue, 20 Oct 2026 17:27:23 +0000 Subject: [PATCH 2/4] Validate workflow handlers while preserving async query and validator paths --- CHANGELOG.md | 15 ++++++ docs/reference/workflow.md | 27 +++++++++++ src/durable_workflow/worker.py | 2 + src/durable_workflow/workflow.py | 48 +++++++++++++++++-- .../unsupported-async-workflow-signal.json | 10 ++++ tests/test_replay_regression_corpus.py | 12 +++++ tests/test_workflow_handler_validation.py | 8 +++- 7 files changed, 116 insertions(+), 6 deletions(-) create mode 100644 tests/fixtures/replay_regressions/unsupported-async-workflow-signal.json 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..19fc001 100644 --- a/src/durable_workflow/worker.py +++ b/src/durable_workflow/worker.py @@ -3373,6 +3373,7 @@ async def _run_query_task_core(self, task: dict[str, Any], *, client: Client | N 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, + allow_async=True, ) await authority.confirm() async def execute() -> Any: @@ -3881,6 +3882,7 @@ async def _run_update_validation_task(self, task: dict[str, Any]) -> str: 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, + allow_async=True, ) if inspect.isawaitable(result): result = await result diff --git a/src/durable_workflow/workflow.py b/src/durable_workflow/workflow.py index f98ce98..0856635 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,25 @@ 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: + result = handler(*args) + 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. @@ -3099,6 +3133,7 @@ def _prepare_query_state( prepare_local_activities: bool = False, prepare_local_activity_groups: bool = False, local_activity_cancellation_policies: tuple[str, ...] = (), + allow_async: bool = False, ) -> Callable[[], Any]: """Prepare the query so a worker can recheck authority after replay.""" try: @@ -3133,7 +3168,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=allow_async, + ) except QueryFailed: raise except Exception as exc: @@ -3276,7 +3313,8 @@ 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) @@ -3296,6 +3334,7 @@ def validate_update( prepare_local_activities: bool = False, prepare_local_activity_groups: bool = False, local_activity_cancellation_policies: tuple[str, ...] = (), + allow_async: bool = False, ) -> Any: """Replay state and invoke only the declared pre-accept update validator. @@ -3350,7 +3389,7 @@ def validate_update( raise UpdateValidationFailed(f"update validator {update_name!r} is not available") try: - result = validator(*list(args)) + result = _invoke_workflow_handler(validator, list(args), workflow_cls, validator_name, allow_async=allow_async) if result is False: raise ValueError(f"update validator {update_name!r} returned false") return result @@ -4240,6 +4279,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 +5325,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_workflow_handler_validation.py b/tests/test_workflow_handler_validation.py index 77d776d..ddf8568 100644 --- a/tests/test_workflow_handler_validation.py +++ b/tests/test_workflow_handler_validation.py @@ -38,8 +38,12 @@ def run(self: Any, ctx: Any) -> str: return definition -@pytest.mark.parametrize("kind", ["signal", "query", "update", "update_validator"]) -@pytest.mark.parametrize("shape", ["coroutine", "async_generator", "generator"]) +@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( From 106d9e01310b79247dc16897e390837549459f89 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Tue, 20 Oct 2026 17:49:50 +0000 Subject: [PATCH 3/4] Keep sync handler result checks at their public execution boundary --- src/durable_workflow/worker.py | 7 +- src/durable_workflow/workflow.py | 81 +++++++++++++++++------ tests/test_worker.py | 6 +- tests/test_workflow_handler_validation.py | 28 ++++---- 4 files changed, 84 insertions(+), 38 deletions(-) diff --git a/src/durable_workflow/worker.py b/src/durable_workflow/worker.py index 19fc001..c01226d 100644 --- a/src/durable_workflow/worker.py +++ b/src/durable_workflow/worker.py @@ -109,12 +109,12 @@ StartChildWorkflow, UpsertMemo, _prepare_query_state, + _prepare_update_validation, _validate_workflow_entrypoint, _workflow_handler_registries, apply_update, commands_to_server_commands, replay, - validate_update, ) log = logging.getLogger("durable_workflow.worker") @@ -3373,7 +3373,6 @@ async def _run_query_task_core(self, task: dict[str, Any], *, client: Client | N 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, - allow_async=True, ) await authority.confirm() async def execute() -> Any: @@ -3868,7 +3867,7 @@ 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, @@ -3882,8 +3881,8 @@ async def _run_update_validation_task(self, task: dict[str, Any]) -> str: 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, - allow_async=True, ) + 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 0856635..6cc4a6f 100644 --- a/src/durable_workflow/workflow.py +++ b/src/durable_workflow/workflow.py @@ -167,7 +167,10 @@ def _workflow_handler_registries( def _invoke_workflow_handler( handler: Callable[..., Any], args: list[Any], cls: type, method: str, *, allow_async: bool = False, ) -> Any: - result = handler(*args) + 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) @@ -3133,7 +3136,6 @@ def _prepare_query_state( prepare_local_activities: bool = False, prepare_local_activity_groups: bool = False, local_activity_cancellation_policies: tuple[str, ...] = (), - allow_async: bool = False, ) -> Callable[[], Any]: """Prepare the query so a worker can recheck authority after replay.""" try: @@ -3169,7 +3171,7 @@ def _prepare_query_state( def invoke() -> Any: try: return _invoke_workflow_handler( - handler, list(args) if args is not None else [], workflow_cls, method_name, allow_async=allow_async, + handler, list(args) if args is not None else [], workflow_cls, method_name, allow_async=True, ) except QueryFailed: raise @@ -3200,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, @@ -3208,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( @@ -3334,8 +3341,41 @@ def validate_update( prepare_local_activities: bool = False, prepare_local_activity_groups: bool = False, local_activity_cancellation_policies: tuple[str, ...] = (), - allow_async: bool = False, ) -> 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, + 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], + *, + 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, ...] = (), +) -> Callable[[], Any]: """Replay state and invoke only the declared pre-accept update validator. The replayed instance is discarded after validation. This helper does not @@ -3388,20 +3428,23 @@ def validate_update( if validator is None: raise UpdateValidationFailed(f"update validator {update_name!r} is not available") - try: - result = _invoke_workflow_handler(validator, list(args), workflow_cls, validator_name, allow_async=allow_async) - if result is False: - raise ValueError(f"update validator {update_name!r} returned false") - return result - except UpdateRejected: - raise - except Exception as exc: - errors = exc.errors if isinstance(exc, InvalidArgument) else None - raise UpdateRejected( - str(exc) or f"update validator {update_name!r} rejected the request", - exception_type=type(exc).__name__, - validation_errors=errors, - ) from exc + def invoke() -> Any: + try: + 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 + except UpdateRejected: + raise + except Exception as exc: + errors = exc.errors if isinstance(exc, InvalidArgument) else None + raise UpdateRejected( + str(exc) or f"update validator {update_name!r} rejected the request", + exception_type=type(exc).__name__, + validation_errors=errors, + ) from exc + + return invoke def _accepted_update_payload( 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 index ddf8568..27ad5e1 100644 --- a/tests/test_workflow_handler_validation.py +++ b/tests/test_workflow_handler_validation.py @@ -1,7 +1,7 @@ from __future__ import annotations from collections.abc import AsyncIterator, Generator -from functools import wraps +from functools import partial, wraps from typing import Any import pytest @@ -50,17 +50,19 @@ 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"): - if entry == "definition": - workflow.defn(name="handler-validation")(definition) - elif entry == "worker": - Worker(Client("http://127.0.0.1:1"), task_queue="unused", workflows=[definition]) - elif entry == "environment": - WorkflowEnvironment().register_workflow(definition) - elif entry == "replayer": - workflow.Replayer(workflows=[definition]) - else: - workflow.replay(definition, [], []) + invoke() @pytest.mark.parametrize("shape", ["coroutine", "async_generator", "generator"]) @@ -71,8 +73,10 @@ def test_wrapped_handlers_are_validated(shape: str) -> None: 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"): - workflow.defn(name="handler-validation")(_definition("signal", receive, False)) + decorate(definition) @pytest.mark.parametrize("kind", ["signal", "query", "update", "update_validator"]) From c9ed469c19972e6854cf65248753310fe9c7a3b5 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Tue, 20 Oct 2026 18:12:21 +0000 Subject: [PATCH 4/4] refactor: separate validator replay options and invocation --- src/durable_workflow/worker.py | 19 ++++--- src/durable_workflow/workflow.py | 92 +++++++++++++++++++------------- 2 files changed, 65 insertions(+), 46 deletions(-) diff --git a/src/durable_workflow/worker.py b/src/durable_workflow/worker.py index c01226d..b6a734e 100644 --- a/src/durable_workflow/worker.py +++ b/src/durable_workflow/worker.py @@ -110,6 +110,7 @@ UpsertMemo, _prepare_query_state, _prepare_update_validation, + _UpdateValidationReplayOptions, _validate_workflow_entrypoint, _workflow_handler_registries, apply_update, @@ -3873,14 +3874,16 @@ async def _run_update_validation_task(self, task: dict[str, Any]) -> str: 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): diff --git a/src/durable_workflow/workflow.py b/src/durable_workflow/workflow.py index 6cc4a6f..d47e0e9 100644 --- a/src/durable_workflow/workflow.py +++ b/src/durable_workflow/workflow.py @@ -3326,6 +3326,18 @@ def apply_update( 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]], @@ -3347,11 +3359,13 @@ def validate_update( result = _prepare_update_validation( workflow_cls, history_events, start_input, update_name, args, - 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, + 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 {} @@ -3367,14 +3381,7 @@ def _prepare_update_validation( update_name: str, args: list[Any], *, - 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, ...] = (), + replay_options: _UpdateValidationReplayOptions, ) -> Callable[[], Any]: """Replay state and invoke only the declared pre-accept update validator. @@ -3382,7 +3389,7 @@ def _prepare_update_validation( 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: @@ -3390,14 +3397,14 @@ def _prepare_update_validation( 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( @@ -3428,23 +3435,32 @@ def _prepare_update_validation( if validator is None: raise UpdateValidationFailed(f"update validator {update_name!r} is not available") - def invoke() -> Any: - try: - 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 - except UpdateRejected: - raise - except Exception as exc: - errors = exc.errors if isinstance(exc, InvalidArgument) else None - raise UpdateRejected( - str(exc) or f"update validator {update_name!r} rejected the request", - exception_type=type(exc).__name__, - validation_errors=errors, - ) from exc + return lambda: _invoke_update_validator(validator, args, workflow_cls, validator_name, update_name) - return invoke + +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 = _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 + except UpdateRejected: + raise + except Exception as exc: + errors = exc.errors if isinstance(exc, InvalidArgument) else None + raise UpdateRejected( + str(exc) or f"update validator {update_name!r} rejected the request", + exception_type=type(exc).__name__, + validation_errors=errors, + ) from exc def _accepted_update_payload(