Merge branch 'main' into heartbeat

Author: timl3136

cadence/_internal/workflow/context.py

@@ -5,7 +5,10 @@
 
 from cadence._internal.workflow.statemachine.decision_manager import DecisionManager
 from cadence.api.v1.common_pb2 import ActivityType
-from cadence.api.v1.decision_pb2 import ScheduleActivityTaskDecisionAttributes
+from cadence.api.v1.decision_pb2 import (
+    ScheduleActivityTaskDecisionAttributes,
+    StartTimerDecisionAttributes,
+)
 from cadence.api.v1.tasklist_pb2 import TaskList, TaskListKind
 from cadence.data_converter import DataConverter
 from cadence.workflow import WorkflowContext, WorkflowInfo, ResultType, ActivityOptions
@@ -97,6 +100,15 @@ async def execute_activity(
 
         return cast(ResultType, result)
 
+    async def start_timer(self, duration: timedelta):
+        if duration.total_seconds() <= 0:  # shortcut
+            return
+        await self._decision_manager.start_timer(
+            StartTimerDecisionAttributes(
+                start_to_fire_timeout=duration,
+            )
+        )
+
     def set_replay_mode(self, replay: bool) -> None:
         """Set whether the workflow is currently in replay mode."""
         self._replay_mode = replay

cadence/workflow.py

@@ -44,6 +44,10 @@ async def execute_activity(
     )
 
 
+async def sleep(duration: timedelta) -> None:
+    return await WorkflowContext.get().start_timer(duration)
+
+
 T = TypeVar("T", bound=Callable[..., Any])
 C = TypeVar("C")
 
@@ -277,6 +281,9 @@ async def execute_activity(
         **kwargs: Unpack[ActivityOptions],
     ) -> ResultType: ...
 
+    @abstractmethod
+    async def start_timer(self, duration: timedelta) -> None: ...
+
     @contextmanager
     def _activate(self) -> Iterator["WorkflowContext"]:
         token = WorkflowContext._var.set(self)

tests/integration_tests/workflow/test_timer.py

@@ -0,0 +1,135 @@
+import asyncio
+from datetime import timedelta
+
+from cadence import Registry, workflow
+from cadence.api.v1.history_pb2 import EventFilterType
+from cadence.api.v1.service_workflow_pb2 import (
+    GetWorkflowExecutionHistoryRequest,
+    GetWorkflowExecutionHistoryResponse,
+)
+from tests.integration_tests.helper import CadenceHelper
+
+
+registry = Registry()
+
+
+@registry.activity()
+async def echo(message: str) -> str:
+    return message
+
+
+@registry.workflow()
+class TimerWorkflow:
+    @workflow.run
+    async def run(self) -> str:
+        await workflow.sleep(timedelta(seconds=1))
+        await echo("hello")
+        return "hello"
+
+
+@registry.workflow()
+class TimerCancelWorkflow:
+    @workflow.run
+    async def run(self) -> str:
+        task = asyncio.create_task(workflow.sleep(timedelta(seconds=1)))
+        await echo("hello")
+        task.cancel()
+        return "hello"
+
+
+async def test_timer(helper: CadenceHelper):
+    async with helper.worker(registry) as worker:
+        execution = await worker.client.start_workflow(
+            "TimerWorkflow",
+            task_list=worker.task_list,
+            execution_start_to_close_timeout=timedelta(seconds=10),
+        )
+
+        # wait for close event
+        await worker.client.workflow_stub.GetWorkflowExecutionHistory(
+            GetWorkflowExecutionHistoryRequest(
+                domain=worker.client.domain,
+                workflow_execution=execution,
+                wait_for_new_event=True,
+                history_event_filter_type=EventFilterType.EVENT_FILTER_TYPE_CLOSE_EVENT,
+                skip_archival=True,
+            )
+        )
+
+        response: GetWorkflowExecutionHistoryResponse = (
+            await worker.client.workflow_stub.GetWorkflowExecutionHistory(
+                GetWorkflowExecutionHistoryRequest(
+                    domain=worker.client.domain,
+                    workflow_execution=execution,
+                )
+            )
+        )
+
+        events = response.history.events
+
+        timer_started_events = [
+            e for e in events if e.HasField("timer_started_event_attributes")
+        ]
+        timer_fired_events = [
+            e for e in events if e.HasField("timer_fired_event_attributes")
+        ]
+        assert len(timer_started_events) == 1
+        assert len(timer_fired_events) == 1
+
+        activity_scheduled_events = [
+            e for e in events if e.HasField("activity_task_scheduled_event_attributes")
+        ]
+        assert len(activity_scheduled_events) == 1
+
+        timer_started_time = timer_started_events[0].event_time.ToDatetime()
+        activity_scheduled_time = activity_scheduled_events[0].event_time.ToDatetime()
+        assert activity_scheduled_time >= timer_started_time + timedelta(seconds=1)
+
+
+async def test_timer_cancel(helper: CadenceHelper):
+    async with helper.worker(registry) as worker:
+        execution = await worker.client.start_workflow(
+            "TimerCancelWorkflow",
+            task_list=worker.task_list,
+            execution_start_to_close_timeout=timedelta(seconds=10),
+        )
+
+        # wait for close event
+        await worker.client.workflow_stub.GetWorkflowExecutionHistory(
+            GetWorkflowExecutionHistoryRequest(
+                domain=worker.client.domain,
+                workflow_execution=execution,
+                wait_for_new_event=True,
+                history_event_filter_type=EventFilterType.EVENT_FILTER_TYPE_CLOSE_EVENT,
+                skip_archival=True,
+            )
+        )
+
+        response: GetWorkflowExecutionHistoryResponse = (
+            await worker.client.workflow_stub.GetWorkflowExecutionHistory(
+                GetWorkflowExecutionHistoryRequest(
+                    domain=worker.client.domain,
+                    workflow_execution=execution,
+                )
+            )
+        )
+
+        events = response.history.events
+
+        timer_started_events = [
+            e for e in events if e.HasField("timer_started_event_attributes")
+        ]
+        timer_canceled_events = [
+            e for e in events if e.HasField("timer_canceled_event_attributes")
+        ]
+        assert len(timer_started_events) == 1
+        assert len(timer_canceled_events) == 1
+
+        activity_scheduled_events = [
+            e for e in events if e.HasField("activity_task_scheduled_event_attributes")
+        ]
+        assert len(activity_scheduled_events) == 1
+
+        timer_started_time = timer_started_events[0].event_time.ToDatetime()
+        activity_scheduled_time = activity_scheduled_events[0].event_time.ToDatetime()
+        assert activity_scheduled_time < timer_started_time + timedelta(seconds=1)