diff --git a/docs/sdk-reference.md b/docs/sdk-reference.md index e408ead..94e2198 100644 --- a/docs/sdk-reference.md +++ b/docs/sdk-reference.md @@ -405,6 +405,12 @@ language, target service language, operation name, request payload, service-call id, response or failure surface, and optional artifact metadata to the recorded result. +Workflow entry points use synchronous `def run(...)` methods, either returning +a result directly or yielding durable commands. `async def` and async-generator +entry points are rejected with `InvalidWorkflowDefinition` during definition or +worker registration. Async activity handlers remain supported. Direct replay +also rejects async entry points and awaitable results. + ## Workflow signals, queries, and updates Signals mutate workflow state during replay: diff --git a/pyproject.toml b/pyproject.toml index 4463e1e..17b84af 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "durable-workflow" -version = "2.5.1" +version = "2.5.2" description = "Python client and worker SDK for Durable Workflow Cloud and self-hosted Server" readme = "README.md" requires-python = ">=3.10" @@ -71,8 +71,8 @@ durable-workflow-replay-conformance = "durable_workflow.replay_conformance:main" durable-workflow-workflow-updates-conformance = "durable_workflow.workflow_updates_conformance:main" [tool.durable-workflow] -product-train = "2.5.1" -registry-version = "2.5.1" +product-train = "2.5.2" +registry-version = "2.5.2" supported-server-versions = "2.5.16" worker-protocol-version = "1.19" control-plane-version = "2" diff --git a/src/durable_workflow/__init__.py b/src/durable_workflow/__init__.py index ecde0b5..9a6b909 100644 --- a/src/durable_workflow/__init__.py +++ b/src/durable_workflow/__init__.py @@ -81,6 +81,7 @@ ExternalPayloadUnavailable, ExternalPayloadUnsupported, InvalidArgument, + InvalidWorkflowDefinition, NamespaceNotFound, NexusOperationFailed, NonDeterministicReplayError, @@ -369,6 +370,7 @@ "NexusOperationResult", "NexusServiceCall", "NonDeterministicReplayError", + "InvalidWorkflowDefinition", "NoopMetrics", "QueryFailed", "RuntimeCapabilityUnsupported", diff --git a/src/durable_workflow/errors.py b/src/durable_workflow/errors.py index ed9af6e..031adbe 100644 --- a/src/durable_workflow/errors.py +++ b/src/durable_workflow/errors.py @@ -27,6 +27,10 @@ class DurableWorkflowError(Exception): """Base class for every exception raised by the SDK.""" +class InvalidWorkflowDefinition(DurableWorkflowError, ValueError): + """Workflow authoring uses an unsupported entry point or result.""" + + class DurableOperationCancelled(DurableWorkflowError): """A durable selection member was explicitly cancelled.""" diff --git a/src/durable_workflow/testing.py b/src/durable_workflow/testing.py index 7226a0b..f3229b9 100644 --- a/src/durable_workflow/testing.py +++ b/src/durable_workflow/testing.py @@ -51,6 +51,7 @@ def test_my_workflow(): StartChildWorkflow, StartTimer, UpsertSearchAttributes, + _validate_workflow_entrypoint, registry, replay, ) @@ -134,6 +135,7 @@ def register_workflow(self, workflow_cls: type) -> None: f"{workflow_cls!r} is not a workflow class; " "decorate it with @workflow.defn(name=...) first." ) + _validate_workflow_entrypoint(workflow_cls) self._registered_workflows[name] = workflow_cls def signal( diff --git a/src/durable_workflow/worker.py b/src/durable_workflow/worker.py index 4336802..291bf56 100644 --- a/src/durable_workflow/worker.py +++ b/src/durable_workflow/worker.py @@ -104,6 +104,7 @@ ScheduleActivity, StartChildWorkflow, UpsertMemo, + _validate_workflow_entrypoint, apply_update, commands_to_server_commands, query_state, @@ -1023,7 +1024,10 @@ def __init__( ) -> None: self.client = client self.task_queue = task_queue - self.workflows = {_workflow_name(w): w for w in workflows} + self.workflows: dict[str, type] = {} + for definition in workflows: + _validate_workflow_entrypoint(definition) + self.workflows[_workflow_name(definition)] = definition self.workflow_definition_fingerprints = { workflow_type: _workflow_definition_fingerprint(workflow_cls) for workflow_type, workflow_cls in self.workflows.items() diff --git a/src/durable_workflow/workflow.py b/src/durable_workflow/workflow.py index 8a0b2f7..1234f10 100644 --- a/src/durable_workflow/workflow.py +++ b/src/durable_workflow/workflow.py @@ -21,6 +21,7 @@ import base64 import contextlib import hashlib +import inspect import logging import math import random @@ -58,6 +59,7 @@ ChildWorkflowFailed, ChildWorkflowTerminated, DurableOperationCancelled, + InvalidWorkflowDefinition, NexusOperationFailed, NonDeterministicReplayError, QueryFailed, @@ -97,6 +99,31 @@ def _authored_scope_cleanup() -> dict[str, str] | None: MESSAGE_STREAM_MAX_BATCH = 100 +def _validate_workflow_entrypoint(cls: type) -> None: + entrypoint = getattr(cls, "run", None) + if not callable(entrypoint): + return + targets = [inspect.unwrap(entrypoint)] + if not inspect.isroutine(targets[0]): + targets.append(inspect.unwrap(targets[0].__call__)) + if any(inspect.iscoroutinefunction(target) or inspect.isasyncgenfunction(target) for target in targets): + name = getattr(cls, "__workflow_name__", cls.__name__) + raise InvalidWorkflowDefinition( + f"unsupported_workflow_entrypoint: workflow {name!r} ({cls.__qualname__}.run) " + "must use a synchronous function or generator; yield durable commands instead of async def." + ) + + +def _validate_workflow_result(value: Any, cls: type) -> None: + if inspect.isawaitable(value) or inspect.isasyncgen(value): + if inspect.iscoroutine(value): + value.close() + raise InvalidWorkflowDefinition( + f"unsupported_workflow_result: {cls.__qualname__}.run returned an async value; " + "use a synchronous result or yield durable commands." + ) + + def defn(*, name: str) -> Callable[[type[_WorkflowT]], type[_WorkflowT]]: """Register a class as a workflow type under a language-neutral name. @@ -108,6 +135,7 @@ def defn(*, name: str) -> Callable[[type[_WorkflowT]], type[_WorkflowT]]: def wrap(cls: type[_WorkflowT]) -> type[_WorkflowT]: cls.__workflow_name__ = name # type: ignore[attr-defined] + _validate_workflow_entrypoint(cls) signals: dict[str, str] = {} queries: dict[str, str] = {} updates: dict[str, str] = {} @@ -4037,6 +4065,7 @@ def _replay_state( allow_cancellation_scope_authoring: bool = False, allow_cancellation_scope_delivery: bool = False, ) -> _ReplayState: + _validate_workflow_entrypoint(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: @@ -5172,6 +5201,7 @@ def _same_logical_condition_wait(opened: Mapping[str, Any], cmd: WaitCondition) except Exception as exc: return _state([_fail_workflow_from_exception(exc)]) if not hasattr(gen, "__next__"): + _validate_workflow_result(gen, workflow_cls) if isinstance(gen, ContinueAsNew): _assert_no_unconsumed_history("continue as new") return _state([gen]) @@ -6032,6 +6062,7 @@ def _scope_call_membership(command: Any) -> str: return next(iter(memberships)) def _terminal_state(value: Any, *, include_pending: bool) -> _ReplayState: + _validate_workflow_result(value, workflow_cls) _apply_due_receivers() _consume_terminal_condition_reopens() commands = list(pending) if include_pending else [] @@ -6784,10 +6815,10 @@ def _advance_selection_clock(base: int, size: int, failure: BaseException | None try: return _terminal_state(stop.value, include_pending=True) except Exception as exc: - if isinstance(exc, NonDeterministicReplayError): + if isinstance(exc, (NonDeterministicReplayError, InvalidWorkflowDefinition)): raise return _state(pending + [_fail_workflow_from_exception(exc)]) - except NonDeterministicReplayError: + except (NonDeterministicReplayError, InvalidWorkflowDefinition): raise except LocalActivityExecutionAborted: raise diff --git a/tests/fixtures/replay_regressions/unsupported-async-workflow-entrypoint.json b/tests/fixtures/replay_regressions/unsupported-async-workflow-entrypoint.json new file mode 100644 index 0000000..7a7ab88 --- /dev/null +++ b/tests/fixtures/replay_regressions/unsupported-async-workflow-entrypoint.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-entrypoint", + "protocol_version": "1.19", + "bindings": ["python"], + "workflow": {"type": "tests.replay.unsupported-async", "input": [], "payload_codec": "avro"}, + "history": [{"event_type": "WorkflowStarted", "payload": {}}], + "expected": {"error": "unsupported_workflow_entrypoint"} +} diff --git a/tests/test_replay_regression_corpus.py b/tests/test_replay_regression_corpus.py index 371c309..2f1713f 100644 --- a/tests/test_replay_regression_corpus.py +++ b/tests/test_replay_regression_corpus.py @@ -251,7 +251,15 @@ def run(self, ctx: WorkflowContext): # type: ignore[no-untyped-def] return (yield ctx.start_child_workflow("child", [])) +class UnsupportedAsyncWorkflow: + __workflow_name__ = "tests.replay.unsupported-async" + + async def run(self, ctx: WorkflowContext) -> str: + return "body was never executed" + + WORKFLOWS = [ + UnsupportedAsyncWorkflow, ChildPolicyAuthorWorkflow, ColdReplacementSatisfiedConditionWorkflow, CooperativeReopenedConditionCleanupWorkflow, diff --git a/tests/test_workflow_entrypoint_validation.py b/tests/test_workflow_entrypoint_validation.py new file mode 100644 index 0000000..8c60de3 --- /dev/null +++ b/tests/test_workflow_entrypoint_validation.py @@ -0,0 +1,116 @@ +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.testing import WorkflowEnvironment + + +class CoroutineWorkflow: + async def run(self, ctx: Any) -> str: + return "never executed" + + +class AsyncGeneratorWorkflow: + async def run(self, ctx: Any) -> AsyncIterator[str]: + yield "unsupported" + + +class AsyncCallable: + async def __call__(self, ctx: Any) -> str: + return "unsupported" + + +class CallableWorkflow: + run = AsyncCallable() + + +class PartialWorkflow: + run = staticmethod(partial(CoroutineWorkflow.run, object())) + + +ASYNC_DEFINITIONS = [CoroutineWorkflow, AsyncGeneratorWorkflow, CallableWorkflow, PartialWorkflow] + + +@pytest.mark.parametrize("definition", ASYNC_DEFINITIONS) +def test_definition_rejects_async_entrypoints(definition: type) -> None: + with pytest.raises(ValueError, match="unsupported_workflow_entrypoint.*synchronous"): + workflow.defn(name="invalid-async")(definition) + + +@pytest.mark.parametrize("definition", ASYNC_DEFINITIONS) +def test_worker_rejects_undecorated_async_entrypoints_before_network(definition: type) -> None: + with pytest.raises(ValueError, match=definition.__name__): + Worker(Client("http://127.0.0.1:1"), task_queue="unused", workflows=[definition]) + + +@pytest.mark.parametrize("definition", ASYNC_DEFINITIONS) +def test_direct_replay_rejects_async_without_creating_a_coroutine(definition: type) -> None: + with pytest.raises(ValueError, match="unsupported_workflow_entrypoint"): + workflow.replay(definition, [], []) + + +def test_test_environment_revalidates_entrypoint_after_decoration() -> None: + @workflow.defn(name="changed-entrypoint") + class ChangedWorkflow: + def run(self, ctx: Any) -> str: + return "valid" + + ChangedWorkflow.run = CoroutineWorkflow.run # type: ignore[assignment] + with pytest.raises(ValueError, match="changed-entrypoint"): + WorkflowEnvironment().register_workflow(ChangedWorkflow) + + +def test_wrapped_and_inherited_async_entrypoints_are_rejected() -> None: + class InheritedWorkflow(CoroutineWorkflow): + pass + + class WrappedWorkflow: + @wraps(CoroutineWorkflow.run) + def run(self, ctx: Any) -> Any: + return CoroutineWorkflow.run(self, ctx) + + for definition in [InheritedWorkflow, WrappedWorkflow]: + with pytest.raises(ValueError, match="unsupported_workflow_entrypoint"): + workflow.replay(definition, [], []) + + +@pytest.mark.parametrize("generator", [False, True]) +def test_runtime_awaitable_result_is_not_completed(generator: bool) -> None: + class AwaitableResultWorkflow: + def run(self, ctx: Any) -> Any: + return CoroutineWorkflow.run(self, ctx) + + class GeneratorAwaitableResultWorkflow: + def run(self, ctx: Any) -> Generator[Any, Any, Any]: + if False: + yield None + return CoroutineWorkflow.run(self, ctx) + + definition = GeneratorAwaitableResultWorkflow if generator else AwaitableResultWorkflow + with pytest.raises(ValueError, match="unsupported_workflow_result"): + workflow.replay(definition, [], []) + + +def test_synchronous_and_generator_workflows_remain_supported() -> None: + @workflow.defn(name="valid-sync") + class SynchronousWorkflow: + def run(self, ctx: Any) -> str: + return "done" + + @workflow.defn(name="valid-generator") + class GeneratorWorkflow: + def run(self, ctx: Any) -> Generator[Any, Any, str]: + if False: + yield None + return "done" + + Worker(Client("http://127.0.0.1:1"), task_queue="unused", workflows=[SynchronousWorkflow, GeneratorWorkflow]) + environment = WorkflowEnvironment() + for definition in [SynchronousWorkflow, GeneratorWorkflow]: + assert environment.execute_workflow(definition) == "done" + assert workflow.replay(definition, [], []).commands[0].result == "done" # type: ignore[union-attr]