diff --git a/changelog/breaking-changes/solitary-semaphore-sloth.md b/changelog/breaking-changes/solitary-semaphore-sloth.md new file mode 100644 index 000000000..fcf7ed76e --- /dev/null +++ b/changelog/breaking-changes/solitary-semaphore-sloth.md @@ -0,0 +1 @@ +Experimental: a Temporal operation handler can now send only one signal (including signal-with-start) per invocation through `TemporalNexusClient.client`; a second raises a `BAD_REQUEST` error, unless the first signal failed. Signals sent through `temporalio.nexus.client()` are not limited. diff --git a/changelog/fixed/twice-shy-carrier-pigeon.md b/changelog/fixed/twice-shy-carrier-pigeon.md new file mode 100644 index 000000000..ee68fddba --- /dev/null +++ b/changelog/fixed/twice-shy-carrier-pigeon.md @@ -0,0 +1 @@ +Experimental: signals and signal-with-start sent through `TemporalNexusClient.client` in a Temporal operation handler now reuse the Nexus operation request ID, so a retried Nexus task no longer delivers the same signal twice. Signals sent through `temporalio.nexus.client()` are unchanged. diff --git a/temporalio/nexus/_operation_context.py b/temporalio/nexus/_operation_context.py index a91b72b94..cef6c3d3b 100644 --- a/temporalio/nexus/_operation_context.py +++ b/temporalio/nexus/_operation_context.py @@ -73,6 +73,12 @@ "temporal-nexus-backing-start-context" ) +# Set while a signal is sent through TemporalNexusClient.client so the request reuses +# the Nexus request ID and the server deduplicates it when the Nexus task is retried. +_temporal_operation_signal_context: ContextVar[bool] = ContextVar( + "temporal-operation-signal-context" +) + @dataclass(frozen=True) class Info: @@ -186,6 +192,19 @@ def _in_nexus_backing_start_context() -> bool: # type:ignore[reportUnusedClass] return _temporal_nexus_backing_start_context.get(False) +@contextmanager +def _operation_signal_context() -> Generator[None]: # pyright: ignore[reportUnusedFunction] + token = _temporal_operation_signal_context.set(True) + try: + yield + finally: + _temporal_operation_signal_context.reset(token) + + +def _in_operation_signal_context() -> bool: + return _temporal_operation_signal_context.get(False) + + _OperationCtxT = TypeVar("_OperationCtxT", bound=OperationContext) @@ -796,6 +815,8 @@ def _apply_nexus_context_to_signal_workflow_request( # pyright: ignore[reportUn nexus_ctx = _try_start_operation_context() if nexus_ctx is not None: req.links.extend(nexus_ctx._get_request_links()) + if _in_operation_signal_context(): + req.request_id = nexus_ctx.nexus_context.request_id def _apply_signal_workflow_response_to_nexus_context( # pyright: ignore[reportUnusedFunction] @@ -814,6 +835,8 @@ def _apply_nexus_context_to_signal_with_start_workflow_request( # pyright: igno nexus_ctx = _try_start_operation_context() if nexus_ctx is not None and not _in_nexus_backing_start_context(): req.links.extend(nexus_ctx._get_request_links()) + if _in_operation_signal_context(): + req.request_id = nexus_ctx.nexus_context.request_id def _apply_nexus_context_to_start_workflow_request( # pyright: ignore[reportUnusedFunction] diff --git a/temporalio/nexus/_operation_handlers.py b/temporalio/nexus/_operation_handlers.py index 36c44e96c..1aeb1f788 100644 --- a/temporalio/nexus/_operation_handlers.py +++ b/temporalio/nexus/_operation_handlers.py @@ -28,6 +28,8 @@ from temporalio.nexus._temporal_client import ( TemporalNexusClient, TemporalOperationResult, + _SignalSlot, + _temporal_operation_signal_slot, _TemporalNexusClient, ) from temporalio.nexus._token import OperationToken, OperationTokenType, WorkflowHandle @@ -195,7 +197,11 @@ async def start( """ nexus_client = _TemporalNexusClient() start_ctx = TemporalStartOperationContext._from_start_operation_context(ctx) - result = await self.start_operation(start_ctx, nexus_client, input) + token = _temporal_operation_signal_slot.set(_SignalSlot()) + try: + result = await self.start_operation(start_ctx, nexus_client, input) + finally: + _temporal_operation_signal_slot.reset(token) return result._to_nexus_result() async def cancel(self, ctx: CancelOperationContext, token: str) -> None: diff --git a/temporalio/nexus/_single_signal_interceptor.py b/temporalio/nexus/_single_signal_interceptor.py new file mode 100644 index 000000000..26f54dde5 --- /dev/null +++ b/temporalio/nexus/_single_signal_interceptor.py @@ -0,0 +1,60 @@ +from __future__ import annotations + +import weakref +from collections.abc import Iterator +from contextlib import contextmanager +from typing import Any + +import temporalio.client +from temporalio.nexus._operation_context import _operation_signal_context +from temporalio.nexus._temporal_client import _temporal_operation_signal_slot + +_single_signal_clients: weakref.WeakKeyDictionary[ + temporalio.client.Client, temporalio.client.Client +] = weakref.WeakKeyDictionary() + + +def _single_signal_client( # pyright: ignore[reportUnusedFunction] + base: temporalio.client.Client, +) -> temporalio.client.Client: + client = _single_signal_clients.get(base) + if client is None: + config = base.config() + # Outermost, so a failure from any other interceptor frees the slot. + config["interceptors"] = [_SingleSignalInterceptor(), *config["interceptors"]] + client = temporalio.client.Client(**config) + _single_signal_clients[base] = client + return client + + +class _SingleSignalInterceptor(temporalio.client.Interceptor): + def intercept_client( + self, next: temporalio.client.OutboundInterceptor + ) -> temporalio.client.OutboundInterceptor: + return _SingleSignalOutboundInterceptor(next) + + +@contextmanager +def _reserve_signal() -> Iterator[None]: + slot = _temporal_operation_signal_slot.get() + if slot is None: + yield + return + with slot.reserve(), _operation_signal_context(): + yield + + +class _SingleSignalOutboundInterceptor(temporalio.client.OutboundInterceptor): + async def signal_workflow( + self, input: temporalio.client.SignalWorkflowInput + ) -> None: + with _reserve_signal(): + await self.next.signal_workflow(input) + + async def start_workflow( + self, input: temporalio.client.StartWorkflowInput + ) -> temporalio.client.WorkflowHandle[Any, Any]: + if input.start_signal is None: + return await self.next.start_workflow(input) + with _reserve_signal(): + return await self.next.start_workflow(input) diff --git a/temporalio/nexus/_temporal_client.py b/temporalio/nexus/_temporal_client.py index 4fd34d939..f0934ff45 100644 --- a/temporalio/nexus/_temporal_client.py +++ b/temporalio/nexus/_temporal_client.py @@ -3,6 +3,7 @@ from abc import ABC, abstractmethod from collections.abc import Awaitable, Callable, Iterator, Mapping, Sequence from contextlib import contextmanager +from contextvars import ContextVar from dataclasses import dataclass from datetime import timedelta from typing import ( @@ -96,6 +97,35 @@ def _to_nexus_result( ) +class _SignalSlot: + """Allows one in-flight or successful signal per operation handler invocation.""" + + def __init__(self) -> None: + self._taken = False + + @contextmanager + def reserve(self) -> Iterator[None]: + # Every signal through TemporalNexusClient.client carries the operation's + # request ID, so the server would deduplicate a second one against the + # first and report it delivered. + if self._taken: + raise HandlerError( + "Only one signal can be sent per operation handler invocation through TemporalNexusClient.client", + type=HandlerErrorType.BAD_REQUEST, + ) + self._taken = True + try: + yield + except BaseException: + self._taken = False + raise + + +_temporal_operation_signal_slot: ContextVar[_SignalSlot | None] = ContextVar( + "temporal-operation-signal-slot", default=None +) + + class TemporalNexusClient(ABC): """Nexus-aware wrapper around a Temporal Client. @@ -108,6 +138,14 @@ class TemporalNexusClient(ABC): def client(self) -> temporalio.client.Client: """The underlying Temporal Client + Signals sent through this client from the operation handler reuse the Nexus + operation's request ID so a retried Nexus task does not deliver them twice. + Because they share that ID, only one signal may be sent per operation handler + invocation; a second raises a ``BAD_REQUEST`` :py:class:`nexusrpc.HandlerError` + instead of being silently dropped by the server. A signal that fails does not + count toward the limit. Signals sent through :py:func:`temporalio.nexus.client` + are not limited, but are not deduplicated across Nexus task retries. + .. warning:: This API is experimental and unstable. """ @@ -595,7 +633,12 @@ def __init__(self) -> None: @property def client(self) -> temporalio.client.Client: """Return the Temporal client for the active Nexus operation.""" - return self._temporal_context.client + # temporalio.client imports this module, so it can't be imported at load time. + from temporalio.nexus._single_signal_interceptor import ( + _single_signal_client, + ) + + return _single_signal_client(self._temporal_context.client) @contextmanager def _reserve_async_start(self) -> Iterator[None]: diff --git a/tests/nexus/test_temporal_operation.py b/tests/nexus/test_temporal_operation.py index 1fa7e0c3e..e72ad4cd8 100644 --- a/tests/nexus/test_temporal_operation.py +++ b/tests/nexus/test_temporal_operation.py @@ -16,13 +16,17 @@ from typing_extensions import override import temporalio.exceptions +import temporalio.service from temporalio import activity, nexus, workflow from temporalio.api.activity.v1 import ActivityExecutionInfo from temporalio.api.common.v1 import Link from temporalio.client import ( ActivityExecutionStatus, Client, + Interceptor, NexusOperationFailureError, + OutboundInterceptor, + SignalWorkflowInput, WorkflowExecutionStatus, WorkflowFailureError, WorkflowUpdateStage, @@ -34,6 +38,7 @@ ) from temporalio.nexus._token import OperationToken, OperationTokenType from temporalio.testing import WorkflowEnvironment +from temporalio.testing._workflow import _client_with_interceptors from temporalio.worker import Worker from tests.helpers import EventType, assert_event_subsequence, assert_eventually from tests.helpers.nexus import ( @@ -126,6 +131,11 @@ class TestService: custom_cancel_activity: Operation[str, None] double_start_activity: Operation[Input, None] mixed_start: Operation[Input, None] + double_signal: Operation[str, None] + concurrent_signal: Operation[str, None] + unlimited_signals: Operation[str, None] + signal_after_failed_signal: Operation[str, None] + signal_handler_retry: Operation[str, None] @service_handler(service=TestService) @@ -138,6 +148,8 @@ def __init__(self) -> None: self.started_custom_cancel_activity = asyncio.Event() self.custom_cancel_activity_called = asyncio.Event() self.bad_update_stage_error: ValueError | None = None + self.signal_handler_retry_request_ids: list[str] = [] + self.concurrent_signal_results: list[BaseException | None] = [] @nexus.temporal_operation async def echo( @@ -416,6 +428,94 @@ async def mixed_start( ) return nexus.TemporalOperationResult.sync(None) + @nexus.temporal_operation + async def double_signal( + self, + _ctx: nexus.TemporalStartOperationContext, + client: nexus.TemporalNexusClient, + input: str, + ) -> nexus.TemporalOperationResult[None]: + await client.client.start_workflow( + SignalCountWorkflow.run, + id=input, + task_queue=nexus.info().task_queue, + start_signal="my_signal", + ) + # Rejected even though it targets a different workflow. + await client.client.get_workflow_handle(f"{input}-other").signal( + SignalCountWorkflow.my_signal + ) + raise RuntimeError("Expected second signal to be rejected") + + @nexus.temporal_operation + async def concurrent_signal( + self, + _ctx: nexus.TemporalStartOperationContext, + client: nexus.TemporalNexusClient, + input: str, + ) -> nexus.TemporalOperationResult[None]: + handle = client.client.get_workflow_handle(input) + self.concurrent_signal_results = list( + await asyncio.gather( + handle.signal(SignalCountWorkflow.my_signal), + handle.signal(SignalCountWorkflow.my_signal), + return_exceptions=True, + ) + ) + return nexus.TemporalOperationResult.sync(None) + + @nexus.temporal_operation + async def unlimited_signals( + self, + _ctx: nexus.TemporalStartOperationContext, + client: nexus.TemporalNexusClient, + input: str, + ) -> nexus.TemporalOperationResult[None]: + await client.client.get_workflow_handle(input).signal( + SignalCountWorkflow.my_signal + ) + handle = nexus.client().get_workflow_handle(input) + await handle.signal(SignalCountWorkflow.my_signal) + await handle.signal(SignalCountWorkflow.my_signal) + return nexus.TemporalOperationResult.sync(None) + + @nexus.temporal_operation + async def signal_after_failed_signal( + self, + _ctx: nexus.TemporalStartOperationContext, + client: nexus.TemporalNexusClient, + input: str, + ) -> nexus.TemporalOperationResult[None]: + handle = client.client.get_workflow_handle(input) + try: + await handle.signal(SignalCountWorkflow.my_signal) + except temporalio.service.RPCError: + await handle.signal(SignalCountWorkflow.my_signal) + return nexus.TemporalOperationResult.sync(None) + raise RuntimeError("Expected first signal to fail") + + @nexus.temporal_operation + async def signal_handler_retry( + self, + ctx: nexus.TemporalStartOperationContext, + client: nexus.TemporalNexusClient, + input: str, + ) -> nexus.TemporalOperationResult[None]: + await client.client.start_workflow( + SignalCountWorkflow.run, + id=input, + task_queue=nexus.info().task_queue, + start_signal="my_signal", + ) + self.signal_handler_retry_request_ids.append(ctx.request_id) + if len(self.signal_handler_retry_request_ids) == 1: + raise nexusrpc.HandlerError( + "force handler redelivery after signaling", + type=HandlerErrorType.INTERNAL, + retryable_override=True, + ) + return nexus.TemporalOperationResult.sync(None) + @operation_handler def custom_cancel_activity(self) -> nexus.TemporalOperationHandler[str, None]: started = self.started_custom_cancel_activity @@ -1218,6 +1318,218 @@ async def test_temporal_operation_mixed_start_raises_handler_err( ) +@workflow.defn +class SignalCountWorkflow: + def __init__(self) -> None: + self.count = 0 + self.done = False + + @workflow.run + async def run(self) -> int: + await workflow.wait_condition(lambda: self.done) + return self.count + + @workflow.signal + def my_signal(self) -> None: + self.count += 1 + + @workflow.signal + def finish(self) -> None: + self.done = True + + +async def test_temporal_operation_double_signal_raises_handler_err( + client: Client, env: WorkflowEnvironment +): + if env.supports_time_skipping_v1: + pytest.skip( + "Standalone Nexus Operation tests don't work with time-skipping server" + ) + + task_queue = str(uuid.uuid4()) + endpoint_name = make_nexus_endpoint_name(task_queue) + await env.create_nexus_endpoint(endpoint_name, task_queue) + target_id = f"double-signal-{uuid.uuid4()}" + async with Worker( + env.client, + task_queue=task_queue, + nexus_service_handlers=[TestServiceHandler()], + workflows=[SignalCountWorkflow], + ): + nexus_client = client.create_nexus_client(TestService, endpoint_name) + with pytest.raises(NexusOperationFailureError) as err: + await nexus_client.execute_operation( + TestService.double_signal, target_id, id=str(uuid.uuid4()) + ) + + assert isinstance(err.value.cause, nexusrpc.HandlerError) + assert err.value.cause.type == HandlerErrorType.BAD_REQUEST + assert ( + "Only one signal can be sent per operation handler invocation" + in err.value.cause.message + ) + + +async def test_temporal_operation_concurrent_signal_raises_handler_err( + client: Client, env: WorkflowEnvironment +): + if env.supports_time_skipping_v1: + pytest.skip( + "Standalone Nexus Operation tests don't work with time-skipping server" + ) + + task_queue = str(uuid.uuid4()) + endpoint_name = make_nexus_endpoint_name(task_queue) + await env.create_nexus_endpoint(endpoint_name, task_queue) + handler = TestServiceHandler() + async with Worker( + env.client, + task_queue=task_queue, + nexus_service_handlers=[handler], + workflows=[SignalCountWorkflow], + ): + handle = await client.start_workflow( + SignalCountWorkflow.run, + id=f"concurrent-signal-{uuid.uuid4()}", + task_queue=task_queue, + ) + nexus_client = client.create_nexus_client(TestService, endpoint_name) + await nexus_client.execute_operation( + TestService.concurrent_signal, handle.id, id=str(uuid.uuid4()) + ) + + [err] = [r for r in handler.concurrent_signal_results if r is not None] + assert isinstance(err, nexusrpc.HandlerError) + assert err.type == HandlerErrorType.BAD_REQUEST + + await handle.signal(SignalCountWorkflow.finish) + assert await handle.result() == 1 + + +async def test_temporal_operation_nexus_client_signals_not_limited( + client: Client, env: WorkflowEnvironment +): + if env.supports_time_skipping_v1: + pytest.skip( + "Standalone Nexus Operation tests don't work with time-skipping server" + ) + + task_queue = str(uuid.uuid4()) + endpoint_name = make_nexus_endpoint_name(task_queue) + await env.create_nexus_endpoint(endpoint_name, task_queue) + async with Worker( + env.client, + task_queue=task_queue, + nexus_service_handlers=[TestServiceHandler()], + workflows=[SignalCountWorkflow], + ): + handle = await client.start_workflow( + SignalCountWorkflow.run, + id=f"unlimited-signals-{uuid.uuid4()}", + task_queue=task_queue, + ) + nexus_client = client.create_nexus_client(TestService, endpoint_name) + await nexus_client.execute_operation( + TestService.unlimited_signals, handle.id, id=str(uuid.uuid4()) + ) + + # A shared request ID would have deduplicated all three into one. + await handle.signal(SignalCountWorkflow.finish) + assert await handle.result() == 3 + + +class _LoseFirstSignalResponseInterceptor(Interceptor): + """Applies the first signal, then raises as if its response were lost.""" + + def __init__(self) -> None: + super().__init__() + self.lost = False + + def intercept_client(self, next: OutboundInterceptor) -> OutboundInterceptor: + interceptor = self + + class _Outbound(OutboundInterceptor): + async def signal_workflow(self, input: SignalWorkflowInput) -> None: + await self.next.signal_workflow(input) + if not interceptor.lost: + interceptor.lost = True + raise temporalio.service.RPCError( + "lost response", + temporalio.service.RPCStatusCode.DEADLINE_EXCEEDED, + b"", + ) + + return _Outbound(next) + + +async def test_temporal_operation_failed_signal_allows_retry( + client: Client, env: WorkflowEnvironment +): + if env.supports_time_skipping_v1: + pytest.skip( + "Standalone Nexus Operation tests don't work with time-skipping server" + ) + + task_queue = str(uuid.uuid4()) + endpoint_name = make_nexus_endpoint_name(task_queue) + await env.create_nexus_endpoint(endpoint_name, task_queue) + async with Worker( + _client_with_interceptors(env.client, _LoseFirstSignalResponseInterceptor()), + task_queue=task_queue, + nexus_service_handlers=[TestServiceHandler()], + workflows=[SignalCountWorkflow], + ): + handle = await client.start_workflow( + SignalCountWorkflow.run, + id=f"failed-signal-{uuid.uuid4()}", + task_queue=task_queue, + ) + nexus_client = client.create_nexus_client(TestService, endpoint_name) + # The first signal was applied but reported as failed. The failure frees + # the limit, and the resend reuses the request ID so it is deduplicated. + await nexus_client.execute_operation( + TestService.signal_after_failed_signal, + handle.id, + id=str(uuid.uuid4()), + ) + + await handle.signal(SignalCountWorkflow.finish) + assert await handle.result() == 1 + + +async def test_temporal_operation_signal_deduped_across_handler_retry( + client: Client, env: WorkflowEnvironment +): + if env.supports_time_skipping_v1: + pytest.skip( + "Standalone Nexus Operation tests don't work with time-skipping server" + ) + + task_queue = str(uuid.uuid4()) + endpoint_name = make_nexus_endpoint_name(task_queue) + await env.create_nexus_endpoint(endpoint_name, task_queue) + target_id = f"signal-retry-{uuid.uuid4()}" + handler = TestServiceHandler() + async with Worker( + env.client, + task_queue=task_queue, + nexus_service_handlers=[handler], + workflows=[SignalCountWorkflow], + ): + nexus_client = client.create_nexus_client(TestService, endpoint_name) + await nexus_client.execute_operation( + TestService.signal_handler_retry, target_id, id=str(uuid.uuid4()) + ) + + request_ids = handler.signal_handler_retry_request_ids + assert len(request_ids) == 2 + assert request_ids[0] and request_ids[0] == request_ids[1] + + handle = client.get_workflow_handle(target_id) + await handle.signal(SignalCountWorkflow.finish) + assert await handle.result() == 1 + + @workflow.defn class SyncResultCaller: @workflow.run