From c301d52867c35ab60d6418e4d389332ce3880936 Mon Sep 17 00:00:00 2001 From: Nathanael Mortensen Date: Mon, 3 Aug 2026 15:23:17 -0700 Subject: [PATCH] feat!: Return Futures Instead of Coroutines Add WorkflowFuture, a covariant asyncio.Future subclass, to cadence.workflow. Update everything to use it rather than Awaitable or async. Returning coroutines results in confusing behavior, especially as Cadence enforces strict determinism. Decisions are scheduled not when someone calls a method, but when they await on the returned coroutine. --- cadence/_internal/activity/_definition.py | 11 +- .../workflow/child_workflow_future.py | 62 +-- cadence/_internal/workflow/context.py | 76 ++-- .../statemachine/activity_state_machine.py | 9 +- .../child_workflow_execution_state_machine.py | 10 +- .../workflow/statemachine/decision_manager.py | 52 ++- .../statemachine/decision_state_machine.py | 22 +- cadence/_internal/workflow/waiter.py | 32 +- cadence/activity.py | 4 +- cadence/testing/_workflow_environment.py | 135 ++++--- cadence/workflow/__init__.py | 2 + cadence/workflow/_api.py | 52 ++- cadence/workflow/_context.py | 30 +- cadence/workflow/_interceptor.py | 47 +-- cadence/workflow/_types.py | 49 ++- .../statemachine/test_decision_manager.py | 157 ++++++-- .../workflow/test_context_child_workflow.py | 281 +++++++------- .../workflow/test_context_retry_policy.py | 9 +- .../test_context_signal_child_workflow.py | 106 +++--- .../cadence/_internal/workflow/test_waiter.py | 17 + .../testing/test_workflow_environment.py | 39 +- .../test_nondeterministic_workflows.py | 358 ++++++++---------- .../workflow/test_heartbeat.py | 32 +- .../workflow/test_interceptor.py | 14 +- .../integration_tests/workflow/test_timer.py | 5 +- 25 files changed, 883 insertions(+), 728 deletions(-) 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"