diff --git a/changelog/breaking-changes/solitary-semaphore-sloth.md b/changelog/breaking-changes/solitary-semaphore-sloth.md new file mode 100644 index 000000000..1c08671f3 --- /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; a second raises a `BAD_REQUEST` error, unless the first signal failed. Signals from other Nexus operation handlers 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..cf1b469ca --- /dev/null +++ b/changelog/fixed/twice-shy-carrier-pigeon.md @@ -0,0 +1 @@ +Experimental: signals and signal-with-start sent from a Temporal operation handler now reuse the Nexus operation request ID, so a retried Nexus task no longer delivers the same signal twice. diff --git a/temporalio/client/_impl.py b/temporalio/client/_impl.py index 7168caf14..72986ffa7 100644 --- a/temporalio/client/_impl.py +++ b/temporalio/client/_impl.py @@ -171,12 +171,15 @@ async def start_workflow( req, temporalio.api.workflowservice.v1.SignalWithStartWorkflowExecutionRequest, ): - resp = await self._client.workflow_service.signal_with_start_workflow_execution( - req, - retry=True, - metadata=input.rpc_metadata, - timeout=input.rpc_timeout, - ) + with temporalio.nexus._operation_context._signal_request_nexus_context( + req + ): + resp = await self._client.workflow_service.signal_with_start_workflow_execution( + req, + retry=True, + metadata=input.rpc_metadata, + timeout=input.rpc_timeout, + ) first_execution_run_id = resp.first_execution_run_id or ( resp.run_id if resp.started else None ) @@ -255,9 +258,6 @@ async def _build_signal_with_start_workflow_execution_request( await data_converter.encode(input.start_signal_args) ) await self._populate_start_workflow_execution_request(req, input) - temporalio.nexus._operation_context._apply_nexus_context_to_signal_with_start_workflow_request( - req - ) return req async def _build_update_with_start_start_workflow_execution_request( @@ -499,12 +499,10 @@ async def signal_workflow(self, input: SignalWorkflowInput) -> None: req.input.payloads.extend(await data_converter.encode(input.args)) if input.headers is not None: # type:ignore[reportUnnecessaryComparison] await self._apply_headers(input.headers, req.header.fields) - temporalio.nexus._operation_context._apply_nexus_context_to_signal_workflow_request( - req - ) - resp = await self._client.workflow_service.signal_workflow_execution( - req, retry=True, metadata=input.rpc_metadata, timeout=input.rpc_timeout - ) + with temporalio.nexus._operation_context._signal_request_nexus_context(req): + resp = await self._client.workflow_service.signal_workflow_execution( + req, retry=True, metadata=input.rpc_metadata, timeout=input.rpc_timeout + ) temporalio.nexus._operation_context._apply_signal_workflow_response_to_nexus_context( resp ) diff --git a/temporalio/nexus/_operation_context.py b/temporalio/nexus/_operation_context.py index b4d590bbe..92c74ac30 100644 --- a/temporalio/nexus/_operation_context.py +++ b/temporalio/nexus/_operation_context.py @@ -10,7 +10,7 @@ MutableMapping, Sequence, ) -from contextlib import contextmanager +from contextlib import contextmanager, nullcontext from contextvars import ContextVar from dataclasses import dataclass from datetime import timedelta @@ -75,6 +75,45 @@ ) +class _SignalSlot: + """Allows one in-flight or successful signal per Temporal operation handler invocation.""" + + def __init__(self) -> None: + self._taken = False + + @staticmethod + @contextmanager + def scope() -> Generator[None]: + token = _temporal_operation_signal_slot.set(_SignalSlot()) + try: + yield + finally: + _temporal_operation_signal_slot.reset(token) + + @contextmanager + def reserve(self) -> Generator[None]: + # Every signal from the handler 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 nexusrpc.HandlerError( + "Only one signal can be sent per Temporal operation handler invocation", + type=nexusrpc.HandlerErrorType.BAD_REQUEST, + ) + self._taken = True + try: + yield + except BaseException: + self._taken = False + raise + + +# Set for the duration of a Temporal operation handler's start so its signals reuse the +# Nexus request ID and are not delivered twice when the Nexus task is retried. +_temporal_operation_signal_slot: ContextVar[_SignalSlot | None] = ContextVar( + "temporal-operation-signal-slot", default=None +) + + @dataclass(frozen=True) class Info: """Information about the running Nexus operation. @@ -824,13 +863,36 @@ def _apply_query_workflow_response_to_nexus_context( # pyright: ignore[reportUn nexus_ctx._add_response_link(resp.link) -def _apply_nexus_context_to_signal_workflow_request( # pyright: ignore[reportUnusedFunction] - req: temporalio.api.workflowservice.v1.SignalWorkflowExecutionRequest, -) -> None: - """Apply the current Nexus operation context to a workflow signal request.""" +@contextmanager +def _signal_request_nexus_context( # pyright: ignore[reportUnusedFunction] + req: ( + temporalio.api.workflowservice.v1.SignalWorkflowExecutionRequest + | temporalio.api.workflowservice.v1.SignalWithStartWorkflowExecutionRequest + ), +) -> Generator[None]: + """Apply the current Nexus operation context to a signal or signal-with-start RPC. + + Inside a Temporal operation handler, the request reuses the Nexus request ID and takes + the invocation's only signal, which is released if the RPC fails. + """ nexus_ctx = _try_start_operation_context() - if nexus_ctx is not None: + is_backing_start = ( + isinstance( + req, + temporalio.api.workflowservice.v1.SignalWithStartWorkflowExecutionRequest, + ) + and _in_nexus_backing_start_context() + ) + + slot = None + if nexus_ctx is not None and not is_backing_start: req.links.extend(nexus_ctx._get_request_links()) + slot = _temporal_operation_signal_slot.get() + if slot is not None: + req.request_id = nexus_ctx.nexus_context.request_id + + with slot.reserve() if slot is not None else nullcontext(): + yield def _apply_signal_workflow_response_to_nexus_context( # pyright: ignore[reportUnusedFunction] @@ -842,15 +904,6 @@ def _apply_signal_workflow_response_to_nexus_context( # pyright: ignore[reportU nexus_ctx._add_response_link(resp.link) -def _apply_nexus_context_to_signal_with_start_workflow_request( # pyright: ignore[reportUnusedFunction] - req: temporalio.api.workflowservice.v1.SignalWithStartWorkflowExecutionRequest, -) -> None: - """Apply the current Nexus operation context to a signal-with-start request.""" - 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()) - - def _apply_nexus_context_to_start_workflow_request( # pyright: ignore[reportUnusedFunction] req: temporalio.api.workflowservice.v1.StartWorkflowExecutionRequest, ) -> None: diff --git a/temporalio/nexus/_operation_handlers.py b/temporalio/nexus/_operation_handlers.py index 36c44e96c..b51507f71 100644 --- a/temporalio/nexus/_operation_handlers.py +++ b/temporalio/nexus/_operation_handlers.py @@ -23,6 +23,7 @@ from temporalio.nexus._operation_context import ( TemporalCancelOperationContext, TemporalStartOperationContext, + _SignalSlot, _temporal_cancel_operation_context, ) from temporalio.nexus._temporal_client import ( @@ -171,6 +172,10 @@ class TemporalOperationHandler(OperationHandler[InputT, OutputT], ABC): """Operation handler for Nexus operations that interact with Temporal. Implementations override the start_operation method. + Signals sent during the start reuse the Nexus request ID, so retried Nexus tasks don't + deliver them twice. Only one signal (including signal-with-start) may be sent per start; + another raises a ``BAD_REQUEST`` :py:class:`nexusrpc.HandlerError` unless the first failed. + .. warning:: This API is experimental and unstable. """ @@ -195,7 +200,8 @@ async def start( """ nexus_client = _TemporalNexusClient() start_ctx = TemporalStartOperationContext._from_start_operation_context(ctx) - result = await self.start_operation(start_ctx, nexus_client, input) + with _SignalSlot.scope(): + result = await self.start_operation(start_ctx, nexus_client, input) return result._to_nexus_result() async def cancel(self, ctx: CancelOperationContext, token: str) -> None: diff --git a/tests/nexus/test_temporal_operation.py b/tests/nexus/test_temporal_operation.py index 1fa7e0c3e..030db6f6c 100644 --- a/tests/nexus/test_temporal_operation.py +++ b/tests/nexus/test_temporal_operation.py @@ -10,15 +10,22 @@ from nexusrpc.handler import ( CancelOperationContext, OperationTaskCancellation, + StartOperationContext, operation_handler, service_handler, + sync_operation, ) 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.api.workflowservice.v1 import ( + SignalWorkflowExecutionRequest, + SignalWorkflowExecutionResponse, +) from temporalio.client import ( ActivityExecutionStatus, Client, @@ -126,6 +133,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 +150,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 +430,85 @@ 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) + + @sync_operation + async def unlimited_signals(self, _ctx: StartOperationContext, input: str) -> None: + handle = nexus.client().get_workflow_handle(input) + for _ in range(3): + await handle.signal(SignalCountWorkflow.my_signal) + + @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 +1311,213 @@ 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 Temporal 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_sync_operation_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()) + ) + + await handle.signal(SignalCountWorkflow.finish) + assert await handle.result() == 3 + + +async def test_temporal_operation_failed_signal_allows_retry( + client: Client, env: WorkflowEnvironment, monkeypatch: pytest.MonkeyPatch +): + 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) + + workflow_service = env.client.workflow_service + signal_workflow_execution = workflow_service.signal_workflow_execution + lost = False + + async def lose_first_signal_response( + req: SignalWorkflowExecutionRequest, *args: Any, **kwargs: Any + ) -> SignalWorkflowExecutionResponse: + nonlocal lost + resp = await signal_workflow_execution(req, *args, **kwargs) + if not lost: + lost = True + raise temporalio.service.RPCError( + "lost response", temporalio.service.RPCStatusCode.DEADLINE_EXCEEDED, b"" + ) + return resp + + monkeypatch.setattr( + workflow_service, "signal_workflow_execution", lose_first_signal_response + ) + async with Worker( + env.client, + 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