diff --git a/cadence/_internal/activity/_definition.py b/cadence/_internal/activity/_definition.py index 70d7424a..7c4623ab 100644 --- a/cadence/_internal/activity/_definition.py +++ b/cadence/_internal/activity/_definition.py @@ -21,7 +21,12 @@ ) from cadence._internal.fn_signature import FnSignature -from cadence.workflow import ActivityOptions, WorkflowContext, execute_activity +from cadence.workflow import ( + ActivityOptions, + WorkflowContext, + execute_activity, + WorkflowFuture, +) T = TypeVar("T") P = ParamSpec("P") @@ -80,9 +85,9 @@ def with_options(self, **kwargs: Unpack[ActivityOptions]) -> Self: res._execution_options = new_opts return res - async def execute(self, *args: P.args, **kwargs: P.kwargs) -> R: + def execute(self, *args: P.args, **kwargs: P.kwargs) -> WorkflowFuture[R]: result_type = cast(type[R], self._signature.return_type) - return await execute_activity( + return execute_activity( self._name, result_type, *self._signature.params_from_call(args, kwargs), diff --git a/cadence/_internal/workflow/child_workflow_future.py b/cadence/_internal/workflow/child_workflow_future.py index 71df0d9a..86d5f811 100644 --- a/cadence/_internal/workflow/child_workflow_future.py +++ b/cadence/_internal/workflow/child_workflow_future.py @@ -1,44 +1,46 @@ import asyncio -from typing import Any, Awaitable, Generator, Type, cast - -from cadence.api.v1.common_pb2 import Payload -from cadence.data_converter import DataConverter -from cadence.workflow import ResultType, WorkflowContext - - -class ChildWorkflowFutureImpl(Awaitable[ResultType]): +from typing import Any + +from cadence._internal.workflow.statemachine.decision_state_machine import ( + CancelFn, + ConverterFn, + DecisionFuture, +) +from cadence.api.v1.common_pb2 import WorkflowExecution +from cadence.workflow import ( + ChildWorkflowFuture, + ResultType, + WorkflowContext, + WorkflowFuture, +) + + +class ChildWorkflowFutureImpl( + DecisionFuture[ResultType], ChildWorkflowFuture[ResultType] +): def __init__( self, + loop: asyncio.AbstractEventLoop | None = None, + request_cancel: CancelFn | None = None, + converter: ConverterFn[ResultType] | None = None, + *, workflow_id: str, - run_id: str, - result_future: "asyncio.Future[Payload]", - result_type: Type[ResultType], - data_converter: DataConverter, + execution: WorkflowFuture[WorkflowExecution], ) -> None: + super().__init__(loop, request_cancel, converter) self._workflow_id = workflow_id - self._run_id = run_id - self._result_future = result_future - self._result_type = result_type - self._data_converter = data_converter + self._execution = execution @property def workflow_id(self) -> str: return self._workflow_id @property - def run_id(self) -> str: - return self._run_id + def execution(self) -> WorkflowFuture[WorkflowExecution]: + return self._execution - def cancel(self) -> bool: - """Request cancellation of the child workflow.""" - return self._result_future.cancel() - - async def signal(self, signal_name: str, *args: Any) -> None: + def signal(self, signal_name: str, *args: Any) -> WorkflowFuture[None]: """Send a signal to this child workflow.""" - ctx = WorkflowContext.get() - await ctx.signal_child_workflow(self._workflow_id, signal_name, *args) - - def __await__(self) -> Generator[Any, None, ResultType]: - payload: Payload = yield from self._result_future.__await__() - result = self._data_converter.from_data(payload, [self._result_type])[0] - return cast(ResultType, result) + return WorkflowContext.get().signal_child_workflow( + self._workflow_id, signal_name, *args + ) diff --git a/cadence/_internal/workflow/context.py b/cadence/_internal/workflow/context.py index 300d9446..e24d48f0 100644 --- a/cadence/_internal/workflow/context.py +++ b/cadence/_internal/workflow/context.py @@ -7,7 +7,6 @@ from math import ceil from typing import Iterator, Optional, Any, Unpack, Type, cast, Callable, Mapping -from cadence._internal.workflow.child_workflow_future import ChildWorkflowFutureImpl from cadence._internal.workflow.deterministic_event_loop import DeterministicEventLoop from cadence._internal.workflow.deterministic_event_loop import FatalDecisionError from cadence._internal.workflow.memo import memo_to_proto @@ -51,6 +50,7 @@ SearchAttributeType, WorkflowCancellationInfo, WorkflowContext, + WorkflowFuture, WorkflowInfo, DEFAULT_VERSION, ) @@ -83,13 +83,13 @@ def info(self) -> WorkflowInfo: def data_converter(self) -> DataConverter: return self.info().data_converter - async def execute_activity( + def execute_activity( self, activity: str, result_type: Type[ResultType], *args: Any, **kwargs: Unpack[ActivityOptions], - ) -> ResultType: + ) -> WorkflowFuture[ResultType]: opts: ActivityOptions = {**_DEFAULT_ACTIVITY_OPTIONS, **kwargs} if "schedule_to_close_timeout" not in opts and ( "schedule_to_start_timeout" not in opts @@ -137,26 +137,11 @@ async def execute_activity( ) set_header(schedule_attributes, self._context_propagators) - future = self._decision_manager.schedule_activity(schedule_attributes) - result_payload = await future - - result = self.data_converter().from_data(result_payload, [result_type])[0] - - return cast(ResultType, result) - - async def execute_child_workflow( - self, - workflow_type: str, - result_type: Type[ResultType], - *args: Any, - **kwargs: Unpack[ChildWorkflowOptions], - ) -> ResultType: - future = await self.start_child_workflow( - workflow_type, result_type, *args, **kwargs + return self._decision_manager.schedule_activity( + schedule_attributes, self.data_converter(), result_type ) - return await future - async def start_child_workflow( + def execute_child_workflow( self, workflow_type: str, result_type: Type[ResultType], @@ -166,19 +151,11 @@ async def start_child_workflow( schedule_attributes = self._build_child_workflow_attrs( workflow_type, *args, **kwargs ) - execution_future, result_future = ( - self._decision_manager.schedule_child_workflow( - schedule_attributes, - parent_workflow_run_id=self._info.workflow_run_id, - ) - ) - workflow_execution = await execution_future - return ChildWorkflowFutureImpl( - workflow_id=workflow_execution.workflow_id, - run_id=workflow_execution.run_id, - result_future=result_future, - result_type=result_type, - data_converter=self.data_converter(), + return self._decision_manager.schedule_child_workflow( + schedule_attributes, + self.data_converter(), + result_type, + parent_workflow_run_id=self._info.workflow_run_id, ) def _build_child_workflow_attrs( @@ -244,39 +221,40 @@ def _build_child_workflow_attrs( return schedule_attributes - async def signal_child_workflow( + def signal_child_workflow( self, child_workflow_id: str, signal_name: str, *args: Any, - ) -> None: + ) -> WorkflowFuture[None]: if not child_workflow_id: raise ValueError("child_workflow_id must not be empty") - await self._signal_workflow( + return self._signal_workflow( child_workflow_id, signal_name, args, child_workflow_only=True ) - async def signal_external_workflow( + def signal_external_workflow( self, workflow_id: str, signal_name: str, *args: Any, run_id: str = "", domain: str = "", - ) -> None: + ) -> WorkflowFuture[None]: if not workflow_id: raise ValueError("workflow_id must not be empty") - await self._signal_workflow( + return self._signal_workflow( workflow_id, signal_name, args, run_id=run_id, domain=domain ) - async def start_timer(self, duration: timedelta): + def start_timer(self, duration: timedelta) -> WorkflowFuture[None]: if duration.total_seconds() <= 0: # shortcut - return + res = WorkflowFuture[None]() + res.set_result(None) + return res attributes = StartTimerDecisionAttributes() attributes.start_to_fire_timeout.FromTimedelta(duration) - future = self._decision_manager.start_timer(attributes) - await future + return self._decision_manager.start_timer(attributes) def set_replay_mode(self, replay: bool) -> None: """Set whether the workflow is currently in replay mode.""" @@ -413,9 +391,9 @@ def get_replay_current_time(self) -> Optional[datetime]: """Get the current replay timestamp.""" return self._replay_current_time - async def wait_condition(self, predicate: Callable[[], bool]) -> None: + def wait_condition(self, predicate: Callable[[], bool]) -> WorkflowFuture[None]: loop = cast(DeterministicEventLoop, get_running_loop()) - await loop.create_waiter(predicate) + return loop.create_waiter(predicate) def request_cancel( self, attrs: WorkflowExecutionCancelRequestedEventAttributes @@ -433,7 +411,7 @@ def is_cancel_requested(self) -> bool: def inject_propagated_headers(self) -> dict[str, bytes]: return inject_headers(self._context_propagators) - async def _signal_workflow( + def _signal_workflow( self, workflow_id: str, signal_name: str, @@ -442,7 +420,7 @@ async def _signal_workflow( run_id: str = "", domain: str = "", child_workflow_only: bool = False, - ) -> None: + ) -> WorkflowFuture[None]: if not signal_name: raise ValueError("signal_name must not be empty") attrs = SignalExternalWorkflowExecutionDecisionAttributes( @@ -455,7 +433,7 @@ async def _signal_workflow( input=self.data_converter().to_data(list(args)), child_workflow_only=child_workflow_only, ) - await self._decision_manager.signal_external_workflow(attrs) + return self._decision_manager.signal_external_workflow(attrs) def _round_to_nearest_second(delta: timedelta) -> timedelta: diff --git a/cadence/_internal/workflow/statemachine/activity_state_machine.py b/cadence/_internal/workflow/statemachine/activity_state_machine.py index 684b61a3..c88f37e1 100644 --- a/cadence/_internal/workflow/statemachine/activity_state_machine.py +++ b/cadence/_internal/workflow/statemachine/activity_state_machine.py @@ -1,3 +1,5 @@ +from typing import Any + from cadence._internal.workflow.statemachine.decision_state_machine import ( DecisionState, DecisionType, @@ -10,7 +12,6 @@ record_immediate_cancel, ) from cadence.api.v1 import decision, history -from cadence.api.v1.common_pb2 import Payload from cadence.api.v1.workflow_pb2 import TimeoutType from cadence.error import ActivityFailure @@ -19,12 +20,12 @@ class ActivityStateMachine(BaseDecisionStateMachine): request: decision.ScheduleActivityTaskDecisionAttributes - completed: DecisionFuture[Payload] + completed: DecisionFuture[Any] def __init__( self, request: decision.ScheduleActivityTaskDecisionAttributes, - completed: DecisionFuture[Payload], + completed: DecisionFuture[Any], ) -> None: super().__init__() self.request = request @@ -82,7 +83,7 @@ def handle_completed( self, event: history.ActivityTaskCompletedEventAttributes ) -> None: self._transition(DecisionState.COMPLETED) - self._resolve(self.completed, result=event.result) + self.completed.complete_with_payload(event.result) @activity_events.event() def handle_failed(self, event: history.ActivityTaskFailedEventAttributes) -> None: diff --git a/cadence/_internal/workflow/statemachine/child_workflow_execution_state_machine.py b/cadence/_internal/workflow/statemachine/child_workflow_execution_state_machine.py index 2ba3a762..9c36edae 100644 --- a/cadence/_internal/workflow/statemachine/child_workflow_execution_state_machine.py +++ b/cadence/_internal/workflow/statemachine/child_workflow_execution_state_machine.py @@ -1,5 +1,7 @@ from __future__ import annotations +from typing import Any + from cadence._internal.workflow.statemachine.decision_state_machine import ( BaseDecisionStateMachine, DecisionFuture, @@ -12,7 +14,7 @@ record_immediate_cancel, ) from cadence.api.v1 import decision, history -from cadence.api.v1.common_pb2 import Payload, WorkflowExecution +from cadence.api.v1.common_pb2 import WorkflowExecution from cadence.error import ( ChildWorkflowExecutionCanceled, ChildWorkflowExecutionFailed, @@ -29,14 +31,14 @@ class ChildWorkflowExecutionStateMachine(BaseDecisionStateMachine): request: decision.StartChildWorkflowExecutionDecisionAttributes execution: DecisionFuture[WorkflowExecution] - result: DecisionFuture[Payload] + result: DecisionFuture[Any] _run_id: str | None def __init__( self, request: decision.StartChildWorkflowExecutionDecisionAttributes, execution: DecisionFuture[WorkflowExecution], - result: DecisionFuture[Payload], + result: DecisionFuture[Any], ) -> None: super().__init__() self.request = request @@ -130,7 +132,7 @@ def handle_completed( self, event: history.ChildWorkflowExecutionCompletedEventAttributes ) -> None: self._transition(DecisionState.COMPLETED) - self._resolve(self.result, result=event.result) + self.result.complete_with_payload(event.result) @child_workflow_events.event() def handle_failed( diff --git a/cadence/_internal/workflow/statemachine/decision_manager.py b/cadence/_internal/workflow/statemachine/decision_manager.py index 5c8b6d15..103d2a6e 100644 --- a/cadence/_internal/workflow/statemachine/decision_manager.py +++ b/cadence/_internal/workflow/statemachine/decision_manager.py @@ -3,8 +3,9 @@ from collections import OrderedDict from contextlib import contextmanager from dataclasses import dataclass, field -from typing import Any, Dict, Type, ClassVar, List, Iterator +from typing import Any, Dict, Type, ClassVar, List, Iterator, cast +from cadence._internal.workflow.child_workflow_future import ChildWorkflowFutureImpl from cadence._internal.workflow.statemachine.activity_state_machine import ( activity_events, ActivityStateMachine, @@ -19,6 +20,7 @@ ) from cadence._internal.workflow.statemachine.decision_state_machine import ( CancelFn, + ConverterFn, DecisionId, DecisionState, DecisionStateMachine, @@ -61,6 +63,8 @@ ) from cadence.api.v1 import decision, history from cadence.api.v1.common_pb2 import Payload, WorkflowExecution +from cadence.data_converter import DataConverter +from cadence.workflow import ChildWorkflowFuture, ResultType, WorkflowFuture logger = logging.getLogger(__name__) @@ -72,6 +76,18 @@ def _consume_future_exception(future: asyncio.Future[Any]) -> None: future.exception() +def _payload_converter( + decision_id: DecisionId, converter: DataConverter, result_type: Type[T] +) -> ConverterFn[T]: + def convert(payload: Payload) -> T: + try: + return cast(T, converter.from_data(payload, [result_type])[0]) + except Exception as e: + raise RuntimeError(f"Failed to convert result for {decision_id}") from e + + return convert + + @dataclass(frozen=True) class EventDispatch: decision_type: DecisionType @@ -141,13 +157,18 @@ def __init__(self, event_loop: asyncio.AbstractEventLoop): # ----- Activity API ----- def schedule_activity( - self, attrs: decision.ScheduleActivityTaskDecisionAttributes - ) -> asyncio.Future[Payload]: + self, + attrs: decision.ScheduleActivityTaskDecisionAttributes, + converter: DataConverter, + result_type: Type[ResultType], + ) -> WorkflowFuture[ResultType]: attrs.activity_id = self._next_id() if self._replaying: self._determinism_tracker.validate_action(attrs) decision_id = DecisionId(DecisionType.ACTIVITY, attrs.activity_id) - future: DecisionFuture[Payload] = self._create_future(decision_id) + future: DecisionFuture[ResultType] = self._create_future( + decision_id, _payload_converter(decision_id, converter, result_type) + ) machine = ActivityStateMachine(attrs, future) self._add_state_machine(machine) @@ -157,7 +178,7 @@ def schedule_activity( def start_timer( self, attrs: decision.StartTimerDecisionAttributes - ) -> asyncio.Future[None]: + ) -> WorkflowFuture[None]: attrs.timer_id = self._next_id() if self._replaying: self._determinism_tracker.validate_action(attrs) @@ -172,9 +193,11 @@ def start_timer( def schedule_child_workflow( self, attrs: decision.StartChildWorkflowExecutionDecisionAttributes, + converter: DataConverter, + result_type: Type[ResultType], *, parent_workflow_run_id: str, - ) -> tuple[DecisionFuture[WorkflowExecution], DecisionFuture[Payload]]: + ) -> ChildWorkflowFuture[ResultType]: if not attrs.workflow_id: attrs.workflow_id = f"{parent_workflow_run_id}_{self._next_id()}" @@ -183,17 +206,23 @@ def schedule_child_workflow( decision_id = DecisionId(DecisionType.CHILD_WORKFLOW, attrs.workflow_id) execution: DecisionFuture[WorkflowExecution] = self._create_future(decision_id) execution.add_done_callback(_consume_future_exception) - result: DecisionFuture[Payload] = self._create_future(decision_id) + result = ChildWorkflowFutureImpl[ResultType]( + self._event_loop, + self._create_cancel_callback(decision_id), + _payload_converter(decision_id, converter, result_type), + workflow_id=attrs.workflow_id, + execution=execution, + ) machine = ChildWorkflowExecutionStateMachine(attrs, execution, result) self._add_state_machine(machine) - return execution, result + return result # ----- Signal External Workflow API ----- def signal_external_workflow( self, attrs: decision.SignalExternalWorkflowExecutionDecisionAttributes, - ) -> asyncio.Future[None]: + ) -> WorkflowFuture[None]: signal_id = self._next_id() attrs.control = signal_id.encode("utf-8") if self._replaying: @@ -575,10 +604,13 @@ def collect_pending_decisions(self) -> List[decision.Decision]: return decisions - def _create_future(self, decision_id: DecisionId) -> DecisionFuture[T]: + def _create_future( + self, decision_id: DecisionId, converter: ConverterFn[T] | None = None + ) -> DecisionFuture[T]: return DecisionFuture[T]( self._event_loop, self._create_cancel_callback(decision_id), + converter, ) def _create_cancel_callback(self, decision_id: DecisionId) -> CancelFn: diff --git a/cadence/_internal/workflow/statemachine/decision_state_machine.py b/cadence/_internal/workflow/statemachine/decision_state_machine.py index c7210739..16629cba 100644 --- a/cadence/_internal/workflow/statemachine/decision_state_machine.py +++ b/cadence/_internal/workflow/statemachine/decision_state_machine.py @@ -8,6 +8,8 @@ from cadence.api.v1 import ( decision_pb2 as decision, ) +from cadence.api.v1.common_pb2 import Payload +from cadence.workflow import WorkflowFuture class DecisionState(Enum): @@ -106,18 +108,36 @@ def _resolve( T = TypeVar("T") CancelFn = Callable[[str | None], bool] +ConverterFn = Callable[[Payload], T] -class DecisionFuture(asyncio.Future[T]): +class DecisionFuture(WorkflowFuture[T]): def __init__( self, loop: asyncio.AbstractEventLoop | None = None, request_cancel: CancelFn | None = None, + converter: ConverterFn[T] | None = None, ) -> None: super().__init__(loop=loop) if request_cancel is None: request_cancel = self.force_cancel self._request_cancel = request_cancel + self._converter = converter + + def complete_with_payload(self, payload: Payload) -> None: + # Local cancellation may have already resolved this future; see BaseDecisionStateMachine._resolve. + if self.done(): + return + if self._converter: + try: + result = self._converter(payload) + except Exception as e: + self.set_exception(e) + else: + self.set_result(result) + else: + # type ignore because we don't know T, but it should be the caller's responsibility to only call this if T is Payload + self.set_result(payload) # type: ignore def force_cancel(self, message: str | None = None) -> bool: return super().cancel(message) diff --git a/cadence/_internal/workflow/waiter.py b/cadence/_internal/workflow/waiter.py index 9436f684..4febd408 100644 --- a/cadence/_internal/workflow/waiter.py +++ b/cadence/_internal/workflow/waiter.py @@ -1,37 +1,27 @@ -from asyncio import AbstractEventLoop, Future -from typing import Callable, Any, Generator, Awaitable +from asyncio import AbstractEventLoop +from typing import Callable +from cadence.workflow import WorkflowFuture -class Waiter(Awaitable[None]): - """Awaitable that resolves when ``predicate()`` becomes truthy.""" - __slots__ = ("_predicate", "_future") +class Waiter(WorkflowFuture[None]): + """Future that resolves when ``predicate()`` becomes truthy.""" + + __slots__ = ("_predicate",) def __init__(self, predicate: Callable[[], bool], loop: AbstractEventLoop) -> None: + super().__init__(loop=loop) self._predicate = predicate - self._future: Future[None] = loop.create_future() - - def __await__(self) -> Generator[Any, None, None]: - return self._future.__await__() - - def done(self) -> bool: - return self._future.done() - - def exception(self) -> BaseException | None: - return self._future.exception() - - def result(self) -> None: - self._future.result() def poll(self) -> bool: """Re-evaluate the predicate. Returns True when settled (and evictable).""" - if self._future.done(): + if self.done(): return True try: if self._predicate(): - self._future.set_result(None) + self.set_result(None) return True return False except BaseException as exc: - self._future.set_exception(exc) + self.set_exception(exc) return True diff --git a/cadence/activity.py b/cadence/activity.py index 2ad9e6d6..f2577cf1 100644 --- a/cadence/activity.py +++ b/cadence/activity.py @@ -32,7 +32,7 @@ SyncMethodImpl, ) from cadence._internal.fn_signature import FnSignature -from cadence.workflow import ActivityOptions +from cadence.workflow import ActivityOptions, WorkflowFuture @dataclass(frozen=True) @@ -199,7 +199,7 @@ def name(self) -> str: ... def with_options(self, **kwargs: Unpack[ActivityOptions]) -> Self: ... - async def execute(self, *args: P.args, **kwargs: P.kwargs) -> R: ... + def execute(self, *args: P.args, **kwargs: P.kwargs) -> WorkflowFuture[R]: ... class _SyncActivityDefinition(ActivityDefinition[P, R], Protocol): diff --git a/cadence/testing/_workflow_environment.py b/cadence/testing/_workflow_environment.py index 47b00f32..e48de2f4 100644 --- a/cadence/testing/_workflow_environment.py +++ b/cadence/testing/_workflow_environment.py @@ -52,10 +52,12 @@ from typing import ( Any, Callable, + Coroutine, Optional, Sequence, Tuple, Type, + TypeVar, Union, Unpack, cast, @@ -93,6 +95,7 @@ SearchAttributeType, WorkflowContext, WorkflowDefinition, + WorkflowFuture, WorkflowInfo, WorkflowInterceptorFactory, ) @@ -102,6 +105,47 @@ logger = logging.getLogger(__name__) +_F = TypeVar("_F", bound="Future[Any]") + + +def _drive_future(coro: Coroutine[Any, Any, Any], future: _F) -> _F: + """Run ``coro`` as a task and copy its outcome into ``future``. + + Cancelling ``future`` cancels the underlying task. + """ + task = future.get_loop().create_task(coro) + + def _copy(t: "asyncio.Task[Any]") -> None: + if future.done(): + return + if t.cancelled(): + future.cancel() + elif (exc := t.exception()) is not None: + future.set_exception(exc) + else: + future.set_result(t.result()) + + def _propagate_cancel(f: "Future[Any]") -> None: + if f.cancelled(): + task.cancel() + + task.add_done_callback(_copy) + future.add_done_callback(_propagate_cancel) + return future + + +def _resolved() -> WorkflowFuture[None]: + future = WorkflowFuture[None]() + future.set_result(None) + return future + + +def _future_from_coroutine( + coro: Coroutine[Any, Any, ResultType], +) -> WorkflowFuture[ResultType]: + return _drive_future(coro, WorkflowFuture[ResultType](loop=get_running_loop())) + + def _headers_after_proto_roundtrip(headers: Mapping[str, bytes]) -> dict[str, bytes]: """Round-trip headers through proto, matching production decision encoding.""" proto = header_from_dict(headers) @@ -213,31 +257,21 @@ def info(self) -> WorkflowInfo: def data_converter(self) -> DataConverter: return self._info.data_converter - async def execute_activity( + def execute_activity( self, activity: str, result_type: Type[ResultType], *args: Any, **kwargs: Unpack[ActivityOptions], - ) -> ResultType: + ) -> WorkflowFuture[ResultType]: outbound = self.inject_propagated_headers() - return await self._env._invoke_activity( - activity, result_type, args, self._info, outbound - ) - - async def execute_child_workflow( - self, - workflow_type: str, - result_type: Type[ResultType], - *args: Any, - **kwargs: Unpack[ChildWorkflowOptions], - ) -> ResultType: - future = await self.start_child_workflow( - workflow_type, result_type, *args, **kwargs + return _future_from_coroutine( + self._env._invoke_activity( + activity, result_type, args, self._info, outbound + ) ) - return await future - async def start_child_workflow( + def execute_child_workflow( self, workflow_type: str, result_type: Type[ResultType], @@ -245,7 +279,7 @@ async def start_child_workflow( **kwargs: Unpack[ChildWorkflowOptions], ) -> ChildWorkflowFuture[ResultType]: outbound = self.inject_propagated_headers() - return await self._env._start_child_workflow( + return self._env._execute_child_workflow( workflow_type, result_type, args, @@ -254,19 +288,17 @@ async def start_child_workflow( headers=outbound, ) - async def start_timer(self, duration: timedelta) -> None: + def start_timer(self, duration: timedelta) -> WorkflowFuture[None]: if duration.total_seconds() <= 0: - return None - # Suspend on a heap-backed future instead of advancing the clock inline. + return _resolved() + # Return a heap-backed future instead of advancing the clock inline. # The driver fires it (advancing the virtual clock to its deadline) once # the workflow has no other progress to make. See the module docstring. - future = self._env._schedule_timer(duration) - await future - return None + return self._env._schedule_timer(duration) - async def wait_condition(self, predicate: Callable[[], bool]) -> None: + def wait_condition(self, predicate: Callable[[], bool]) -> WorkflowFuture[None]: loop = cast(DeterministicEventLoop, get_running_loop()) - await loop.create_waiter(predicate) + return loop.create_waiter(predicate) def side_effect( self, @@ -331,30 +363,29 @@ def upsert_search_attributes( merged.update(attributes) self._info = replace(self._info, search_attributes=merged) - async def signal_child_workflow( + def signal_child_workflow( self, child_workflow_id: str, signal_name: str, *args: Any, - ) -> None: + ) -> WorkflowFuture[None]: """Signal a child workflow by its workflow id. Routing is identical to :meth:`signal_external_workflow`: the signal is buffered and delivered to the matching execution after the current - decision finishes. Note that child workflows started via - ``start_child_workflow`` run inline to completion, so they can only be - signaled while still running. + decision finishes. """ self._env._enqueue_signal(child_workflow_id, signal_name, args) + return _resolved() - async def signal_external_workflow( + def signal_external_workflow( self, workflow_id: str, signal_name: str, *args: Any, run_id: str = "", domain: str = "", - ) -> None: + ) -> WorkflowFuture[None]: """Signal another workflow running in this environment. The signal is buffered and delivered to the target execution after the @@ -363,6 +394,7 @@ async def signal_external_workflow( dropped, matching the behavior of signaling a closed workflow. """ self._env._enqueue_signal(workflow_id, signal_name, args) + return _resolved() def is_cancel_requested(self) -> bool: return False @@ -717,7 +749,7 @@ def __exit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None: # Internal helpers used by MockClient / contexts # ------------------------------------------------------------------ - def _schedule_timer(self, duration: timedelta) -> "Future[None]": + def _schedule_timer(self, duration: timedelta) -> WorkflowFuture[None]: """Register a timer on the currently-driven execution's heap. Returns a future that the driver resolves once the virtual clock @@ -727,7 +759,7 @@ def _schedule_timer(self, duration: timedelta) -> "Future[None]": if execution is None: raise RuntimeError("start_timer called outside of a workflow decision") loop = cast(DeterministicEventLoop, get_running_loop()) - future: "Future[None]" = loop.create_future() + future = WorkflowFuture[None](loop=loop) execution.add_timer(self._now + duration, future) return future @@ -908,7 +940,7 @@ async def _run_real_activity( result = await result return result - async def _start_child_workflow( + def _execute_child_workflow( self, workflow_type: str, result_type: Type[ResultType], @@ -935,19 +967,26 @@ async def _start_child_workflow( ) input_payload = self._data_converter.to_data(list(args)) - result_payload = await self._run_workflow_coro( - definition, child_info, input_payload, headers - ) - loop = cast(DeterministicEventLoop, get_running_loop()) - result_future: "Future[Payload]" = loop.create_future() - result_future.set_result(result_payload) - return ChildWorkflowFutureImpl( - workflow_id=child_id, - run_id=child_run_id, - result_future=result_future, - result_type=result_type, - data_converter=self._data_converter, + async def _run() -> ResultType: + result_payload = await self._run_workflow_coro( + definition, child_info, input_payload, headers + ) + return cast( + ResultType, + self._data_converter.from_data(result_payload, [result_type])[0], + ) + + loop = get_running_loop() + execution = WorkflowFuture[WorkflowExecution](loop=loop) + execution.set_result( + WorkflowExecution(workflow_id=child_id, run_id=child_run_id) + ) + return _drive_future( + _run(), + ChildWorkflowFutureImpl[ResultType]( + loop, workflow_id=child_id, execution=execution + ), ) async def _run_workflow_coro( diff --git a/cadence/workflow/__init__.py b/cadence/workflow/__init__.py index 8ad8d96b..8ab4472b 100644 --- a/cadence/workflow/__init__.py +++ b/cadence/workflow/__init__.py @@ -35,6 +35,7 @@ ResultType, RetryPolicy, SearchAttributeType, + WorkflowFuture, ) __all__ = [ @@ -53,6 +54,7 @@ "WorkflowDecorator", "WorkflowDefinition", "WorkflowDefinitionOptions", + "WorkflowFuture", "WorkflowInfo", "WorkflowInterceptor", "WorkflowInterceptorFactory", diff --git a/cadence/workflow/_api.py b/cadence/workflow/_api.py index 2ccd0609..efbcd7af 100644 --- a/cadence/workflow/_api.py +++ b/cadence/workflow/_api.py @@ -17,27 +17,28 @@ ChildWorkflowOptions, ResultType, SearchAttributeType, + WorkflowFuture, ) -async def execute_activity( +def execute_activity( activity: str, result_type: Type[ResultType], *args: Any, **kwargs: Unpack[ActivityOptions], -) -> ResultType: - return await WorkflowContext.get().execute_activity( +) -> WorkflowFuture[ResultType]: + return WorkflowContext.get().execute_activity( activity, result_type, *args, **kwargs ) -async def execute_child_workflow( +def execute_child_workflow( workflow_type: str, result_type: Type[ResultType], *args: Any, **kwargs: Unpack[ChildWorkflowOptions], -) -> ResultType: - return await WorkflowContext.get().execute_child_workflow( +) -> ChildWorkflowFuture[ResultType]: + return WorkflowContext.get().execute_child_workflow( workflow_type, result_type, *args, **kwargs ) @@ -47,19 +48,29 @@ async def start_child_workflow( result_type: Type[ResultType], *args: Any, **kwargs: Unpack[ChildWorkflowOptions], -) -> "ChildWorkflowFuture[ResultType]": - return await WorkflowContext.get().start_child_workflow( - workflow_type, result_type, *args, **kwargs - ) +) -> ChildWorkflowFuture[ResultType]: + """Start a child workflow and wait until it has started. - -async def signal_external_workflow( + Returns the same future as execute_child_workflow, with its execution resolved. + """ + future = execute_child_workflow(workflow_type, result_type, *args, **kwargs) + try: + await future.execution + except Exception: + # The handle fails with the same start error, but the caller never receives it. + if future.done() and not future.cancelled(): + future.exception() + raise + return future + + +def signal_external_workflow( workflow_id: str, signal_name: str, *args: Any, run_id: str = "", domain: str = "", -) -> None: +) -> WorkflowFuture[None]: """Send a signal to an external workflow execution. Args: @@ -69,23 +80,24 @@ async def signal_external_workflow( run_id: Target run ID. Empty string targets the currently running execution. domain: Target domain. Empty string defaults to the current workflow's domain. """ - await WorkflowContext.get().signal_external_workflow( + return WorkflowContext.get().signal_external_workflow( workflow_id, signal_name, *args, run_id=run_id, domain=domain ) -async def sleep(duration: timedelta) -> None: - return await WorkflowContext.get().start_timer(duration) +def sleep(duration: timedelta) -> WorkflowFuture[None]: + return WorkflowContext.get().start_timer(duration) -async def wait_condition(predicate: Callable[[], bool]) -> None: - """Block until predicate returns True. +def wait_condition(predicate: Callable[[], bool]) -> WorkflowFuture[None]: + """Return a future that resolves once predicate returns True. The predicate is re-evaluated after any workflow state change (signal delivery, activity completion, timer firing). - If the predicate is already True, returns immediately. + If the predicate is already True, the future is already resolved. + Cancelling the future stops the predicate from being evaluated. """ - await WorkflowContext.get().wait_condition(predicate) + return WorkflowContext.get().wait_condition(predicate) def side_effect( diff --git a/cadence/workflow/_context.py b/cadence/workflow/_context.py index 7a0943ee..3d6daa6e 100644 --- a/cadence/workflow/_context.py +++ b/cadence/workflow/_context.py @@ -17,6 +17,7 @@ ChildWorkflowOptions, ResultType, SearchAttributeType, + WorkflowFuture, ) @@ -51,55 +52,46 @@ def info(self) -> WorkflowInfo: ... def data_converter(self) -> DataConverter: ... @abstractmethod - async def execute_activity( + def execute_activity( self, activity: str, result_type: Type[ResultType], *args: Any, **kwargs: Unpack[ActivityOptions], - ) -> ResultType: ... + ) -> WorkflowFuture[ResultType]: ... @abstractmethod - async def execute_child_workflow( + def execute_child_workflow( self, workflow_type: str, result_type: Type[ResultType], *args: Any, **kwargs: Unpack[ChildWorkflowOptions], - ) -> ResultType: ... + ) -> ChildWorkflowFuture[ResultType]: ... @abstractmethod - async def start_child_workflow( - self, - workflow_type: str, - result_type: Type[ResultType], - *args: Any, - **kwargs: Unpack[ChildWorkflowOptions], - ) -> "ChildWorkflowFuture[ResultType]": ... - - @abstractmethod - async def signal_child_workflow( + def signal_child_workflow( self, child_workflow_id: str, signal_name: str, *args: Any, - ) -> None: ... + ) -> WorkflowFuture[None]: ... @abstractmethod - async def signal_external_workflow( + def signal_external_workflow( self, workflow_id: str, signal_name: str, *args: Any, run_id: str = "", domain: str = "", - ) -> None: ... + ) -> WorkflowFuture[None]: ... @abstractmethod - async def start_timer(self, duration: timedelta) -> None: ... + def start_timer(self, duration: timedelta) -> WorkflowFuture[None]: ... @abstractmethod - async def wait_condition(self, predicate: Callable[[], bool]) -> None: ... + def wait_condition(self, predicate: Callable[[], bool]) -> WorkflowFuture[None]: ... @abstractmethod def side_effect( diff --git a/cadence/workflow/_interceptor.py b/cadence/workflow/_interceptor.py index a8c26b4d..5df69216 100644 --- a/cadence/workflow/_interceptor.py +++ b/cadence/workflow/_interceptor.py @@ -11,6 +11,7 @@ ChildWorkflowOptions, ResultType, SearchAttributeType, + WorkflowFuture, ) @@ -34,72 +35,58 @@ def data_converter(self) -> DataConverter: return self._delegate.data_converter() @override - async def execute_activity( + def execute_activity( self, activity: str, result_type: Type[ResultType], *args: Any, **kwargs: Unpack[ActivityOptions], - ) -> ResultType: - return await self._delegate.execute_activity( - activity, result_type, *args, **kwargs - ) - - @override - async def execute_child_workflow( - self, - workflow_type: str, - result_type: Type[ResultType], - *args: Any, - **kwargs: Unpack[ChildWorkflowOptions], - ) -> ResultType: - return await self._delegate.execute_child_workflow( - workflow_type, result_type, *args, **kwargs - ) + ) -> WorkflowFuture[ResultType]: + return self._delegate.execute_activity(activity, result_type, *args, **kwargs) @override - async def start_child_workflow( + def execute_child_workflow( self, workflow_type: str, result_type: Type[ResultType], *args: Any, **kwargs: Unpack[ChildWorkflowOptions], - ) -> "ChildWorkflowFuture[ResultType]": - return await self._delegate.start_child_workflow( + ) -> ChildWorkflowFuture[ResultType]: + return self._delegate.execute_child_workflow( workflow_type, result_type, *args, **kwargs ) @override - async def signal_child_workflow( + def signal_child_workflow( self, child_workflow_id: str, signal_name: str, *args: Any, - ) -> None: - await self._delegate.signal_child_workflow( + ) -> WorkflowFuture[None]: + return self._delegate.signal_child_workflow( child_workflow_id, signal_name, *args ) @override - async def signal_external_workflow( + def signal_external_workflow( self, workflow_id: str, signal_name: str, *args: Any, run_id: str = "", domain: str = "", - ) -> None: - await self._delegate.signal_external_workflow( + ) -> WorkflowFuture[None]: + return self._delegate.signal_external_workflow( workflow_id, signal_name, *args, run_id=run_id, domain=domain ) @override - async def start_timer(self, duration: timedelta) -> None: - await self._delegate.start_timer(duration) + def start_timer(self, duration: timedelta) -> WorkflowFuture[None]: + return self._delegate.start_timer(duration) @override - async def wait_condition(self, predicate: Callable[[], bool]) -> None: - await self._delegate.wait_condition(predicate) + def wait_condition(self, predicate: Callable[[], bool]) -> WorkflowFuture[None]: + return self._delegate.wait_condition(predicate) @override def side_effect( diff --git a/cadence/workflow/_types.py b/cadence/workflow/_types.py index 32f76077..c3513f3f 100644 --- a/cadence/workflow/_types.py +++ b/cadence/workflow/_types.py @@ -1,16 +1,14 @@ +from asyncio import Future from datetime import datetime, timedelta from typing import ( Any, - Awaitable, - Generator, - Protocol, TypeVar, TypedDict, Union, - runtime_checkable, ) from cadence.api.v1 import workflow_pb2 +from cadence.api.v1.common_pb2 import WorkflowExecution ResultType = TypeVar("ResultType") ResultType_co = TypeVar("ResultType_co", covariant=True) @@ -46,6 +44,21 @@ class ActivityOptions(TypedDict, total=False): retry_policy: RetryPolicy +# See https://github.com/python/typeshed/pull/8781 . +# Future objects in Python are mutable, so their type parameter is invariant. You can't replace Future[any] with Future[str]. +# We however operate on covariant types, because our typing comes from user functions. +# Functions are always covariant because a function that returns type any is always allowed to return subclass str. +# As a result, we need a new type that accepts a covariant type parameter rather than using Future directly. +# asyncio already does this with the Task type, but our Futures aren't tasks. +# Python's current typing doesn't have a way to express this, so we end up with an ignore to allow it. +class WorkflowFuture(Future[ResultType_co]): # type: ignore[type-var] + """Futures returned as part of a Workflow. + + They may be canceled, but they cannot be mutated(set_result, set_exception) as they reflect an external asynchronous + operation. Workflow code can freely use normal asyncio Futures. + """ + + class ChildWorkflowOptions(TypedDict, total=False): workflow_id: str domain: str @@ -59,22 +72,26 @@ class ChildWorkflowOptions(TypedDict, total=False): memo: dict[str, Any] -@runtime_checkable -class ChildWorkflowFuture(Awaitable[ResultType_co], Protocol[ResultType_co]): - """Handle to a started child workflow, awaitable for its result.""" +class ChildWorkflowFuture(WorkflowFuture[ResultType_co]): + """Handle to a child workflow. Resolves to the child's result. + + Cancelling the future requests cancellation of the child workflow. + """ @property - def workflow_id(self) -> str: ... + def workflow_id(self) -> str: + raise NotImplementedError @property - def run_id(self) -> str: ... + def execution(self) -> WorkflowFuture[WorkflowExecution]: + """Resolves once the child has started, or raises if it failed to start.""" + raise NotImplementedError - def cancel(self) -> bool: - """Request cancellation of the child workflow.""" - ... + @property + def run_id(self) -> str: + """Run ID of the child. Only available once :attr:`execution` is done.""" + return self.execution.result().run_id - async def signal(self, signal_name: str, *args: Any) -> None: + def signal(self, signal_name: str, *args: Any) -> WorkflowFuture[None]: """Send a signal to this child workflow.""" - ... - - def __await__(self) -> Generator[Any, None, ResultType_co]: ... + raise NotImplementedError diff --git a/tests/cadence/_internal/workflow/statemachine/test_decision_manager.py b/tests/cadence/_internal/workflow/statemachine/test_decision_manager.py index 6c16d446..7720e80b 100644 --- a/tests/cadence/_internal/workflow/statemachine/test_decision_manager.py +++ b/tests/cadence/_internal/workflow/statemachine/test_decision_manager.py @@ -21,27 +21,36 @@ ) from cadence.api.v1 import history, decision from cadence.api.v1.common_pb2 import Payload, WorkflowExecution, WorkflowType +from cadence.data_converter import DefaultDataConverter + +result = "success!" +result_payload = Payload(data=b'"success!"') +converter = DefaultDataConverter() async def test_activity_dispatch(): decisions = DecisionManager(asyncio.get_event_loop()) activity_result = decisions.schedule_activity( - decision.ScheduleActivityTaskDecisionAttributes() + decision.ScheduleActivityTaskDecisionAttributes(), + converter, + str, ) decisions.handle_history_event(activity_scheduled(1, "0")) decisions.handle_history_event(activity_started(2, 1)) - decisions.handle_history_event(activity_completed(3, 1, Payload(data=b"completed"))) + decisions.handle_history_event(activity_completed(3, 1, result_payload)) assert activity_result.done() is True - assert activity_result.result() == Payload(data=b"completed") + assert activity_result.result() == result async def test_simple_cancellation(): decisions = DecisionManager(asyncio.get_event_loop()) activity_result = decisions.schedule_activity( - decision.ScheduleActivityTaskDecisionAttributes() + decision.ScheduleActivityTaskDecisionAttributes(), + converter, + str, ) activity_result.cancel() @@ -53,7 +62,9 @@ async def test_cancellation_not_immediate(): decisions = DecisionManager(asyncio.get_event_loop()) activity_result = decisions.schedule_activity( - decision.ScheduleActivityTaskDecisionAttributes() + decision.ScheduleActivityTaskDecisionAttributes(), + converter, + str, ) decisions.handle_history_event(activity_scheduled(1, "0")) cancelled = activity_result.cancel() @@ -67,7 +78,9 @@ async def test_cancellation_with_message_cancels_future_immediately(): decisions = DecisionManager(asyncio.get_event_loop()) activity_result = decisions.schedule_activity( - decision.ScheduleActivityTaskDecisionAttributes() + decision.ScheduleActivityTaskDecisionAttributes(), + converter, + str, ) decisions.handle_history_event(activity_scheduled(1, "0")) cancelled = activity_result.cancel("workflow cancelled") @@ -79,11 +92,43 @@ async def test_cancellation_with_message_cancels_future_immediately(): activity_result.result() +async def test_completion_after_cancellation_with_message_is_ignored(): + decisions = DecisionManager(asyncio.get_event_loop()) + + activity_result = decisions.schedule_activity( + decision.ScheduleActivityTaskDecisionAttributes(), + converter, + str, + ) + decisions.handle_history_event(activity_scheduled(1, "0")) + activity_result.cancel("workflow cancelled") + decisions.handle_history_event(activity_completed(2, 1, result_payload)) + + assert activity_result.cancelled() is True + + +async def test_activity_result_conversion_failure(): + decisions = DecisionManager(asyncio.get_event_loop()) + + activity_result = decisions.schedule_activity( + decision.ScheduleActivityTaskDecisionAttributes(), + converter, + int, + ) + decisions.handle_history_event(activity_scheduled(1, "0")) + decisions.handle_history_event(activity_completed(2, 1, result_payload)) + + with pytest.raises(RuntimeError, match="Failed to convert result"): + activity_result.result() + + async def test_cancellation_completed(): decisions = DecisionManager(asyncio.get_event_loop()) activity_result = decisions.schedule_activity( - decision.ScheduleActivityTaskDecisionAttributes() + decision.ScheduleActivityTaskDecisionAttributes(), + converter, + str, ) decisions.handle_history_event(activity_scheduled(1, "0")) activity_result.cancel() @@ -114,10 +159,14 @@ async def test_collect_decisions(): decisions = DecisionManager(asyncio.get_event_loop()) activity1 = decisions.schedule_activity( - decision.ScheduleActivityTaskDecisionAttributes() + decision.ScheduleActivityTaskDecisionAttributes(), + converter, + str, ) activity2 = decisions.schedule_activity( - decision.ScheduleActivityTaskDecisionAttributes() + decision.ScheduleActivityTaskDecisionAttributes(), + converter, + str, ) # Order matters @@ -140,7 +189,9 @@ async def test_collect_decisions(): async def test_collect_decisions_ignore_empty(): decisions = DecisionManager(asyncio.get_event_loop()) - _ = decisions.schedule_activity(decision.ScheduleActivityTaskDecisionAttributes()) + _ = decisions.schedule_activity( + decision.ScheduleActivityTaskDecisionAttributes(), converter, str + ) decisions.handle_history_event(activity_scheduled(1, "0")) assert decisions.collect_pending_decisions() == [] @@ -180,10 +231,14 @@ async def test_collection_decisions_reordering(): decisions = DecisionManager(asyncio.get_event_loop()) activity1 = decisions.schedule_activity( - decision.ScheduleActivityTaskDecisionAttributes() + decision.ScheduleActivityTaskDecisionAttributes(), + converter, + str, ) activity2 = decisions.schedule_activity( - decision.ScheduleActivityTaskDecisionAttributes() + decision.ScheduleActivityTaskDecisionAttributes(), + converter, + str, ) assert decisions.collect_pending_decisions() == [ @@ -230,7 +285,9 @@ async def test_record_marker_collects_in_workflow_order(): decisions.start_timer(decision.StartTimerDecisionAttributes()) recorded = decisions.record_marker(marker_attrs) - decisions.schedule_activity(decision.ScheduleActivityTaskDecisionAttributes()) + decisions.schedule_activity( + decision.ScheduleActivityTaskDecisionAttributes(), converter, str + ) assert recorded == Payload(data=b"marker") assert decisions.collect_pending_decisions() == [ @@ -391,7 +448,9 @@ async def test_mutable_side_effect_does_not_shift_other_decision_ids_on_replay() ) assert has_history_update is True decisions.record_mutable_side_effect("config", access_count, Payload()) - decisions.schedule_activity(decision.ScheduleActivityTaskDecisionAttributes()) + decisions.schedule_activity( + decision.ScheduleActivityTaskDecisionAttributes(), converter, str + ) async def test_mutable_side_effect_replay_applies_marker_at_matching_access_count(): @@ -483,20 +542,23 @@ async def test_replayed_marker_details_come_from_history_not_current_value(): async def test_child_workflow_dispatch(): decisions = DecisionManager(asyncio.get_event_loop()) - execution, result = decisions.schedule_child_workflow( + result = decisions.schedule_child_workflow( decision.StartChildWorkflowExecutionDecisionAttributes( workflow_id="child-wf-1", workflow_type=WorkflowType(name="MyWorkflow"), ), + converter, + str, parent_workflow_run_id="parent-run", ) + execution = result.execution decisions.handle_history_event(child_wf_initiated(1, "child-wf-1")) decisions.handle_history_event( child_wf_started(2, started_event_id=1, wf_id="child-wf-1", run_id="run-1") ) decisions.handle_history_event( - child_wf_completed(3, started_event_id=1, result=Payload(data=b"done")) + child_wf_completed(3, started_event_id=1, result=result_payload) ) assert execution.done() is True @@ -504,7 +566,47 @@ async def test_child_workflow_dispatch(): workflow_id="child-wf-1", run_id="run-1" ) assert result.done() is True - assert result.result() == Payload(data=b"done") + assert result.result() == "success!" + + +async def test_child_workflow_id_available_before_start(): + decisions = DecisionManager(asyncio.get_event_loop()) + + result = decisions.schedule_child_workflow( + decision.StartChildWorkflowExecutionDecisionAttributes( + workflow_type=WorkflowType(name="MyWorkflow"), + ), + converter, + str, + parent_workflow_run_id="parent-run", + ) + + assert result.workflow_id == "parent-run_0" + assert result.execution.done() is False + + +async def test_child_workflow_result_conversion_failure(): + decisions = DecisionManager(asyncio.get_event_loop()) + + result = decisions.schedule_child_workflow( + decision.StartChildWorkflowExecutionDecisionAttributes( + workflow_id="child-wf-1", + workflow_type=WorkflowType(name="MyWorkflow"), + ), + converter, + int, + parent_workflow_run_id="parent-run", + ) + decisions.handle_history_event(child_wf_initiated(1, "child-wf-1")) + decisions.handle_history_event( + child_wf_started(2, started_event_id=1, wf_id="child-wf-1", run_id="run-1") + ) + decisions.handle_history_event( + child_wf_completed(3, started_event_id=1, result=result_payload) + ) + + with pytest.raises(RuntimeError, match="Failed to convert result"): + result.result() async def test_schedule_child_workflow_generates_workflow_id(): @@ -514,8 +616,8 @@ async def test_schedule_child_workflow_generates_workflow_id(): workflow_id="", workflow_type=WorkflowType(name="MyWorkflow"), ) - _execution, result = decisions.schedule_child_workflow( - attrs, parent_workflow_run_id="parent-run-xyz" + result = decisions.schedule_child_workflow( + attrs, converter, str, parent_workflow_run_id="parent-run-xyz" ) assert attrs.workflow_id == "parent-run-xyz_0" @@ -525,22 +627,25 @@ async def test_schedule_child_workflow_generates_workflow_id(): child_wf_started(2, started_event_id=1, wf_id=attrs.workflow_id, run_id="run-1") ) decisions.handle_history_event( - child_wf_completed(3, started_event_id=1, result=Payload(data=b"done")) + child_wf_completed(3, started_event_id=1, result=result_payload) ) - assert result.result() == Payload(data=b"done") + assert result.result() == "success!" async def test_child_workflow_initiation_failed_dispatch(): decisions = DecisionManager(asyncio.get_event_loop()) - execution, result = decisions.schedule_child_workflow( + result = decisions.schedule_child_workflow( decision.StartChildWorkflowExecutionDecisionAttributes( workflow_id="child-wf-1", workflow_type=WorkflowType(name="MyWorkflow"), ), + converter, + str, parent_workflow_run_id="parent-run", ) + execution = result.execution decisions.handle_history_event(child_wf_initiated(1, "child-wf-1")) decisions.handle_history_event( @@ -563,11 +668,13 @@ async def test_child_workflow_initiation_failed_dispatch(): async def test_child_workflow_cancel_dispatch(): decisions = DecisionManager(asyncio.get_event_loop()) - _execution, result = decisions.schedule_child_workflow( + result = decisions.schedule_child_workflow( decision.StartChildWorkflowExecutionDecisionAttributes( workflow_id="child-wf-1", workflow_type=WorkflowType(name="MyWorkflow"), ), + converter, + str, parent_workflow_run_id="parent-run", ) @@ -607,11 +714,13 @@ async def test_child_workflow_errors_are_child_workflow_error(): """All child workflow errors share a common base class.""" decisions = DecisionManager(asyncio.get_event_loop()) - _execution, result = decisions.schedule_child_workflow( + result = decisions.schedule_child_workflow( decision.StartChildWorkflowExecutionDecisionAttributes( workflow_id="child-wf-1", workflow_type=WorkflowType(name="MyWorkflow"), ), + converter, + str, parent_workflow_run_id="parent-run", ) diff --git a/tests/cadence/_internal/workflow/test_context_child_workflow.py b/tests/cadence/_internal/workflow/test_context_child_workflow.py index f0f98ed2..b52f2020 100644 --- a/tests/cadence/_internal/workflow/test_context_child_workflow.py +++ b/tests/cadence/_internal/workflow/test_context_child_workflow.py @@ -1,84 +1,108 @@ import asyncio from datetime import timedelta +from typing import Any from unittest.mock import MagicMock import pytest from google.protobuf.duration_pb2 import Duration -from cadence._internal.workflow.child_workflow_future import ChildWorkflowFutureImpl -from cadence._internal.workflow.context import Context -from cadence.api.v1 import workflow_pb2 -from cadence.api.v1.common_pb2 import WorkflowExecution, WorkflowType +from cadence._internal.workflow.context import Context, _activate_workflow_context +from cadence._internal.workflow.statemachine.decision_manager import DecisionManager +from cadence.api.v1 import history, workflow_pb2 +from cadence.api.v1.common_pb2 import Failure, WorkflowExecution, WorkflowType from cadence.api.v1.decision_pb2 import StartChildWorkflowExecutionDecisionAttributes from cadence.api.v1.tasklist_pb2 import TaskList, TaskListKind from cadence.data_converter import DefaultDataConverter -from cadence.error import ChildWorkflowExecutionFailed -from cadence.workflow import WorkflowInfo +from cadence.error import ( + ChildWorkflowExecutionFailed, + StartChildWorkflowExecutionFailed, +) +from cadence import workflow +from cadence.workflow import ChildWorkflowFuture, WorkflowInfo def _make_ctx(run_id: str = "rid") -> tuple[Context, MagicMock]: - dm = MagicMock() - dc = DefaultDataConverter() + dm = MagicMock(wraps=DecisionManager(asyncio.get_running_loop())) info = WorkflowInfo( workflow_type="ParentWf", workflow_domain="domain", workflow_id="wid", workflow_run_id=run_id, workflow_task_list="tl", - data_converter=dc, + data_converter=DefaultDataConverter(), ) ctx = Context(info, dm) return ctx, dm -def _setup_schedule_mock( - dm: MagicMock, - result_value: object = "result", - loop: asyncio.AbstractEventLoop | None = None, - workflow_execution: WorkflowExecution | None = None, -) -> None: - dc = DefaultDataConverter() - loop = loop or asyncio.get_event_loop() - if workflow_execution is None: - workflow_execution = WorkflowExecution( - workflow_id="default-child", run_id="default-run" +def _start_child_workflow( + ctx: Context, *args: Any, **kwargs: Any +) -> asyncio.Future[ChildWorkflowFuture[Any]]: + with _activate_workflow_context(ctx): + # The task copies the active context when it is created. + return asyncio.ensure_future(workflow.start_child_workflow(*args, **kwargs)) + + +def _start_child(dm: MagicMock, workflow_id: str, run_id: str = "run-1") -> None: + dm.handle_history_event( + history.HistoryEvent( + event_id=1, + start_child_workflow_execution_initiated_event_attributes=history.StartChildWorkflowExecutionInitiatedEventAttributes( + workflow_id=workflow_id, + ), + ) + ) + dm.handle_history_event( + history.HistoryEvent( + event_id=2, + child_workflow_execution_started_event_attributes=history.ChildWorkflowExecutionStartedEventAttributes( + initiated_event_id=1, + workflow_execution=WorkflowExecution( + workflow_id=workflow_id, run_id=run_id + ), + ), ) - execution_future: asyncio.Future = loop.create_future() - execution_future.set_result(workflow_execution) - result_future: asyncio.Future = loop.create_future() - result_future.set_result(dc.to_data([result_value])) - dm.schedule_child_workflow = MagicMock( - return_value=(execution_future, result_future) ) -def _setup_schedule_mock_error( - dm: MagicMock, error: Exception, loop: asyncio.AbstractEventLoop | None = None -) -> None: - loop = loop or asyncio.get_event_loop() - execution_future: asyncio.Future = loop.create_future() - execution_future.set_result( - WorkflowExecution(workflow_id="default-child", run_id="default-run") +def _complete_child(dm: MagicMock, result: object) -> None: + dm.handle_history_event( + history.HistoryEvent( + event_id=3, + child_workflow_execution_completed_event_attributes=history.ChildWorkflowExecutionCompletedEventAttributes( + initiated_event_id=1, + result=DefaultDataConverter().to_data([result]), + ), + ) ) - result_future: asyncio.Future = loop.create_future() - result_future.set_exception(error) - dm.schedule_child_workflow = MagicMock( - return_value=(execution_future, result_future) + + +def _fail_child(dm: MagicMock, reason: str) -> None: + dm.handle_history_event( + history.HistoryEvent( + event_id=3, + child_workflow_execution_failed_event_attributes=history.ChildWorkflowExecutionFailedEventAttributes( + initiated_event_id=1, + failure=Failure(reason=reason), + ), + ) ) @pytest.mark.asyncio async def test_execute_child_workflow_basic(): ctx, dm = _make_ctx() - _setup_schedule_mock(dm) - result = await ctx.execute_child_workflow( + future = ctx.execute_child_workflow( "ChildWf", str, "arg1", "arg2", execution_start_to_close_timeout=timedelta(minutes=10), ) + _start_child(dm, future.workflow_id) + _complete_child(dm, "result") + result = await future dm.schedule_child_workflow.assert_called_once() attrs: StartChildWorkflowExecutionDecisionAttributes = ( @@ -97,10 +121,9 @@ async def test_execute_child_workflow_basic(): @pytest.mark.asyncio async def test_execute_child_workflow_with_memo(): ctx, dm = _make_ctx() - _setup_schedule_mock(dm) dc = ctx.data_converter() - await ctx.execute_child_workflow( + ctx.execute_child_workflow( "ChildWf", str, execution_start_to_close_timeout=timedelta(minutes=10), @@ -118,9 +141,8 @@ async def test_execute_child_workflow_with_memo(): @pytest.mark.asyncio async def test_execute_child_workflow_auto_generates_workflow_id(): ctx, dm = _make_ctx(run_id="test-run-id") - _setup_schedule_mock(dm) - await ctx.execute_child_workflow( + ctx.execute_child_workflow( "ChildWf", str, execution_start_to_close_timeout=timedelta(minutes=5), @@ -129,7 +151,7 @@ async def test_execute_child_workflow_auto_generates_workflow_id(): attrs: StartChildWorkflowExecutionDecisionAttributes = ( dm.schedule_child_workflow.call_args[0][0] ) - assert attrs.workflow_id == "" + assert attrs.workflow_id == "test-run-id_0" assert dm.schedule_child_workflow.call_args.kwargs["parent_workflow_run_id"] == ( "test-run-id" ) @@ -138,9 +160,8 @@ async def test_execute_child_workflow_auto_generates_workflow_id(): @pytest.mark.asyncio async def test_execute_child_workflow_uses_provided_workflow_id(): ctx, dm = _make_ctx() - _setup_schedule_mock(dm) - await ctx.execute_child_workflow( + ctx.execute_child_workflow( "ChildWf", str, workflow_id="my-child-1", @@ -160,7 +181,7 @@ async def test_execute_child_workflow_raises_without_execution_timeout(): with pytest.raises( ValueError, match="execution_start_to_close_timeout is required" ): - await ctx.execute_child_workflow("ChildWf", str) + ctx.execute_child_workflow("ChildWf", str) @pytest.mark.asyncio @@ -173,7 +194,7 @@ async def test_execute_child_workflow_raises_for_invalid_execution_timeout( with pytest.raises( ValueError, match="execution_start_to_close_timeout must be greater than 0" ): - await ctx.execute_child_workflow( + ctx.execute_child_workflow( "ChildWf", str, execution_start_to_close_timeout=timeout, @@ -190,7 +211,7 @@ async def test_execute_child_workflow_raises_for_invalid_task_timeout( with pytest.raises( ValueError, match="task_start_to_close_timeout must be greater than 0" ): - await ctx.execute_child_workflow( + ctx.execute_child_workflow( "ChildWf", str, execution_start_to_close_timeout=timedelta(minutes=5), @@ -206,7 +227,7 @@ async def test_execute_child_workflow_raises_for_invalid_workflow_id_reuse_polic ValueError, match="workflow_id_reuse_policy cannot be WORKFLOW_ID_REUSE_POLICY_INVALID", ): - await ctx.execute_child_workflow( + ctx.execute_child_workflow( "ChildWf", str, execution_start_to_close_timeout=timedelta(minutes=5), @@ -217,9 +238,8 @@ async def test_execute_child_workflow_raises_for_invalid_workflow_id_reuse_polic @pytest.mark.asyncio async def test_execute_child_workflow_default_values(): ctx, dm = _make_ctx() - _setup_schedule_mock(dm) - await ctx.execute_child_workflow( + ctx.execute_child_workflow( "ChildWf", str, execution_start_to_close_timeout=timedelta(minutes=5), @@ -239,9 +259,8 @@ async def test_execute_child_workflow_default_values(): @pytest.mark.asyncio async def test_execute_child_workflow_custom_options(): ctx, dm = _make_ctx() - _setup_schedule_mock(dm) - await ctx.execute_child_workflow( + ctx.execute_child_workflow( "ChildWf", str, execution_start_to_close_timeout=timedelta(minutes=5), @@ -263,9 +282,8 @@ async def test_execute_child_workflow_custom_options(): @pytest.mark.asyncio async def test_execute_child_workflow_retry_policy(): ctx, dm = _make_ctx() - _setup_schedule_mock(dm) - await ctx.execute_child_workflow( + ctx.execute_child_workflow( "ChildWf", str, execution_start_to_close_timeout=timedelta(minutes=5), @@ -289,14 +307,16 @@ async def test_execute_child_workflow_retry_policy(): @pytest.mark.asyncio async def test_execute_child_workflow_deserializes_result(): ctx, dm = _make_ctx() - _setup_schedule_mock(dm, result_value=42) - result = await ctx.execute_child_workflow( + future = ctx.execute_child_workflow( "ChildWf", int, execution_start_to_close_timeout=timedelta(minutes=5), ) + _start_child(dm, future.workflow_id) + _complete_child(dm, 42) + result = await future assert result == 42 assert isinstance(result, int) @@ -304,23 +324,24 @@ async def test_execute_child_workflow_deserializes_result(): @pytest.mark.asyncio async def test_execute_child_workflow_propagates_errors(): ctx, dm = _make_ctx() - failure = ChildWorkflowExecutionFailed("child failed", failure=None) - _setup_schedule_mock_error(dm, failure) + + future = ctx.execute_child_workflow( + "ChildWf", + str, + execution_start_to_close_timeout=timedelta(minutes=5), + ) + _start_child(dm, future.workflow_id) + _fail_child(dm, "child failed") with pytest.raises(ChildWorkflowExecutionFailed, match="child failed"): - await ctx.execute_child_workflow( - "ChildWf", - str, - execution_start_to_close_timeout=timedelta(minutes=5), - ) + await future @pytest.mark.asyncio async def test_execute_child_workflow_cron_schedule(): ctx, dm = _make_ctx() - _setup_schedule_mock(dm) - await ctx.execute_child_workflow( + ctx.execute_child_workflow( "ChildWf", str, execution_start_to_close_timeout=timedelta(minutes=5), @@ -334,18 +355,21 @@ async def test_execute_child_workflow_cron_schedule(): @pytest.mark.asyncio -async def test_start_child_workflow_returns_future_with_ids(): +async def test_start_child_workflow_waits_for_child_to_start(): ctx, dm = _make_ctx() - _setup_schedule_mock( - dm, - workflow_execution=WorkflowExecution(workflow_id="child-1", run_id="run-1"), - ) - future = await ctx.start_child_workflow( + start = _start_child_workflow( + ctx, "ChildWf", str, + workflow_id="child-1", execution_start_to_close_timeout=timedelta(minutes=5), ) + await asyncio.sleep(0) + assert start.done() is False + + _start_child(dm, "child-1", run_id="run-1") + future = await start assert future.workflow_id == "child-1" assert future.run_id == "run-1" @@ -354,101 +378,76 @@ async def test_start_child_workflow_returns_future_with_ids(): @pytest.mark.asyncio async def test_start_child_workflow_future_awaitable(): ctx, dm = _make_ctx() - _setup_schedule_mock( - dm, - result_value=42, - workflow_execution=WorkflowExecution(workflow_id="child-1", run_id="run-1"), - ) - future = await ctx.start_child_workflow( + start = _start_child_workflow( + ctx, "ChildWf", int, + workflow_id="child-1", execution_start_to_close_timeout=timedelta(minutes=5), ) - result = await future + await asyncio.sleep(0) + _start_child(dm, "child-1") + future = await start + _complete_child(dm, 42) + result = await future assert result == 42 assert isinstance(result, int) @pytest.mark.asyncio -async def test_start_child_workflow_future_cancel(): +async def test_child_workflow_future_cancel_requests_cancellation(): ctx, dm = _make_ctx() - _setup_schedule_mock( - dm, - workflow_execution=WorkflowExecution(workflow_id="child-1", run_id="run-1"), - ) - future = await ctx.start_child_workflow( + future = ctx.execute_child_workflow( "ChildWf", str, + workflow_id="child-1", execution_start_to_close_timeout=timedelta(minutes=5), ) - assert isinstance(future, ChildWorkflowFutureImpl) - future._result_future = MagicMock() - future._result_future.cancel = MagicMock(return_value=True) - - cancelled = future.cancel() - - assert cancelled is True - future._result_future.cancel.assert_called_once() - + _start_child(dm, "child-1", run_id="run-1") -@pytest.mark.asyncio -async def test_start_child_workflow_raises_without_timeout(): - ctx, _dm = _make_ctx() + future.cancel() - with pytest.raises( - ValueError, match="execution_start_to_close_timeout is required" - ): - await ctx.start_child_workflow("ChildWf", str) + decisions = dm.collect_pending_decisions() + assert len(decisions) == 1 + cancel = decisions[0].request_cancel_external_workflow_execution_decision_attributes + assert cancel.workflow_execution == WorkflowExecution( + workflow_id="child-1", run_id="run-1" + ) + assert cancel.child_workflow_only is True @pytest.mark.asyncio -async def test_start_child_workflow_passes_correct_attrs(): +async def test_start_child_workflow_raises_when_start_fails(): ctx, dm = _make_ctx() - _setup_schedule_mock( - dm, - workflow_execution=WorkflowExecution(workflow_id="child-1", run_id="run-1"), - ) - await ctx.start_child_workflow( + start = _start_child_workflow( + ctx, "ChildWf", str, - execution_start_to_close_timeout=timedelta(minutes=10), - workflow_id="my-child", - ) - - dm.schedule_child_workflow.assert_called_once() - attrs: StartChildWorkflowExecutionDecisionAttributes = ( - dm.schedule_child_workflow.call_args[0][0] - ) - assert attrs.workflow_type == WorkflowType(name="ChildWf") - assert attrs.workflow_id == "my-child" - assert attrs.execution_start_to_close_timeout == Duration(seconds=600) - - -@pytest.mark.asyncio -async def test_start_child_workflow_future_propagates_errors(): - ctx, dm = _make_ctx() - loop = asyncio.get_event_loop() - - execution_future: asyncio.Future = loop.create_future() - execution_future.set_result( - WorkflowExecution(workflow_id="child-1", run_id="run-1") + workflow_id="child-1", + execution_start_to_close_timeout=timedelta(minutes=5), ) - result_future: asyncio.Future = loop.create_future() - result_future.set_exception( - ChildWorkflowExecutionFailed("child failed", failure=None) + await asyncio.sleep(0) + dm.handle_history_event( + history.HistoryEvent( + event_id=1, + start_child_workflow_execution_initiated_event_attributes=history.StartChildWorkflowExecutionInitiatedEventAttributes( + workflow_id="child-1", + ), + ) ) - dm.schedule_child_workflow = MagicMock( - return_value=(execution_future, result_future) + dm.handle_history_event( + history.HistoryEvent( + event_id=2, + start_child_workflow_execution_failed_event_attributes=history.StartChildWorkflowExecutionFailedEventAttributes( + initiated_event_id=1, + workflow_id="child-1", + ), + ) ) - future = await ctx.start_child_workflow( - "ChildWf", - str, - execution_start_to_close_timeout=timedelta(minutes=5), - ) - with pytest.raises(ChildWorkflowExecutionFailed, match="child failed"): - await future + with pytest.raises(StartChildWorkflowExecutionFailed): + await start diff --git a/tests/cadence/_internal/workflow/test_context_retry_policy.py b/tests/cadence/_internal/workflow/test_context_retry_policy.py index 2f9d724a..55fa51f3 100644 --- a/tests/cadence/_internal/workflow/test_context_retry_policy.py +++ b/tests/cadence/_internal/workflow/test_context_retry_policy.py @@ -1,5 +1,5 @@ from datetime import timedelta -from unittest.mock import AsyncMock, MagicMock +from unittest.mock import MagicMock import pytest from google.protobuf.duration_pb2 import Duration @@ -15,9 +15,6 @@ @pytest.mark.asyncio async def test_execute_activity_passes_retry_policy_to_schedule(): dm = MagicMock() - dm.schedule_activity = AsyncMock( - return_value=DefaultDataConverter().to_data(["result"]) - ) info = WorkflowInfo( workflow_type="Wf", @@ -29,7 +26,7 @@ async def test_execute_activity_passes_retry_policy_to_schedule(): ) ctx = Context(info, dm) - await ctx.execute_activity( + ctx.execute_activity( "Act", str, retry_policy={ @@ -39,7 +36,7 @@ async def test_execute_activity_passes_retry_policy_to_schedule(): }, ) - dm.schedule_activity.assert_awaited_once() + dm.schedule_activity.assert_called_once() attrs: ScheduleActivityTaskDecisionAttributes = dm.schedule_activity.call_args[0][0] assert attrs.activity_type == ActivityType(name="Act") assert attrs.domain == "domain" diff --git a/tests/cadence/_internal/workflow/test_context_signal_child_workflow.py b/tests/cadence/_internal/workflow/test_context_signal_child_workflow.py index 3dab21c9..f4629135 100644 --- a/tests/cadence/_internal/workflow/test_context_signal_child_workflow.py +++ b/tests/cadence/_internal/workflow/test_context_signal_child_workflow.py @@ -1,10 +1,12 @@ import asyncio from typing import cast -from unittest.mock import AsyncMock, MagicMock, patch +from datetime import timedelta +from unittest.mock import MagicMock import pytest -from cadence._internal.workflow.context import Context +from cadence._internal.workflow.context import Context, _activate_workflow_context +from cadence._internal.workflow.statemachine.decision_manager import DecisionManager from cadence.api.v1.decision_pb2 import ( SignalExternalWorkflowExecutionDecisionAttributes, ) @@ -51,7 +53,7 @@ async def test_signal_child_workflow_basic(): ctx, dm = _make_ctx() _setup_signal_mock(dm) - await ctx.signal_child_workflow("child-wf-1", "my_signal", "arg1") + ctx.signal_child_workflow("child-wf-1", "my_signal", "arg1") dm.signal_external_workflow.assert_called_once() attrs = _attrs(dm) @@ -66,7 +68,7 @@ async def test_signal_child_workflow_serializes_args(): _setup_signal_mock(dm) dc = DefaultDataConverter() - await ctx.signal_child_workflow("child-wf-1", "my_signal", "hello", 42) + ctx.signal_child_workflow("child-wf-1", "my_signal", "hello", 42) deserialized = dc.from_data(_attrs(dm).input, [str, int]) assert deserialized == ["hello", 42] @@ -77,7 +79,7 @@ async def test_signal_child_workflow_sets_child_workflow_only(): ctx, dm = _make_ctx() _setup_signal_mock(dm) - await ctx.signal_child_workflow("child-wf-1", "my_signal") + ctx.signal_child_workflow("child-wf-1", "my_signal") assert _attrs(dm).child_workflow_only is True @@ -87,7 +89,7 @@ async def test_signal_child_workflow_uses_empty_run_id(): ctx, dm = _make_ctx() _setup_signal_mock(dm) - await ctx.signal_child_workflow("child-wf-1", "my_signal") + ctx.signal_child_workflow("child-wf-1", "my_signal") assert _attrs(dm).workflow_execution.run_id == "" @@ -98,7 +100,7 @@ async def test_signal_child_workflow_rejects_empty_id(): _setup_signal_mock(dm) with pytest.raises(ValueError, match="child_workflow_id must not be empty"): - await ctx.signal_child_workflow("", "my_signal") + ctx.signal_child_workflow("", "my_signal") dm.signal_external_workflow.assert_not_called() @@ -113,7 +115,7 @@ async def test_signal_external_workflow_basic(): ctx, dm = _make_ctx() _setup_signal_mock(dm) - await ctx.signal_external_workflow("ext-wf-1", "ping") + ctx.signal_external_workflow("ext-wf-1", "ping") dm.signal_external_workflow.assert_called_once() attrs = _attrs(dm) @@ -127,7 +129,7 @@ async def test_signal_external_workflow_defaults_to_own_domain(): ctx, dm = _make_ctx() _setup_signal_mock(dm) - await ctx.signal_external_workflow("ext-wf-1", "ping") + ctx.signal_external_workflow("ext-wf-1", "ping") assert _attrs(dm).domain == "domain" @@ -137,7 +139,7 @@ async def test_signal_external_workflow_custom_domain(): ctx, dm = _make_ctx() _setup_signal_mock(dm) - await ctx.signal_external_workflow("ext-wf-1", "ping", domain="other-domain") + ctx.signal_external_workflow("ext-wf-1", "ping", domain="other-domain") assert _attrs(dm).domain == "other-domain" @@ -147,7 +149,7 @@ async def test_signal_external_workflow_forwards_run_id(): ctx, dm = _make_ctx() _setup_signal_mock(dm) - await ctx.signal_external_workflow("ext-wf-1", "ping", run_id="run-42") + ctx.signal_external_workflow("ext-wf-1", "ping", run_id="run-42") assert _attrs(dm).workflow_execution.run_id == "run-42" @@ -157,7 +159,7 @@ async def test_signal_external_workflow_default_run_id_is_empty(): ctx, dm = _make_ctx() _setup_signal_mock(dm) - await ctx.signal_external_workflow("ext-wf-1", "ping") + ctx.signal_external_workflow("ext-wf-1", "ping") assert _attrs(dm).workflow_execution.run_id == "" @@ -168,7 +170,7 @@ async def test_signal_external_workflow_serializes_args(): _setup_signal_mock(dm) dc = DefaultDataConverter() - await ctx.signal_external_workflow("ext-wf-1", "ping", "hello", 99) + ctx.signal_external_workflow("ext-wf-1", "ping", "hello", 99) deserialized = dc.from_data(_attrs(dm).input, [str, int]) assert deserialized == ["hello", 99] @@ -180,7 +182,7 @@ async def test_signal_external_workflow_rejects_empty_id(): _setup_signal_mock(dm) with pytest.raises(ValueError, match="workflow_id must not be empty"): - await ctx.signal_external_workflow("", "ping") + ctx.signal_external_workflow("", "ping") dm.signal_external_workflow.assert_not_called() @@ -191,7 +193,7 @@ async def test_signal_external_workflow_rejects_empty_signal_name(): _setup_signal_mock(dm) with pytest.raises(ValueError, match="signal_name must not be empty"): - await ctx.signal_external_workflow("wf-1", "") + ctx.signal_external_workflow("wf-1", "") dm.signal_external_workflow.assert_not_called() @@ -202,7 +204,7 @@ async def test_signal_child_workflow_rejects_empty_signal_name(): _setup_signal_mock(dm) with pytest.raises(ValueError, match="signal_name must not be empty"): - await ctx.signal_child_workflow("child-wf-1", "") + ctx.signal_child_workflow("child-wf-1", "") dm.signal_external_workflow.assert_not_called() @@ -213,48 +215,36 @@ async def test_signal_child_workflow_rejects_empty_signal_name(): @pytest.mark.asyncio -async def test_child_workflow_future_signal_delegates_to_context(): - from cadence._internal.workflow.child_workflow_future import ( - ChildWorkflowFutureImpl, +async def test_child_workflow_future_signal_records_child_only_signal(): + dm = DecisionManager(asyncio.get_running_loop()) + ctx = Context( + WorkflowInfo( + workflow_type="ParentWf", + workflow_domain="domain", + workflow_id="wid", + workflow_run_id="rid", + workflow_task_list="tl", + data_converter=DefaultDataConverter(), + ), + dm, ) - from cadence.workflow import ChildWorkflowFuture - dc = DefaultDataConverter() - result_future: asyncio.Future = asyncio.get_running_loop().create_future() - - future = ChildWorkflowFutureImpl( - workflow_id="child-wf-1", - run_id="", - result_future=result_future, - result_type=str, - data_converter=dc, - ) - assert isinstance(future, ChildWorkflowFuture) - - mock_ctx = AsyncMock() - with patch("cadence.workflow.WorkflowContext.get", return_value=mock_ctx): - await future.signal("hello_signal", "arg1", "arg2") - - mock_ctx.signal_child_workflow.assert_awaited_once_with( - "child-wf-1", "hello_signal", "arg1", "arg2" - ) - - -# --------------------------------------------------------------------------- -# Public cadence.workflow.signal_external_workflow wrapper -# --------------------------------------------------------------------------- - - -@pytest.mark.asyncio -async def test_public_signal_external_workflow_delegates(): - import cadence.workflow as wf - - mock_ctx = AsyncMock() - with patch("cadence.workflow.WorkflowContext.get", return_value=mock_ctx): - await wf.signal_external_workflow( - "wf-id", "my_signal", "a", run_id="r1", domain="d1" + with _activate_workflow_context(ctx): + handle = ctx.execute_child_workflow( + "ChildWf", + str, + workflow_id="child-wf-1", + execution_start_to_close_timeout=timedelta(minutes=5), ) - - mock_ctx.signal_external_workflow.assert_awaited_once_with( - "wf-id", "my_signal", "a", run_id="r1", domain="d1" - ) + signal = handle.signal("hello_signal", "arg1") + + assert signal.done() is False + signals = [ + d.signal_external_workflow_execution_decision_attributes + for d in dm.collect_pending_decisions() + if d.HasField("signal_external_workflow_execution_decision_attributes") + ] + assert len(signals) == 1 + assert signals[0].signal_name == "hello_signal" + assert signals[0].workflow_execution.workflow_id == "child-wf-1" + assert signals[0].child_workflow_only is True diff --git a/tests/cadence/_internal/workflow/test_waiter.py b/tests/cadence/_internal/workflow/test_waiter.py index fca97251..58eabbbe 100644 --- a/tests/cadence/_internal/workflow/test_waiter.py +++ b/tests/cadence/_internal/workflow/test_waiter.py @@ -121,3 +121,20 @@ def test_loop_poll_via_create_waiter(self) -> None: w = loop.create_waiter(lambda: False) assert not w.done() assert w in loop._waiters # type: ignore[attr-defined] + + def test_cancelled_waiter_is_evicted_without_polling(self) -> None: + loop = make_loop() + calls = {"n": 0} + + def predicate() -> bool: + calls["n"] += 1 + return False + + w = loop.create_waiter(predicate) + calls_before = calls["n"] + w.cancel() + loop._run_once() # type: ignore[attr-defined] + + assert w.cancelled() + assert w not in loop._waiters # type: ignore[attr-defined] + assert calls["n"] == calls_before diff --git a/tests/cadence/testing/test_workflow_environment.py b/tests/cadence/testing/test_workflow_environment.py index 9b295603..2487f563 100644 --- a/tests/cadence/testing/test_workflow_environment.py +++ b/tests/cadence/testing/test_workflow_environment.py @@ -10,9 +10,11 @@ from cadence.worker import Registry from cadence.workflow import ( ActivityOptions, + ChildWorkflowFuture, ChildWorkflowOptions, ResultType, WorkflowContext, + WorkflowFuture, WorkflowInterceptor, ) @@ -134,6 +136,19 @@ async def run(self, value: str) -> str: return f"parent:{result}" +@registry.workflow +class StartChildWorkflow: + @workflow.run + async def run(self, value: str) -> str: + handle = await workflow.start_child_workflow( + "ChildWorkflow", + str, + value, + execution_start_to_close_timeout=timedelta(seconds=30), + ) + return await handle + + @registry.workflow class FailingWorkflow: @workflow.run @@ -188,25 +203,23 @@ def _prefix(args: tuple[Any, ...]) -> tuple[Any, ...]: class PrefixingInterceptor(WorkflowInterceptor): - async def execute_activity( + def execute_activity( self, activity: str, result_type: Type[ResultType], *args: Any, **kwargs: Unpack[ActivityOptions], - ) -> ResultType: - return await super().execute_activity( - activity, result_type, *_prefix(args), **kwargs - ) + ) -> WorkflowFuture[ResultType]: + return super().execute_activity(activity, result_type, *_prefix(args), **kwargs) - async def execute_child_workflow( + def execute_child_workflow( self, workflow_type: str, result_type: Type[ResultType], *args: Any, **kwargs: Unpack[ChildWorkflowOptions], - ) -> ResultType: - return await super().execute_child_workflow( + ) -> ChildWorkflowFuture[ResultType]: + return super().execute_child_workflow( workflow_type, result_type, *_prefix(args), **kwargs ) @@ -430,3 +443,13 @@ async def test_workflow_interceptor_wraps_root_and_child_workflows(): assert ( env.get_workflow_result(str) == "workflow:hello intercepted-intercepted-x" ) + + +@pytest.mark.asyncio +async def test_workflow_interceptor_sees_start_child_workflow(): + with TestWorkflowEnvironment( + registry, workflow_interceptor_factory=PrefixingInterceptor + ) as env: + await env.client.start_workflow("StartChildWorkflow", "x", task_list="tl") + + assert env.get_workflow_result(str) == "child:intercepted-x" diff --git a/tests/integration_tests/nondeterminism/test_nondeterministic_workflows.py b/tests/integration_tests/nondeterminism/test_nondeterministic_workflows.py index 175b63d6..f28816b0 100644 --- a/tests/integration_tests/nondeterminism/test_nondeterministic_workflows.py +++ b/tests/integration_tests/nondeterminism/test_nondeterministic_workflows.py @@ -55,29 +55,22 @@ async def run(self) -> str: await first_activity.with_options( schedule_to_close_timeout=timedelta(seconds=10) ).execute("1") # event id 5 - async with asyncio.TaskGroup() as tg: - tg.create_task( - parallel_one.with_options( - schedule_to_close_timeout=timedelta(seconds=10) - ).execute("2") - ) # event id 11 - tg.create_task( - parallel_two.with_options( - schedule_to_close_timeout=timedelta(seconds=10) - ).execute("3") - ) # event id 12 - async with asyncio.TaskGroup() as tg: - tg.create_task( - parallel_three.with_options( - schedule_to_close_timeout=timedelta(seconds=10) - ).execute("4") - ) # event id 20 - tg.create_task( - parallel_three.with_options( - schedule_to_close_timeout=timedelta(seconds=10) - ).execute("5") - ) # event id 21 - + await asyncio.gather( + parallel_one.with_options( + schedule_to_close_timeout=timedelta(seconds=10) + ).execute("2"), # event id 11 + parallel_two.with_options( + schedule_to_close_timeout=timedelta(seconds=10) + ).execute("3"), # event id 12 + ) + await asyncio.gather( + parallel_three.with_options( + schedule_to_close_timeout=timedelta(seconds=10) + ).execute("4"), # event id 20 + parallel_three.with_options( + schedule_to_close_timeout=timedelta(seconds=10) + ).execute("5"), # event id 21 + ) return "hello world" @@ -124,28 +117,22 @@ class NDActivityRemoved: @workflow.run async def run(self) -> str: # await first_activity.with_options(schedule_to_close_timeout=timedelta(seconds=10)).execute("1") - async with asyncio.TaskGroup() as tg: - tg.create_task( - parallel_one.with_options( - schedule_to_close_timeout=timedelta(seconds=10) - ).execute("2") - ) - tg.create_task( - parallel_two.with_options( - schedule_to_close_timeout=timedelta(seconds=10) - ).execute("3") - ) - async with asyncio.TaskGroup() as tg: - tg.create_task( - parallel_three.with_options( - schedule_to_close_timeout=timedelta(seconds=10) - ).execute("4") - ) - tg.create_task( - parallel_three.with_options( - schedule_to_close_timeout=timedelta(seconds=10) - ).execute("5") - ) + await asyncio.gather( + parallel_one.with_options( + schedule_to_close_timeout=timedelta(seconds=10) + ).execute("2"), # event id 11 + parallel_two.with_options( + schedule_to_close_timeout=timedelta(seconds=10) + ).execute("3"), # event id 12 + ) + await asyncio.gather( + parallel_three.with_options( + schedule_to_close_timeout=timedelta(seconds=10) + ).execute("4"), # event id 20 + parallel_three.with_options( + schedule_to_close_timeout=timedelta(seconds=10) + ).execute("5"), # event id 21 + ) return "hello world" @@ -157,28 +144,22 @@ async def run(self) -> str: await parallel_one.with_options( schedule_to_close_timeout=timedelta(seconds=10) ).execute("1") # changed - async with asyncio.TaskGroup() as tg: - tg.create_task( - parallel_one.with_options( - schedule_to_close_timeout=timedelta(seconds=10) - ).execute("2") - ) - tg.create_task( - parallel_two.with_options( - schedule_to_close_timeout=timedelta(seconds=10) - ).execute("3") - ) - async with asyncio.TaskGroup() as tg: - tg.create_task( - parallel_three.with_options( - schedule_to_close_timeout=timedelta(seconds=10) - ).execute("4") - ) - tg.create_task( - parallel_three.with_options( - schedule_to_close_timeout=timedelta(seconds=10) - ).execute("5") - ) + await asyncio.gather( + parallel_one.with_options( + schedule_to_close_timeout=timedelta(seconds=10) + ).execute("2"), + parallel_two.with_options( + schedule_to_close_timeout=timedelta(seconds=10) + ).execute("3"), + ) + await asyncio.gather( + parallel_three.with_options( + schedule_to_close_timeout=timedelta(seconds=10) + ).execute("4"), + parallel_three.with_options( + schedule_to_close_timeout=timedelta(seconds=10) + ).execute("5"), + ) return "hello world" @@ -190,24 +171,20 @@ async def run(self) -> str: await first_activity.with_options( schedule_to_close_timeout=timedelta(seconds=10) ).execute("1") - async with asyncio.TaskGroup() as tg: - # tg.create_task(parallel_one.with_options(schedule_to_close_timeout=timedelta(seconds=10)).execute("2")) - tg.create_task( - parallel_two.with_options( - schedule_to_close_timeout=timedelta(seconds=10) - ).execute("3") - ) - async with asyncio.TaskGroup() as tg: - tg.create_task( - parallel_three.with_options( - schedule_to_close_timeout=timedelta(seconds=10) - ).execute("4") - ) - tg.create_task( - parallel_three.with_options( - schedule_to_close_timeout=timedelta(seconds=10) - ).execute("5") - ) + await asyncio.gather( + # parallel_one.with_options(schedule_to_close_timeout=timedelta(seconds=10)).execute("2"), + parallel_two.with_options( + schedule_to_close_timeout=timedelta(seconds=10) + ).execute("3"), + ) + await asyncio.gather( + parallel_three.with_options( + schedule_to_close_timeout=timedelta(seconds=10) + ).execute("4"), + parallel_three.with_options( + schedule_to_close_timeout=timedelta(seconds=10) + ).execute("5"), + ) return "hello world" @@ -219,33 +196,25 @@ async def run(self) -> str: await first_activity.with_options( schedule_to_close_timeout=timedelta(seconds=10) ).execute("1") - async with asyncio.TaskGroup() as tg: - tg.create_task( - parallel_one.with_options( - schedule_to_close_timeout=timedelta(seconds=10) - ).execute("2") - ) - tg.create_task( - parallel_two.with_options( - schedule_to_close_timeout=timedelta(seconds=10) - ).execute("3") - ) - tg.create_task( - parallel_three.with_options( - schedule_to_close_timeout=timedelta(seconds=10) - ).execute("4") - ) # Changed - async with asyncio.TaskGroup() as tg: - tg.create_task( - parallel_three.with_options( - schedule_to_close_timeout=timedelta(seconds=10) - ).execute("4") - ) - tg.create_task( - parallel_three.with_options( - schedule_to_close_timeout=timedelta(seconds=10) - ).execute("5") - ) + await asyncio.gather( + parallel_one.with_options( + schedule_to_close_timeout=timedelta(seconds=10) + ).execute("2"), + parallel_two.with_options( + schedule_to_close_timeout=timedelta(seconds=10) + ).execute("3"), + parallel_three.with_options( + schedule_to_close_timeout=timedelta(seconds=10) + ).execute("4"), # Changed + ) + await asyncio.gather( + parallel_three.with_options( + schedule_to_close_timeout=timedelta(seconds=10) + ).execute("4"), + parallel_three.with_options( + schedule_to_close_timeout=timedelta(seconds=10) + ).execute("5"), + ) return "hello world" @@ -257,29 +226,23 @@ async def run(self) -> str: await first_activity.with_options( schedule_to_close_timeout=timedelta(seconds=10) ).execute("1") - async with asyncio.TaskGroup() as tg: + await asyncio.gather( # reordered - tg.create_task( - parallel_two.with_options( - schedule_to_close_timeout=timedelta(seconds=10) - ).execute("3") - ) - tg.create_task( - parallel_one.with_options( - schedule_to_close_timeout=timedelta(seconds=10) - ).execute("2") - ) - async with asyncio.TaskGroup() as tg: - tg.create_task( - parallel_three.with_options( - schedule_to_close_timeout=timedelta(seconds=10) - ).execute("4") - ) - tg.create_task( - parallel_three.with_options( - schedule_to_close_timeout=timedelta(seconds=10) - ).execute("5") - ) + parallel_two.with_options( + schedule_to_close_timeout=timedelta(seconds=10) + ).execute("3"), + parallel_one.with_options( + schedule_to_close_timeout=timedelta(seconds=10) + ).execute("2"), + ) + await asyncio.gather( + parallel_three.with_options( + schedule_to_close_timeout=timedelta(seconds=10) + ).execute("4"), + parallel_three.with_options( + schedule_to_close_timeout=timedelta(seconds=10) + ).execute("5"), + ) return "hello world" @@ -291,24 +254,20 @@ async def run(self) -> str: await first_activity.with_options( schedule_to_close_timeout=timedelta(seconds=10) ).execute("1") - async with asyncio.TaskGroup() as tg: - tg.create_task( - parallel_one.with_options( - schedule_to_close_timeout=timedelta(seconds=10) - ).execute("2") - ) - tg.create_task( - parallel_two.with_options( - schedule_to_close_timeout=timedelta(seconds=10) - ).execute("3") - ) - async with asyncio.TaskGroup() as tg: - tg.create_task( - parallel_three.with_options( - schedule_to_close_timeout=timedelta(seconds=10) - ).execute("4") - ) - # tg.create_task(parallel_three.with_options(schedule_to_close_timeout=timedelta(seconds=10)).execute("5")) + await asyncio.gather( + parallel_one.with_options( + schedule_to_close_timeout=timedelta(seconds=10) + ).execute("2"), + parallel_two.with_options( + schedule_to_close_timeout=timedelta(seconds=10) + ).execute("3"), + ) + await asyncio.gather( + parallel_three.with_options( + schedule_to_close_timeout=timedelta(seconds=10) + ).execute("4"), + # parallel_three.with_options(schedule_to_close_timeout=timedelta(seconds=10)).execute("5"), + ) return "hello world" @@ -320,17 +279,14 @@ async def run(self) -> str: await first_activity.with_options( schedule_to_close_timeout=timedelta(seconds=10) ).execute("1") - async with asyncio.TaskGroup() as tg: - tg.create_task( - parallel_one.with_options( - schedule_to_close_timeout=timedelta(seconds=10) - ).execute("2") - ) - tg.create_task( - parallel_two.with_options( - schedule_to_close_timeout=timedelta(seconds=10) - ).execute("3") - ) + await asyncio.gather( + parallel_one.with_options( + schedule_to_close_timeout=timedelta(seconds=10) + ).execute("2"), + parallel_two.with_options( + schedule_to_close_timeout=timedelta(seconds=10) + ).execute("3"), + ) # Skipping an entire decision task @@ -353,28 +309,22 @@ async def run(self) -> str: await first_activity.with_options( schedule_to_close_timeout=timedelta(seconds=10) ).execute("1") - async with asyncio.TaskGroup() as tg: - tg.create_task( - parallel_one.with_options( - schedule_to_close_timeout=timedelta(seconds=10) - ).execute("2") - ) - tg.create_task( - parallel_two.with_options( - schedule_to_close_timeout=timedelta(seconds=10) - ).execute("3") - ) - async with asyncio.TaskGroup() as tg: - tg.create_task( - parallel_three.with_options( - schedule_to_close_timeout=timedelta(seconds=10) - ).execute("4") - ) - tg.create_task( - parallel_three.with_options( - schedule_to_close_timeout=timedelta(seconds=10) - ).execute("5") - ) + await asyncio.gather( + parallel_one.with_options( + schedule_to_close_timeout=timedelta(seconds=10) + ).execute("2"), + parallel_two.with_options( + schedule_to_close_timeout=timedelta(seconds=10) + ).execute("3"), + ) + await asyncio.gather( + parallel_three.with_options( + schedule_to_close_timeout=timedelta(seconds=10) + ).execute("4"), + parallel_three.with_options( + schedule_to_close_timeout=timedelta(seconds=10) + ).execute("5"), + ) # Extra activity at the end await parallel_one.with_options( @@ -391,28 +341,22 @@ async def run(self) -> str: await first_activity.with_options( schedule_to_close_timeout=timedelta(seconds=10) ).execute("1") - async with asyncio.TaskGroup() as tg: - tg.create_task( - parallel_one.with_options( - schedule_to_close_timeout=timedelta(seconds=10) - ).execute("2") - ) - tg.create_task( - parallel_two.with_options( - schedule_to_close_timeout=timedelta(seconds=10) - ).execute("3") - ) - async with asyncio.TaskGroup() as tg: - tg.create_task( - parallel_three.with_options( - schedule_to_close_timeout=timedelta(seconds=10) - ).execute("4") - ) - tg.create_task( - parallel_three.with_options( - schedule_to_close_timeout=timedelta(seconds=10) - ).execute("5") - ) + await asyncio.gather( + parallel_one.with_options( + schedule_to_close_timeout=timedelta(seconds=10) + ).execute("2"), + parallel_two.with_options( + schedule_to_close_timeout=timedelta(seconds=10) + ).execute("3"), + ) + await asyncio.gather( + parallel_three.with_options( + schedule_to_close_timeout=timedelta(seconds=10) + ).execute("4"), + parallel_three.with_options( + schedule_to_close_timeout=timedelta(seconds=10) + ).execute("5"), + ) raise ValueError("failure!") # changed diff --git a/tests/integration_tests/workflow/test_heartbeat.py b/tests/integration_tests/workflow/test_heartbeat.py index 37c6229a..bb30a1b5 100644 --- a/tests/integration_tests/workflow/test_heartbeat.py +++ b/tests/integration_tests/workflow/test_heartbeat.py @@ -59,19 +59,17 @@ def sync_wait_cancel_activity() -> str: class SyncWaitForCancelledWorkflow: @workflow.run async def run(self) -> str: - activity_task = asyncio.create_task( - sync_wait_cancel_activity.with_options( - schedule_to_close_timeout=timedelta(seconds=30), - start_to_close_timeout=timedelta(seconds=30), - heartbeat_timeout=timedelta(seconds=2), - ).execute() - ) + activity_future = sync_wait_cancel_activity.with_options( + schedule_to_close_timeout=timedelta(seconds=30), + start_to_close_timeout=timedelta(seconds=30), + heartbeat_timeout=timedelta(seconds=2), + ).execute() await workflow.sleep(timedelta(seconds=1)) - activity_task.cancel() + activity_future.cancel() try: - await activity_task + await activity_future except asyncio.CancelledError: return "activity cancelled" @@ -82,19 +80,17 @@ async def run(self) -> str: class ActivityCancellationWorkflow: @workflow.run async def run(self) -> str: - activity_task = asyncio.create_task( - cancellable_heartbeat_activity.with_options( - schedule_to_close_timeout=timedelta(seconds=30), - start_to_close_timeout=timedelta(seconds=30), - heartbeat_timeout=timedelta(seconds=2), - ).execute() - ) + activity_future = cancellable_heartbeat_activity.with_options( + schedule_to_close_timeout=timedelta(seconds=30), + start_to_close_timeout=timedelta(seconds=30), + heartbeat_timeout=timedelta(seconds=2), + ).execute() await workflow.sleep(timedelta(seconds=1)) - activity_task.cancel() + activity_future.cancel() try: - await activity_task + await activity_future except asyncio.CancelledError: return "activity cancelled" diff --git a/tests/integration_tests/workflow/test_interceptor.py b/tests/integration_tests/workflow/test_interceptor.py index 6cd98934..a20f52c3 100644 --- a/tests/integration_tests/workflow/test_interceptor.py +++ b/tests/integration_tests/workflow/test_interceptor.py @@ -9,10 +9,12 @@ ) from cadence.workflow import ( ActivityOptions, + ChildWorkflowFuture, ChildWorkflowOptions, ResultType, SearchAttributeType, WorkflowContext, + WorkflowFuture, WorkflowInterceptor, ) from tests.integration_tests.helper import CadenceHelper, DOMAIN_NAME @@ -59,25 +61,25 @@ def _rewrite(args: tuple[Any, ...]) -> tuple[Any, ...]: class RewritingInterceptor(WorkflowInterceptor): - async def execute_activity( + def execute_activity( self, activity: str, result_type: Type[ResultType], *args: Any, **kwargs: Unpack[ActivityOptions], - ) -> ResultType: - return await super().execute_activity( + ) -> WorkflowFuture[ResultType]: + return super().execute_activity( activity, result_type, *_rewrite(args), **kwargs ) - async def execute_child_workflow( + def execute_child_workflow( self, workflow_type: str, result_type: Type[ResultType], *args: Any, **kwargs: Unpack[ChildWorkflowOptions], - ) -> ResultType: - return await super().execute_child_workflow( + ) -> ChildWorkflowFuture[ResultType]: + return super().execute_child_workflow( workflow_type, result_type, *_rewrite(args), **kwargs ) diff --git a/tests/integration_tests/workflow/test_timer.py b/tests/integration_tests/workflow/test_timer.py index 9b2ed3b0..8b251039 100644 --- a/tests/integration_tests/workflow/test_timer.py +++ b/tests/integration_tests/workflow/test_timer.py @@ -1,4 +1,3 @@ -import asyncio from datetime import timedelta from cadence import Registry, workflow @@ -31,9 +30,9 @@ async def run(self) -> str: class TimerCancelWorkflow: @workflow.run async def run(self) -> str: - task = asyncio.create_task(workflow.sleep(timedelta(seconds=1))) + timer = workflow.sleep(timedelta(seconds=1)) await echo("hello") - task.cancel() + timer.cancel() return "hello"