diff --git a/CHANGELOG.md b/CHANGELOG.md index d0d80bc71..c06223aaf 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -20,6 +20,11 @@ to include examples, links to docs, or any other relevant information. ### Added +#### Standalone Activity operator commands + +- `ActivityHandle` now supports operator commands for standalone activities: `pause`, + `unpause`, `update_options` and `restore_original_options`. + ### Changed - System Nexus Signal-with-Start Workflow operations now use the typed diff --git a/temporalio/client/__init__.py b/temporalio/client/__init__.py index 86030b1a8..2eef41a39 100644 --- a/temporalio/client/__init__.py +++ b/temporalio/client/__init__.py @@ -51,8 +51,12 @@ ActivityExecutionCount, ActivityExecutionCountAggregationGroup, ActivityExecutionDescription, + ActivityExecutionOptions, ActivityExecutionStatus, ActivityHandle, + ActivityOptionsKey, + ActivityOptionsKeys, + ActivityOptionsUpdate, AsyncActivityHandle, AsyncActivityIDReference, PendingActivityState, @@ -117,6 +121,7 @@ ListSchedulesInput, ListWorkflowsInput, OutboundInterceptor, + PauseActivityInput, PauseScheduleInput, QueryWorkflowInput, ReportCancellationAsyncActivityInput, @@ -130,7 +135,9 @@ TerminateNexusOperationInput, TerminateWorkflowInput, TriggerScheduleInput, + UnpauseActivityInput, UnpauseScheduleInput, + UpdateActivityOptionsInput, UpdateScheduleInput, UpdateWithStartStartWorkflowInput, UpdateWithStartUpdateWorkflowInput, @@ -227,6 +234,10 @@ "ActivityExecutionAsyncIterator", "ActivityExecution", "ActivityExecutionDescription", + "ActivityExecutionOptions", + "ActivityOptionsKey", + "ActivityOptionsKeys", + "ActivityOptionsUpdate", "ActivityExecutionStatus", "PendingActivityState", "ActivityExecutionCount", @@ -292,6 +303,9 @@ "CancelActivityInput", "TerminateActivityInput", "DescribeActivityInput", + "PauseActivityInput", + "UpdateActivityOptionsInput", + "UnpauseActivityInput", "ListActivitiesInput", "CountActivitiesInput", "StartNexusOperationInput", diff --git a/temporalio/client/_activity.py b/temporalio/client/_activity.py index 138d09dc7..dbc3f8840 100644 --- a/temporalio/client/_activity.py +++ b/temporalio/client/_activity.py @@ -16,6 +16,7 @@ TYPE_CHECKING, Any, Generic, + TypeVar, cast, ) @@ -49,8 +50,11 @@ DescribeActivityInput, FailAsyncActivityInput, HeartbeatAsyncActivityInput, + PauseActivityInput, ReportCancellationAsyncActivityInput, TerminateActivityInput, + UnpauseActivityInput, + UpdateActivityOptionsInput, ) if TYPE_CHECKING: @@ -447,6 +451,9 @@ class ActivityExecutionStatus(IntEnum): TIMED_OUT = int( temporalio.api.enums.v1.ActivityExecutionStatus.ACTIVITY_EXECUTION_STATUS_TIMED_OUT ) + PAUSED = int( + temporalio.api.enums.v1.ActivityExecutionStatus.ACTIVITY_EXECUTION_STATUS_PAUSED + ) class PendingActivityState(IntEnum): @@ -478,6 +485,162 @@ class PendingActivityState(IntEnum): ) +ActivityOptionValueType = TypeVar("ActivityOptionValueType") + + +@dataclass(frozen=True) +class ActivityOptionsKey(Generic[ActivityOptionValueType]): + """Typed key for one updatable activity option. + + Use the keys on :py:class:`ActivityOptionsKeys` rather than constructing + these directly. + + .. warning:: + This API is experimental. + """ + + name: str + """Field-mask path this key updates.""" + + def value_set( + self, value: ActivityOptionValueType + ) -> ActivityOptionsUpdate[ActivityOptionValueType]: + """Create an update that sets this option to the given value.""" + return ActivityOptionsUpdate(self, value) + + def value_unset(self) -> ActivityOptionsUpdate[ActivityOptionValueType]: + """Create an update that clears this option server-side.""" + return ActivityOptionsUpdate(self, None) + + +@dataclass(frozen=True) +class ActivityOptionsUpdate(Generic[ActivityOptionValueType]): + """A single change to an activity's options. + + An option not represented by any update in the call is left untouched; an + update carrying None clears the option. + + .. warning:: + This API is experimental. + """ + + key: ActivityOptionsKey[ActivityOptionValueType] + """Option being changed.""" + + value: ActivityOptionValueType | None + """Value being set, or None to clear the option.""" + + +class ActivityOptionsKeys: + """The activity options that :py:meth:`ActivityHandle.update_options` can change. + + .. warning:: + This API is experimental. + """ + + task_queue: ActivityOptionsKey[str] = ActivityOptionsKey("task_queue.name") + schedule_to_close_timeout: ActivityOptionsKey[timedelta] = ActivityOptionsKey( + "schedule_to_close_timeout" + ) + schedule_to_start_timeout: ActivityOptionsKey[timedelta] = ActivityOptionsKey( + "schedule_to_start_timeout" + ) + start_to_close_timeout: ActivityOptionsKey[timedelta] = ActivityOptionsKey( + "start_to_close_timeout" + ) + heartbeat_timeout: ActivityOptionsKey[timedelta] = ActivityOptionsKey( + "heartbeat_timeout" + ) + start_delay: ActivityOptionsKey[timedelta] = ActivityOptionsKey("start_delay") + retry_policy: ActivityOptionsKey[temporalio.common.RetryPolicy] = ( + ActivityOptionsKey("retry_policy") + ) + priority: ActivityOptionsKey[temporalio.common.Priority] = ActivityOptionsKey( + "priority" + ) + + +@dataclass(frozen=True) +class ActivityExecutionOptions: + """An activity's options as resolved by the server. + + Returned by :py:meth:`ActivityHandle.update_options` and + :py:meth:`ActivityHandle.restore_original_options`. + + .. warning:: + This API is experimental. + """ + + task_queue: str | None + """Task queue the activity is scheduled on.""" + + schedule_to_close_timeout: timedelta | None + """Total time the caller is willing to wait, including retries.""" + + schedule_to_start_timeout: timedelta | None + """Maximum time the activity may wait to be picked up by a worker.""" + + start_to_close_timeout: timedelta | None + """Maximum time for a single attempt.""" + + heartbeat_timeout: timedelta | None + """Maximum allowed time between heartbeats.""" + + start_delay: timedelta | None + """Delay before the first attempt is made available for dispatch.""" + + retry_policy: temporalio.common.RetryPolicy | None + """Retry policy in effect for the activity.""" + + priority: temporalio.common.Priority | None + """Priority of the activity.""" + + @staticmethod + def _from_proto( + options: temporalio.api.activity.v1.ActivityOptions, + ) -> ActivityExecutionOptions: + return ActivityExecutionOptions( + task_queue=options.task_queue.name + if options.HasField("task_queue") + else None, + schedule_to_close_timeout=( + options.schedule_to_close_timeout.ToTimedelta() + if options.HasField("schedule_to_close_timeout") + else None + ), + schedule_to_start_timeout=( + options.schedule_to_start_timeout.ToTimedelta() + if options.HasField("schedule_to_start_timeout") + else None + ), + start_to_close_timeout=( + options.start_to_close_timeout.ToTimedelta() + if options.HasField("start_to_close_timeout") + else None + ), + heartbeat_timeout=( + options.heartbeat_timeout.ToTimedelta() + if options.HasField("heartbeat_timeout") + else None + ), + start_delay=( + options.start_delay.ToTimedelta() + if options.HasField("start_delay") + else None + ), + retry_policy=( + temporalio.common.RetryPolicy.from_proto(options.retry_policy) + if options.HasField("retry_policy") + else None + ), + priority=( + temporalio.common.Priority._from_proto(options.priority) + if options.HasField("priority") + else None + ), + ) + + @dataclass(frozen=True) class ActivityExecutionCount: """Representation of a count from a count activities call. @@ -886,6 +1049,143 @@ async def terminate( ) ) + async def pause( + self, + *, + reason: str | None = None, + rpc_metadata: Mapping[str, str | bytes] = {}, + rpc_timeout: timedelta | None = None, + ) -> None: + """Pause the activity. + + A paused activity is not scheduled or retried until it is unpaused via + :py:meth:`unpause`. + + .. warning:: + This API is experimental. + + Args: + reason: Reason for pausing. Recorded and available via describe. + rpc_metadata: Headers used on the RPC call. + rpc_timeout: Optional RPC deadline to set for the RPC call. + """ + await self._client._impl.pause_activity( + PauseActivityInput( + activity_id=self._id, + activity_run_id=self._run_id, + reason=reason, + rpc_metadata=rpc_metadata, + rpc_timeout=rpc_timeout, + ) + ) + + async def unpause( + self, + *, + reason: str | None = None, + jitter: timedelta | None = None, + rpc_metadata: Mapping[str, str | bytes] = {}, + rpc_timeout: timedelta | None = None, + ) -> None: + """Unpause the activity, allowing it to be scheduled or retried again. + + .. warning:: + This API is experimental. + + Args: + reason: Reason for unpausing. Recorded on the server. + jitter: If set, the activity starts at a random time within this + duration rather than immediately. + rpc_metadata: Headers used on the RPC call. + rpc_timeout: Optional RPC deadline to set for the RPC call. + """ + await self._client._impl.unpause_activity( + UnpauseActivityInput( + activity_id=self._id, + activity_run_id=self._run_id, + reason=reason, + jitter=jitter, + rpc_metadata=rpc_metadata, + rpc_timeout=rpc_timeout, + ) + ) + + async def update_options( + self, + updates: Sequence[ActivityOptionsUpdate[Any]], + *, + rpc_metadata: Mapping[str, str | bytes] = {}, + rpc_timeout: timedelta | None = None, + ) -> ActivityExecutionOptions: + """Update the activity's options. + + Only the options named by ``updates`` are changed; anything not named is + left as-is. An update created with + :py:meth:`ActivityOptionsKey.value_unset` clears that option. + + .. warning:: + This API is experimental. + + Args: + updates: Options to change, built from :py:class:`ActivityOptionsKeys`. + rpc_metadata: Headers used on the RPC call. + rpc_timeout: Optional RPC deadline to set for the RPC call. + + Returns: + The activity options as resolved by the server after the update. + + Raises: + ValueError: If ``updates`` is empty. + """ + if not updates: + raise ValueError( + "update_options requires at least one update; use " + "restore_original_options() to revert options" + ) + return await self._client._impl.update_activity_options( + UpdateActivityOptionsInput( + activity_id=self._id, + activity_run_id=self._run_id, + updates=updates, + restore_original=False, + rpc_metadata=rpc_metadata, + rpc_timeout=rpc_timeout, + ) + ) + + async def restore_original_options( + self, + *, + rpc_metadata: Mapping[str, str | bytes] = {}, + rpc_timeout: timedelta | None = None, + ) -> ActivityExecutionOptions: + """Restore the activity's options to the ones it was created with. + + This is a separate call rather than an option on + :py:meth:`update_options` because the server rejects a request that + combines the restore flag with any other option. + + .. warning:: + This API is experimental. + + Args: + rpc_metadata: Headers used on the RPC call. + rpc_timeout: Optional RPC deadline to set for the RPC call. + + Returns: + The activity options as resolved by the server after the restore. + """ + return await self._client._impl.update_activity_options( + UpdateActivityOptionsInput( + activity_id=self._id, + activity_run_id=self._run_id, + updates=[], + restore_original=True, + rpc_metadata=rpc_metadata, + rpc_timeout=rpc_timeout, + ) + ) + async def describe( self, *, diff --git a/temporalio/client/_impl.py b/temporalio/client/_impl.py index 78471baf7..09d443727 100644 --- a/temporalio/client/_impl.py +++ b/temporalio/client/_impl.py @@ -50,7 +50,9 @@ ActivityExecutionAsyncIterator, ActivityExecutionCount, ActivityExecutionDescription, + ActivityExecutionOptions, ActivityHandle, + ActivityOptionsUpdate, AsyncActivityIDReference, ) from ._exceptions import ( @@ -87,6 +89,7 @@ ListSchedulesInput, ListWorkflowsInput, OutboundInterceptor, + PauseActivityInput, PauseScheduleInput, QueryWorkflowInput, ReportCancellationAsyncActivityInput, @@ -100,7 +103,9 @@ TerminateNexusOperationInput, TerminateWorkflowInput, TriggerScheduleInput, + UnpauseActivityInput, UnpauseScheduleInput, + UpdateActivityOptionsInput, UpdateScheduleInput, UpdateWithStartStartWorkflowInput, UpdateWithStartUpdateWorkflowInput, @@ -681,6 +686,94 @@ async def terminate_activity(self, input: TerminateActivityInput) -> None: timeout=input.rpc_timeout, ) + async def pause_activity(self, input: PauseActivityInput) -> None: + """Pause an activity.""" + await self._client.workflow_service.pause_activity_execution( + temporalio.api.workflowservice.v1.PauseActivityExecutionRequest( + namespace=self._client.namespace, + activity_id=input.activity_id, + run_id=input.activity_run_id or "", + identity=self._client.identity, + request_id=str(uuid.uuid4()), + reason=input.reason or "", + ), + retry=True, + metadata=input.rpc_metadata, + timeout=input.rpc_timeout, + ) + + async def unpause_activity(self, input: UnpauseActivityInput) -> None: + """Unpause an activity.""" + req = temporalio.api.workflowservice.v1.UnpauseActivityExecutionRequest( + namespace=self._client.namespace, + activity_id=input.activity_id, + run_id=input.activity_run_id or "", + identity=self._client.identity, + request_id=str(uuid.uuid4()), + reason=input.reason or "", + ) + if input.jitter is not None: + req.jitter.FromTimedelta(input.jitter) + await self._client.workflow_service.unpause_activity_execution( + req, + retry=True, + metadata=input.rpc_metadata, + timeout=input.rpc_timeout, + ) + + async def update_activity_options( + self, input: UpdateActivityOptionsInput + ) -> ActivityExecutionOptions: + """Update or restore an activity's options.""" + # restore_original is exclusive to all other updates. + if input.restore_original and input.updates: + raise ValueError( + "restore_original cannot be combined with individual option updates" + ) + req = temporalio.api.workflowservice.v1.UpdateActivityExecutionOptionsRequest( + namespace=self._client.namespace, + activity_id=input.activity_id, + run_id=input.activity_run_id or "", + identity=self._client.identity, + request_id=str(uuid.uuid4()), + ) + if input.restore_original: + req.restore_original = True + else: + # For repeated keys, later values override previous ones. + by_path: dict[str, ActivityOptionsUpdate[Any]] = {} + for update in input.updates: + by_path[update.key.name] = update + for name, update in by_path.items(): + req.update_mask.paths.append(name) + if update.value is None: + continue + if name == "task_queue.name": + req.activity_options.task_queue.name = update.value + elif name == "retry_policy": + update.value.apply_to_proto(req.activity_options.retry_policy) + elif name == "priority": + req.activity_options.priority.CopyFrom(update.value._to_proto()) + elif name in ( + "schedule_to_close_timeout", + "schedule_to_start_timeout", + "start_to_close_timeout", + "heartbeat_timeout", + "start_delay", + ): + getattr(req.activity_options, name).FromTimedelta(update.value) + else: + # Reached only if a key is added without a conversion for it here. + raise ValueError(f"No conversion for activity option {name!r}") + + resp = await self._client.workflow_service.update_activity_execution_options( + req, + retry=True, + metadata=input.rpc_metadata, + timeout=input.rpc_timeout, + ) + return ActivityExecutionOptions._from_proto(resp.activity_options) + async def describe_activity( self, input: DescribeActivityInput ) -> ActivityExecutionDescription: diff --git a/temporalio/client/_interceptor.py b/temporalio/client/_interceptor.py index a0310ac19..027b1ace8 100644 --- a/temporalio/client/_interceptor.py +++ b/temporalio/client/_interceptor.py @@ -27,7 +27,9 @@ ActivityExecutionAsyncIterator, ActivityExecutionCount, ActivityExecutionDescription, + ActivityExecutionOptions, ActivityHandle, + ActivityOptionsUpdate, AsyncActivityIDReference, ) from ._nexus import ( @@ -257,6 +259,53 @@ class TerminateActivityInput: rpc_timeout: timedelta | None +@dataclass +class PauseActivityInput: + """Input for :py:meth:`OutboundInterceptor.pause_activity`. + + .. warning:: + This API is experimental. + """ + + activity_id: str + activity_run_id: str | None + reason: str | None + rpc_metadata: Mapping[str, str | bytes] + rpc_timeout: timedelta | None + + +@dataclass +class UnpauseActivityInput: + """Input for :py:meth:`OutboundInterceptor.unpause_activity`. + + .. warning:: + This API is experimental. + """ + + activity_id: str + activity_run_id: str | None + reason: str | None + jitter: timedelta | None + rpc_metadata: Mapping[str, str | bytes] + rpc_timeout: timedelta | None + + +@dataclass +class UpdateActivityOptionsInput: + """Input for :py:meth:`OutboundInterceptor.update_activity_options`. + + .. warning:: + This API is experimental. + """ + + activity_id: str + activity_run_id: str | None + updates: Sequence[ActivityOptionsUpdate[Any]] + restore_original: bool + rpc_metadata: Mapping[str, str | bytes] + rpc_timeout: timedelta | None + + @dataclass class DescribeActivityInput: """Input for :py:meth:`OutboundInterceptor.describe_activity`. @@ -778,6 +827,33 @@ async def terminate_activity(self, input: TerminateActivityInput) -> None: """ await self.next.terminate_activity(input) + async def pause_activity(self, input: PauseActivityInput) -> None: + """Called for every :py:meth:`ActivityHandle.pause` call. + + .. warning:: + This API is experimental. + """ + await self.next.pause_activity(input) + + async def unpause_activity(self, input: UnpauseActivityInput) -> None: + """Called for every :py:meth:`ActivityHandle.unpause` call. + + .. warning:: + This API is experimental. + """ + await self.next.unpause_activity(input) + + async def update_activity_options( + self, input: UpdateActivityOptionsInput + ) -> ActivityExecutionOptions: + """Called for every :py:meth:`ActivityHandle.update_options` and + :py:meth:`ActivityHandle.restore_original_options` call. + + .. warning:: + This API is experimental. + """ + return await self.next.update_activity_options(input) + async def describe_activity( self, input: DescribeActivityInput ) -> ActivityExecutionDescription: diff --git a/tests/test_activity_operator_commands.py b/tests/test_activity_operator_commands.py new file mode 100644 index 000000000..1d2f40ef6 --- /dev/null +++ b/tests/test_activity_operator_commands.py @@ -0,0 +1,384 @@ +"""Tests for the standalone-activity operator commands: pause, unpause and +update_options. +""" + +from __future__ import annotations + +import asyncio +import uuid +from datetime import timedelta +from typing import Any + +import pytest + +import temporalio.api.workflowservice.v1 +from temporalio import activity +from temporalio.client import ( + ActivityExecutionStatus, + ActivityHandle, + ActivityOptionsKeys, + Client, + PendingActivityState, +) +from temporalio.common import Priority, RetryPolicy +from temporalio.exceptions import ApplicationError +from temporalio.testing import WorkflowEnvironment +from temporalio.worker import Worker +from tests.helpers import assert_eventually + +PAUSED_STATES = (PendingActivityState.PAUSED, PendingActivityState.PAUSE_REQUESTED) + + +@activity.defn +async def slow_activity() -> None: + """Long-running activity that heartbeats and runs until cancellation.""" + while True: + activity.heartbeat() + await asyncio.sleep(0.1) + + +@activity.defn +async def quick_activity() -> str: + """Returns immediately. Used with a start delay so it can be paused while scheduled.""" + return "resumed" + + +@activity.defn +async def echo_activity(word: str) -> str: + """Takes an argument and returns a value derived from it, so a completed execution + has both an input and a successful outcome to read back off describe.""" + return f"{word}-echoed" + + +@activity.defn +async def always_fail_activity() -> None: + """Always fails. Paired with a single-attempt retry policy so the activity reaches a + terminal failure outcome rather than retrying.""" + raise ApplicationError("deliberate failure") + + +@activity.defn +async def heartbeat_fail_increment(value: int) -> int: + """Heartbeats, fails the first attempt, then succeeds. + + The description will have input, a result, heartbeat details and a last failure. + """ + activity.heartbeat("heartbeat details") + if activity.info().attempt == 1: + raise ApplicationError("deliberate first-attempt failure") + return value + 1 + + +@activity.defn +async def heartbeat_once_activity() -> None: + """Records heartbeat details on attempt 1, then blocks waiting for cancellation.""" + if activity.info().attempt == 1: + activity.heartbeat("hb-details") + while True: + await asyncio.sleep(0.1) + + +def _skip_if_unsupported(env: WorkflowEnvironment) -> None: + if env.supports_time_skipping: + pytest.skip("Java test server does not support standalone activities") + + +async def _assert_eventually_paused(handle: ActivityHandle) -> None: + async def check() -> None: + desc = await handle.describe() + assert desc.run_state in PAUSED_STATES + + await assert_eventually(check) + + +async def _start_running_slow_activity( + client: Client, task_queue: str, **kwargs: Any +) -> ActivityHandle: + """Start a slow activity and wait until it is actually running on the worker.""" + kwargs.setdefault("start_to_close_timeout", timedelta(seconds=60)) + kwargs.setdefault("heartbeat_timeout", timedelta(seconds=30)) + handle = await client.start_activity( + slow_activity, + id=f"act-{uuid.uuid4()}", + task_queue=task_queue, + **kwargs, + ) + + async def check() -> None: + desc = await handle.describe() + assert desc.run_state == PendingActivityState.STARTED + + await assert_eventually(check) + return handle + + +async def test_unpause_resumes(client: Client, env: WorkflowEnvironment): + _skip_if_unsupported(env) + task_queue = str(uuid.uuid4()) + async with Worker(client, task_queue=task_queue, activities=[quick_activity]): + # Start delayed so the activity sits scheduled and can be paused before it runs. + handle = await client.start_activity( + quick_activity, + id=f"act-{uuid.uuid4()}", + task_queue=task_queue, + start_to_close_timeout=timedelta(seconds=60), + start_delay=timedelta(seconds=30), + ) + await handle.pause(reason="pause-before-unpause") + + # A not-yet-started (scheduled) activity transitions fully to PAUSED. + async def check() -> None: + desc = await handle.describe() + assert desc.run_state == PendingActivityState.PAUSED + + await assert_eventually(check) + + await handle.unpause() + + async def resumed() -> None: + desc = await handle.describe() + assert desc.run_state not in PAUSED_STATES + + await assert_eventually(resumed) + await handle.terminate(reason="cleanup") + + +async def test_describe_paused_activity_reports_paused_status( + client: Client, env: WorkflowEnvironment +): + _skip_if_unsupported(env) + task_queue = str(uuid.uuid4()) + async with Worker(client, task_queue=task_queue, activities=[quick_activity]): + # Start delayed so the activity sits scheduled; pausing from there reaches a true + # PAUSED state rather than the PAUSE_REQUESTED of a running activity. + handle = await client.start_activity( + quick_activity, + id=f"act-{uuid.uuid4()}", + task_queue=task_queue, + start_to_close_timeout=timedelta(seconds=60), + start_delay=timedelta(seconds=30), + ) + assert (await handle.describe()).status == ActivityExecutionStatus.RUNNING + + await handle.pause(reason="hold") + + async def check() -> None: + desc = await handle.describe() + assert desc.status == ActivityExecutionStatus.PAUSED + assert desc.run_state == PendingActivityState.PAUSED + + await assert_eventually(check) + await handle.terminate(reason="cleanup") + + +async def test_update_options_respects_mask(client: Client, env: WorkflowEnvironment): + _skip_if_unsupported(env) + task_queue = str(uuid.uuid4()) + async with Worker(client, task_queue=task_queue, activities=[slow_activity]): + handle = await _start_running_slow_activity( + client, + task_queue, + schedule_to_close_timeout=timedelta(seconds=120), + ) + + updated = await handle.update_options( + [ + ActivityOptionsKeys.start_to_close_timeout.value_set( + timedelta(seconds=90) + ) + ] + ) + + # Only start_to_close changed; schedule_to_close kept its original value. + assert updated.start_to_close_timeout == timedelta(seconds=90) + assert updated.schedule_to_close_timeout == timedelta(seconds=120) + + await handle.terminate(reason="cleanup") + + +async def test_update_options_all_fields(client: Client, env: WorkflowEnvironment): + _skip_if_unsupported(env) + task_queue = str(uuid.uuid4()) + async with Worker(client, task_queue=task_queue, activities=[quick_activity]): + # Start delayed so the activity stays scheduled while every option is updated. + handle = await client.start_activity( + quick_activity, + id=f"act-{uuid.uuid4()}", + task_queue=task_queue, + schedule_to_close_timeout=timedelta(seconds=100), + start_to_close_timeout=timedelta(seconds=30), + start_delay=timedelta(seconds=300), + ) + + updated = await handle.update_options( + [ + ActivityOptionsKeys.task_queue.value_set("updated-tq"), + ActivityOptionsKeys.schedule_to_close_timeout.value_set( + timedelta(seconds=200) + ), + ActivityOptionsKeys.schedule_to_start_timeout.value_set( + timedelta(seconds=15) + ), + ActivityOptionsKeys.start_to_close_timeout.value_set( + timedelta(seconds=90) + ), + ActivityOptionsKeys.heartbeat_timeout.value_set(timedelta(seconds=25)), + ActivityOptionsKeys.retry_policy.value_set( + RetryPolicy( + initial_interval=timedelta(seconds=1), + backoff_coefficient=2.0, + maximum_attempts=7, + ) + ), + ActivityOptionsKeys.priority.value_set(Priority(priority_key=3)), + ActivityOptionsKeys.start_delay.value_set(timedelta(seconds=500)), + ] + ) + + assert updated.task_queue == "updated-tq" + assert updated.schedule_to_close_timeout == timedelta(seconds=200) + assert updated.schedule_to_start_timeout == timedelta(seconds=15) + assert updated.start_to_close_timeout == timedelta(seconds=90) + assert updated.heartbeat_timeout == timedelta(seconds=25) + assert updated.retry_policy is not None + assert updated.retry_policy.maximum_attempts == 7 + assert updated.priority is not None + assert updated.priority.priority_key == 3 + assert updated.start_delay == timedelta(seconds=500) + + desc = await handle.describe() + assert desc.task_queue == "updated-tq" + await handle.terminate(reason="cleanup") + + +async def test_update_options_restore_original( + client: Client, env: WorkflowEnvironment +): + _skip_if_unsupported(env) + task_queue = str(uuid.uuid4()) + async with Worker(client, task_queue=task_queue, activities=[slow_activity]): + handle = await _start_running_slow_activity( + client, task_queue, start_to_close_timeout=timedelta(seconds=45) + ) + + changed = await handle.update_options( + [ + ActivityOptionsKeys.start_to_close_timeout.value_set( + timedelta(seconds=90) + ) + ] + ) + assert changed.start_to_close_timeout == timedelta(seconds=90) + + # Restore alone reverts to the value the activity was created with. + restored = await handle.restore_original_options() + assert restored.start_to_close_timeout == timedelta(seconds=45) + await handle.terminate(reason="cleanup") + + +async def test_update_options_on_paused_activity( + client: Client, env: WorkflowEnvironment +): + _skip_if_unsupported(env) + task_queue = str(uuid.uuid4()) + async with Worker(client, task_queue=task_queue, activities=[slow_activity]): + handle = await _start_running_slow_activity(client, task_queue) + await handle.pause(reason="hold") + await _assert_eventually_paused(handle) + + # Updating options while paused applies, and leaves the activity paused. + updated = await handle.update_options( + [ + ActivityOptionsKeys.start_to_close_timeout.value_set( + timedelta(seconds=99) + ) + ] + ) + assert updated.start_to_close_timeout == timedelta(seconds=99) + + desc = await handle.describe() + assert desc.run_state in PAUSED_STATES + await handle.terminate(reason="cleanup") + + +async def _heartbeat_detail_count(client: Client, handle: ActivityHandle) -> int: + """Count heartbeat payloads the server holds for an activity.""" + resp = await client.workflow_service.describe_activity_execution( + temporalio.api.workflowservice.v1.DescribeActivityExecutionRequest( + namespace=client.namespace, + activity_id=handle.id, + run_id=handle.run_id or "", + include_heartbeat_details=True, + ) + ) + return len(resp.info.heartbeat_details.payloads) + + +async def _start_heartbeat_ready_activity( + client: Client, task_queue: str +) -> ActivityHandle: + """Start a heartbeat-once activity and wait until it has recorded heartbeat details.""" + handle = await client.start_activity( + heartbeat_once_activity, + id=f"act-{uuid.uuid4()}", + task_queue=task_queue, + start_to_close_timeout=timedelta(seconds=60), + heartbeat_timeout=timedelta(seconds=30), + ) + + async def check() -> None: + assert await _heartbeat_detail_count(client, handle) > 0 + + await assert_eventually(check) + return handle + + +async def test_pause_preserves_heartbeat(client: Client, env: WorkflowEnvironment): + _skip_if_unsupported(env) + task_queue = str(uuid.uuid4()) + async with Worker( + client, task_queue=task_queue, activities=[heartbeat_once_activity] + ): + handle = await _start_heartbeat_ready_activity(client, task_queue) + await handle.pause(reason="hold") + await _assert_eventually_paused(handle) + + # Pause never touches heartbeat details. + assert await _heartbeat_detail_count(client, handle) == 1 + await handle.terminate(reason="cleanup") + + +async def test_unpause_preserves_heartbeat(client: Client, env: WorkflowEnvironment): + _skip_if_unsupported(env) + task_queue = str(uuid.uuid4()) + async with Worker( + client, task_queue=task_queue, activities=[heartbeat_once_activity] + ): + handle = await _start_heartbeat_ready_activity(client, task_queue) + await handle.pause(reason="hold") + await _assert_eventually_paused(handle) + await handle.unpause() + + assert await _heartbeat_detail_count(client, handle) == 1 + await handle.terminate(reason="cleanup") + + +async def test_update_options_preserves_heartbeat( + client: Client, env: WorkflowEnvironment +): + _skip_if_unsupported(env) + task_queue = str(uuid.uuid4()) + async with Worker( + client, task_queue=task_queue, activities=[heartbeat_once_activity] + ): + handle = await _start_heartbeat_ready_activity(client, task_queue) + await handle.update_options( + [ + ActivityOptionsKeys.start_to_close_timeout.value_set( + timedelta(seconds=90) + ) + ] + ) + + assert await _heartbeat_detail_count(client, handle) == 1 + await handle.terminate(reason="cleanup") diff --git a/tests/test_activity_operator_commands_interceptor.py b/tests/test_activity_operator_commands_interceptor.py new file mode 100644 index 000000000..3f2cc48f0 --- /dev/null +++ b/tests/test_activity_operator_commands_interceptor.py @@ -0,0 +1,116 @@ +from __future__ import annotations + +from datetime import timedelta +from typing import Any +from unittest.mock import AsyncMock, Mock + +import pytest + +import temporalio.api.activity.v1 +import temporalio.api.workflowservice.v1 +from temporalio.client import ( + ActivityOptionsKeys, + Client, + Interceptor, + OutboundInterceptor, + PauseActivityInput, + UnpauseActivityInput, + UpdateActivityOptionsInput, +) +from temporalio.service import ServiceClient + + +class TracingClientInterceptor(Interceptor): + def __init__(self) -> None: + super().__init__() + self.traces: list[tuple[str, Any]] = [] + + def intercept_client(self, next: OutboundInterceptor) -> OutboundInterceptor: + return TracingClientOutboundInterceptor(self, next) + + +class TracingClientOutboundInterceptor(OutboundInterceptor): + def __init__(self, parent: TracingClientInterceptor, next: OutboundInterceptor): + super().__init__(next) + self._parent = parent + + async def pause_activity(self, input: PauseActivityInput) -> None: + self._parent.traces.append(("pause_activity", input)) + return await super().pause_activity(input) + + async def unpause_activity(self, input: UnpauseActivityInput) -> None: + self._parent.traces.append(("unpause_activity", input)) + return await super().unpause_activity(input) + + async def update_activity_options(self, input: UpdateActivityOptionsInput) -> Any: + self._parent.traces.append(("update_activity_options", input)) + return await super().update_activity_options(input) + + +def _stub_service() -> Any: + service = Mock() + service.pause_activity_execution = AsyncMock( + return_value=temporalio.api.workflowservice.v1.PauseActivityExecutionResponse() + ) + service.unpause_activity_execution = AsyncMock( + return_value=temporalio.api.workflowservice.v1.UnpauseActivityExecutionResponse() + ) + service.update_activity_execution_options = AsyncMock( + return_value=temporalio.api.workflowservice.v1.UpdateActivityExecutionOptionsResponse( + activity_options=temporalio.api.activity.v1.ActivityOptions() + ) + ) + return service + + +@pytest.fixture +def interceptor() -> TracingClientInterceptor: + return TracingClientInterceptor() + + +@pytest.fixture +def client(interceptor: TracingClientInterceptor) -> Client: + service_client = Mock(spec=ServiceClient) + service_client.workflow_service = _stub_service() + service_client.config = Mock(identity="test-identity") + return Client( + service_client=service_client, + namespace="test-namespace", + interceptors=[interceptor], + ) + + +async def test_interceptor_invokes_each_operator_command( + client: Client, interceptor: TracingClientInterceptor +): + handle = client.get_activity_handle("act-1", run_id="run-1") + await handle.pause(reason="pause-reason") + await handle.unpause(reason="unpause-reason", jitter=timedelta(seconds=5)) + await handle.update_options( + [ActivityOptionsKeys.start_to_close_timeout.value_set(timedelta(seconds=90))] + ) + await handle.restore_original_options() + + assert [name for name, _ in interceptor.traces] == [ + "pause_activity", + "unpause_activity", + "update_activity_options", + "update_activity_options", + ] + + for name, input in interceptor.traces: + assert input.activity_id == "act-1", name + assert input.activity_run_id == "run-1", name + + +async def test_interceptor_receives_command_arguments( + client: Client, interceptor: TracingClientInterceptor +): + handle = client.get_activity_handle("act-1") + await handle.pause(reason="pause-reason") + await handle.unpause(reason="unpause-reason", jitter=timedelta(seconds=5)) + + traces = dict(interceptor.traces) + assert traces["pause_activity"].reason == "pause-reason" + assert traces["unpause_activity"].reason == "unpause-reason" + assert traces["unpause_activity"].jitter == timedelta(seconds=5) diff --git a/tests/test_activity_operator_commands_requests.py b/tests/test_activity_operator_commands_requests.py new file mode 100644 index 000000000..c6c0346b0 --- /dev/null +++ b/tests/test_activity_operator_commands_requests.py @@ -0,0 +1,183 @@ +"""Unit tests for the operator-command request fields the server does not surface back.""" + +from __future__ import annotations + +from datetime import timedelta +from typing import Any +from unittest.mock import AsyncMock, Mock + +import pytest + +import temporalio.api.activity.v1 +import temporalio.api.workflowservice.v1 +from temporalio.client import ActivityOptionsKeys, Client +from temporalio.service import ServiceClient + + +class _CapturedService: + """Capture operator command requests.""" + + def __init__(self) -> None: + self.requests: dict[str, Any] = {} + self.pause_activity_execution = self._recorder( + "pause", + temporalio.api.workflowservice.v1.PauseActivityExecutionResponse(), + ) + self.unpause_activity_execution = self._recorder( + "unpause", + temporalio.api.workflowservice.v1.UnpauseActivityExecutionResponse(), + ) + self.update_activity_execution_options = self._recorder( + "update", + temporalio.api.workflowservice.v1.UpdateActivityExecutionOptionsResponse( + activity_options=temporalio.api.activity.v1.ActivityOptions() + ), + ) + + def _recorder(self, name: str, response: Any) -> AsyncMock: + async def record(req: Any, **_kwargs: Any) -> Any: + self.requests[name] = req + return response + + return AsyncMock(side_effect=record) + + +@pytest.fixture +def captured() -> _CapturedService: + return _CapturedService() + + +@pytest.fixture +def client(captured: _CapturedService) -> Client: + service_client = Mock(spec=ServiceClient) + service_client.workflow_service = captured + service_client.config = Mock(identity="test-identity") + return Client(service_client=service_client, namespace="test-namespace") + + +async def test_operator_commands_send_reason_and_jitter( + client: Client, captured: _CapturedService +): + handle = client.get_activity_handle("act-1") + await handle.pause(reason="pause-reason") + await handle.unpause(reason="unpause-reason", jitter=timedelta(seconds=5)) + + assert captured.requests["pause"].reason == "pause-reason" + assert captured.requests["unpause"].reason == "unpause-reason" + assert captured.requests["unpause"].jitter.ToTimedelta() == timedelta(seconds=5) + + +async def test_omitted_jitter_is_left_off_the_wire( + client: Client, captured: _CapturedService +): + # An unset jitter must not be sent as an explicit zero duration, or the server would + # apply "no jitter" instead of its own default. + handle = client.get_activity_handle("act-1") + await handle.unpause() + + assert not captured.requests["unpause"].HasField("jitter") + + +async def test_restore_original_options_is_exclusive( + client: Client, captured: _CapturedService +): + handle = client.get_activity_handle("act-1") + await handle.restore_original_options() + + update = captured.requests["update"] + assert update.restore_original + assert list(update.update_mask.paths) == [] + + +async def test_value_set_of_zero_sends_an_explicit_zero( + client: Client, captured: _CapturedService +): + handle = client.get_activity_handle("act-1") + await handle.update_options( + [ActivityOptionsKeys.heartbeat_timeout.value_set(timedelta(0))] + ) + + update = captured.requests["update"] + assert update.update_mask.paths == ["heartbeat_timeout"] + # Present and zero, which is distinct from absent: the caller asked for zero. + assert update.activity_options.HasField("heartbeat_timeout") + assert update.activity_options.heartbeat_timeout.ToTimedelta() == timedelta(0) + + +async def test_value_unset_names_the_path_but_leaves_the_field_absent( + client: Client, captured: _CapturedService +): + handle = client.get_activity_handle("act-1") + await handle.update_options([ActivityOptionsKeys.heartbeat_timeout.value_unset()]) + + update = captured.requests["update"] + assert update.update_mask.paths == ["heartbeat_timeout"] + assert not update.activity_options.HasField("heartbeat_timeout") + + +async def test_a_repeated_key_resolves_to_its_last_update( + client: Client, captured: _CapturedService +): + handle = client.get_activity_handle("act-1") + await handle.update_options( + [ + ActivityOptionsKeys.heartbeat_timeout.value_set(timedelta(seconds=5)), + ActivityOptionsKeys.heartbeat_timeout.value_unset(), + ] + ) + + update = captured.requests["update"] + # The later unset wins, and the path is named once. + assert update.update_mask.paths == ["heartbeat_timeout"] + assert not update.activity_options.HasField("heartbeat_timeout") + + +async def test_update_options_requires_at_least_one_update(client: Client): + handle = client.get_activity_handle("act-1") + with pytest.raises(ValueError) as err: + await handle.update_options([]) + assert "at least one update" in str(err.value) + + +async def test_restore_original_cannot_be_combined_with_updates(client: Client): + from temporalio.client import UpdateActivityOptionsInput + + with pytest.raises(ValueError) as err: + await client._impl.update_activity_options( + UpdateActivityOptionsInput( + activity_id="act-1", + activity_run_id=None, + updates=[ + ActivityOptionsKeys.heartbeat_timeout.value_set( + timedelta(seconds=25) + ) + ], + restore_original=True, + rpc_metadata={}, + rpc_timeout=None, + ) + ) + assert "cannot be combined" in str(err.value) + + +async def test_update_options_masks_only_changed_options( + client: Client, captured: _CapturedService +): + handle = client.get_activity_handle("act-1") + await handle.update_options( + [ + ActivityOptionsKeys.task_queue.value_set("tq"), + ActivityOptionsKeys.start_to_close_timeout.value_set(timedelta(seconds=90)), + ] + ) + + update = captured.requests["update"] + assert not update.restore_original + assert sorted(update.update_mask.paths) == [ + "start_to_close_timeout", + "task_queue.name", + ] + assert update.activity_options.task_queue.name == "tq" + assert update.activity_options.start_to_close_timeout.ToTimedelta() == timedelta( + seconds=90 + )