From 362bd1ecd71845a703a8c7117598aa8855ba597e Mon Sep 17 00:00:00 2001 From: Alice Lin Date: Wed, 28 Oct 2026 18:30:11 -0700 Subject: [PATCH 1/5] Dedup retried Nexus signals --- .../solitary-semaphore-sloth.md | 1 + changelog/fixed/twice-shy-carrier-pigeon.md | 1 + temporalio/nexus/_operation_context.py | 23 ++ temporalio/nexus/_operation_handlers.py | 8 +- .../nexus/_single_signal_interceptor.py | 60 ++++ temporalio/nexus/_temporal_client.py | 45 ++- tests/nexus/test_temporal_operation.py | 312 ++++++++++++++++++ 7 files changed, 448 insertions(+), 2 deletions(-) create mode 100644 changelog/breaking-changes/solitary-semaphore-sloth.md create mode 100644 changelog/fixed/twice-shy-carrier-pigeon.md create mode 100644 temporalio/nexus/_single_signal_interceptor.py diff --git a/changelog/breaking-changes/solitary-semaphore-sloth.md b/changelog/breaking-changes/solitary-semaphore-sloth.md new file mode 100644 index 0000000000..fcf7ed76e5 --- /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 0000000000..ee68fddba4 --- /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 b4d590bbe7..14cf325c2f 100644 --- a/temporalio/nexus/_operation_context.py +++ b/temporalio/nexus/_operation_context.py @@ -74,6 +74,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: @@ -206,6 +212,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) @@ -831,6 +850,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] @@ -849,6 +870,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 36c44e96c2..1aeb1f788d 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 0000000000..26f54dde52 --- /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 4fd34d9394..f0934ff450 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 1fa7e0c3ee..e72ad4cd88 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 From bbe73302a8fab135f725a8245212755058971f9f Mon Sep 17 00:00:00 2001 From: Alice Lin Date: Thu, 5 Nov 2026 12:04:53 -0700 Subject: [PATCH 2/5] Remove TemporalNexusClient.client and add signal context manager --- .../solitary-semaphore-sloth.md | 3 +- changelog/fixed/twice-shy-carrier-pigeon.md | 2 +- temporalio/client/_impl.py | 28 +++-- temporalio/nexus/_operation_context.py | 93 +++++++++------ temporalio/nexus/_operation_handlers.py | 8 +- .../nexus/_single_signal_interceptor.py | 60 ---------- temporalio/nexus/_temporal_client.py | 63 +---------- tests/nexus/test_temporal_operation.py | 106 ++++++++---------- 8 files changed, 131 insertions(+), 232 deletions(-) delete mode 100644 temporalio/nexus/_single_signal_interceptor.py diff --git a/changelog/breaking-changes/solitary-semaphore-sloth.md b/changelog/breaking-changes/solitary-semaphore-sloth.md index fcf7ed76e5..947d59fd64 100644 --- a/changelog/breaking-changes/solitary-semaphore-sloth.md +++ b/changelog/breaking-changes/solitary-semaphore-sloth.md @@ -1 +1,2 @@ -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. +Experimental: removed `TemporalNexusClient.client`; use `temporalio.nexus.client()` instead. +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 index ee68fddba4..cf1b469cab 100644 --- a/changelog/fixed/twice-shy-carrier-pigeon.md +++ b/changelog/fixed/twice-shy-carrier-pigeon.md @@ -1 +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. +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 7168caf14a..72986ffa70 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 14cf325c2f..ab21d7f0bc 100644 --- a/temporalio/nexus/_operation_context.py +++ b/temporalio/nexus/_operation_context.py @@ -74,10 +74,34 @@ "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" + +class _SignalSlot: + """Allows one in-flight or successful signal per Temporal operation handler invocation.""" + + def __init__(self) -> None: + self._taken = False + + @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 ) @@ -212,19 +236,6 @@ 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) @@ -843,15 +854,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: - req.links.extend(nexus_ctx._get_request_links()) - if _in_operation_signal_context(): - req.request_id = nexus_ctx.nexus_context.request_id + if nexus_ctx is None or ( + isinstance( + req, + temporalio.api.workflowservice.v1.SignalWithStartWorkflowExecutionRequest, + ) + and _in_nexus_backing_start_context() + ): + yield + return + req.links.extend(nexus_ctx._get_request_links()) + slot = _temporal_operation_signal_slot.get() + if slot is None: + yield + return + req.request_id = nexus_ctx.nexus_context.request_id + with slot.reserve(): + yield def _apply_signal_workflow_response_to_nexus_context( # pyright: ignore[reportUnusedFunction] @@ -863,17 +895,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()) - 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] req: temporalio.api.workflowservice.v1.StartWorkflowExecutionRequest, ) -> None: diff --git a/temporalio/nexus/_operation_handlers.py b/temporalio/nexus/_operation_handlers.py index 1aeb1f788d..f5e62ffbc3 100644 --- a/temporalio/nexus/_operation_handlers.py +++ b/temporalio/nexus/_operation_handlers.py @@ -23,13 +23,13 @@ from temporalio.nexus._operation_context import ( TemporalCancelOperationContext, TemporalStartOperationContext, + _SignalSlot, _temporal_cancel_operation_context, + _temporal_operation_signal_slot, ) from temporalio.nexus._temporal_client import ( TemporalNexusClient, TemporalOperationResult, - _SignalSlot, - _temporal_operation_signal_slot, _TemporalNexusClient, ) from temporalio.nexus._token import OperationToken, OperationTokenType, WorkflowHandle @@ -173,6 +173,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. """ diff --git a/temporalio/nexus/_single_signal_interceptor.py b/temporalio/nexus/_single_signal_interceptor.py deleted file mode 100644 index 26f54dde52..0000000000 --- a/temporalio/nexus/_single_signal_interceptor.py +++ /dev/null @@ -1,60 +0,0 @@ -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 f0934ff450..5755c54bf4 100644 --- a/temporalio/nexus/_temporal_client.py +++ b/temporalio/nexus/_temporal_client.py @@ -3,7 +3,6 @@ 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 ( @@ -97,60 +96,16 @@ 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. + Use :py:func:`temporalio.nexus.client` for other workflow interactions, such as + queries and signals. + .. warning:: This API is experimental and unstable. """ - @property - @abstractmethod - 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. - """ - ... - # Overload for no-param workflow @overload async def start_workflow( @@ -630,21 +585,11 @@ def __init__(self) -> None: self._temporal_context = _TemporalStartOperationContext.get() self._started_async = False - @property - def client(self) -> temporalio.client.Client: - """Return the Temporal client for the active Nexus operation.""" - # 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]: if self._started_async: raise HandlerError( - "Only one async operation can be started per operation handler invocation. Use TemporalNexusClient.client for additional workflow interactions", + "Only one async operation can be started per operation handler invocation. Use temporalio.nexus.client() for additional workflow interactions", type=HandlerErrorType.BAD_REQUEST, ) diff --git a/tests/nexus/test_temporal_operation.py b/tests/nexus/test_temporal_operation.py index e72ad4cd88..d470b58e59 100644 --- a/tests/nexus/test_temporal_operation.py +++ b/tests/nexus/test_temporal_operation.py @@ -10,8 +10,10 @@ from nexusrpc.handler import ( CancelOperationContext, OperationTaskCancellation, + StartOperationContext, operation_handler, service_handler, + sync_operation, ) from typing_extensions import override @@ -20,13 +22,14 @@ 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, - Interceptor, NexusOperationFailureError, - OutboundInterceptor, - SignalWorkflowInput, WorkflowExecutionStatus, WorkflowFailureError, WorkflowUpdateStage, @@ -38,7 +41,6 @@ ) 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 ( @@ -334,10 +336,10 @@ async def bad_update_stage_op( async def query_op( self, _ctx: nexus.TemporalStartOperationContext, - client: nexus.TemporalNexusClient, + _client: nexus.TemporalNexusClient, input: str, ) -> nexus.TemporalOperationResult[bool]: - handle = client.client.get_workflow_handle(input) + handle = nexus.client().get_workflow_handle(input) result = await handle.query(BlockingWorkflow.query_done) return nexus.TemporalOperationResult.sync(result) @@ -432,18 +434,20 @@ async def mixed_start( async def double_signal( self, _ctx: nexus.TemporalStartOperationContext, - client: nexus.TemporalNexusClient, + _client: nexus.TemporalNexusClient, input: str, ) -> nexus.TemporalOperationResult[None]: - await client.client.start_workflow( + await nexus.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 + await ( + nexus.client() + .get_workflow_handle(f"{input}-other") + .signal(SignalCountWorkflow.my_signal) ) raise RuntimeError("Expected second signal to be rejected") @@ -451,10 +455,10 @@ async def double_signal( async def concurrent_signal( self, _ctx: nexus.TemporalStartOperationContext, - client: nexus.TemporalNexusClient, + _client: nexus.TemporalNexusClient, input: str, ) -> nexus.TemporalOperationResult[None]: - handle = client.client.get_workflow_handle(input) + handle = nexus.client().get_workflow_handle(input) self.concurrent_signal_results = list( await asyncio.gather( handle.signal(SignalCountWorkflow.my_signal), @@ -464,29 +468,20 @@ async def concurrent_signal( ) 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 - ) + @sync_operation + async def unlimited_signals(self, _ctx: StartOperationContext, input: str) -> None: handle = nexus.client().get_workflow_handle(input) - await handle.signal(SignalCountWorkflow.my_signal) - await handle.signal(SignalCountWorkflow.my_signal) - return nexus.TemporalOperationResult.sync(None) + 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, + _client: nexus.TemporalNexusClient, input: str, ) -> nexus.TemporalOperationResult[None]: - handle = client.client.get_workflow_handle(input) + handle = nexus.client().get_workflow_handle(input) try: await handle.signal(SignalCountWorkflow.my_signal) except temporalio.service.RPCError: @@ -498,10 +493,10 @@ async def signal_after_failed_signal( async def signal_handler_retry( self, ctx: nexus.TemporalStartOperationContext, - client: nexus.TemporalNexusClient, + _client: nexus.TemporalNexusClient, input: str, ) -> nexus.TemporalOperationResult[None]: - await client.client.start_workflow( + await nexus.client().start_workflow( SignalCountWorkflow.run, id=input, task_queue=nexus.info().task_queue, @@ -1365,7 +1360,7 @@ async def test_temporal_operation_double_signal_raises_handler_err( 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" + "Only one signal can be sent per Temporal operation handler invocation" in err.value.cause.message ) @@ -1406,7 +1401,7 @@ async def test_temporal_operation_concurrent_signal_raises_handler_err( assert await handle.result() == 1 -async def test_temporal_operation_nexus_client_signals_not_limited( +async def test_sync_operation_signals_not_limited( client: Client, env: WorkflowEnvironment ): if env.supports_time_skipping_v1: @@ -1433,37 +1428,12 @@ async def test_temporal_operation_nexus_client_signals_not_limited( 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 + client: Client, env: WorkflowEnvironment, monkeypatch: pytest.MonkeyPatch ): if env.supports_time_skipping_v1: pytest.skip( @@ -1473,8 +1443,28 @@ async def test_temporal_operation_failed_signal_allows_retry( 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( - _client_with_interceptors(env.client, _LoseFirstSignalResponseInterceptor()), + env.client, task_queue=task_queue, nexus_service_handlers=[TestServiceHandler()], workflows=[SignalCountWorkflow], From 45a78616eadb89951ca5654572d98b78d785d503 Mon Sep 17 00:00:00 2001 From: Alice Lin <55072906+atol@users.noreply.github.com> Date: Fri, 6 Nov 2026 11:03:08 -0700 Subject: [PATCH 3/5] Reduce yield points Co-authored-by: Alex Mazzeo --- temporalio/nexus/_operation_context.py | 22 +++++++++++----------- 1 file changed, 11 insertions(+), 11 deletions(-) diff --git a/temporalio/nexus/_operation_context.py b/temporalio/nexus/_operation_context.py index ab21d7f0bc..adf34ecaa0 100644 --- a/temporalio/nexus/_operation_context.py +++ b/temporalio/nexus/_operation_context.py @@ -867,22 +867,22 @@ def _signal_request_nexus_context( # pyright: ignore[reportUnusedFunction] the invocation's only signal, which is released if the RPC fails. """ nexus_ctx = _try_start_operation_context() - if nexus_ctx is None or ( + is_backing_start = ( isinstance( req, temporalio.api.workflowservice.v1.SignalWithStartWorkflowExecutionRequest, ) and _in_nexus_backing_start_context() - ): - yield - return - req.links.extend(nexus_ctx._get_request_links()) - slot = _temporal_operation_signal_slot.get() - if slot is None: - yield - return - req.request_id = nexus_ctx.nexus_context.request_id - with slot.reserve(): + ) + + 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 From 74e5d84c45359d890f5cd51547628d8a95fa72a5 Mon Sep 17 00:00:00 2001 From: Alice Lin Date: Fri, 6 Nov 2026 11:14:31 -0700 Subject: [PATCH 4/5] Encapsulate logic in context manager --- temporalio/nexus/_operation_context.py | 11 ++++++++++- temporalio/nexus/_operation_handlers.py | 6 +----- 2 files changed, 11 insertions(+), 6 deletions(-) diff --git a/temporalio/nexus/_operation_context.py b/temporalio/nexus/_operation_context.py index adf34ecaa0..92c74ac308 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 @@ -81,6 +81,15 @@ class _SignalSlot: 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 diff --git a/temporalio/nexus/_operation_handlers.py b/temporalio/nexus/_operation_handlers.py index f5e62ffbc3..b51507f71f 100644 --- a/temporalio/nexus/_operation_handlers.py +++ b/temporalio/nexus/_operation_handlers.py @@ -25,7 +25,6 @@ TemporalStartOperationContext, _SignalSlot, _temporal_cancel_operation_context, - _temporal_operation_signal_slot, ) from temporalio.nexus._temporal_client import ( TemporalNexusClient, @@ -201,11 +200,8 @@ async def start( """ nexus_client = _TemporalNexusClient() start_ctx = TemporalStartOperationContext._from_start_operation_context(ctx) - token = _temporal_operation_signal_slot.set(_SignalSlot()) - try: + with _SignalSlot.scope(): 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: From abcb22c5477854df1fe32bffd88b710cb7ad0db1 Mon Sep 17 00:00:00 2001 From: Alice Lin Date: Fri, 6 Nov 2026 11:52:01 -0700 Subject: [PATCH 5/5] Restore TemporalClient.client --- .../solitary-semaphore-sloth.md | 1 - temporalio/nexus/_temporal_client.py | 20 +++++++++++--- tests/nexus/test_temporal_operation.py | 26 +++++++++---------- 3 files changed, 28 insertions(+), 19 deletions(-) diff --git a/changelog/breaking-changes/solitary-semaphore-sloth.md b/changelog/breaking-changes/solitary-semaphore-sloth.md index 947d59fd64..1c08671f3b 100644 --- a/changelog/breaking-changes/solitary-semaphore-sloth.md +++ b/changelog/breaking-changes/solitary-semaphore-sloth.md @@ -1,2 +1 @@ -Experimental: removed `TemporalNexusClient.client`; use `temporalio.nexus.client()` instead. 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/temporalio/nexus/_temporal_client.py b/temporalio/nexus/_temporal_client.py index 5755c54bf4..4fd34d9394 100644 --- a/temporalio/nexus/_temporal_client.py +++ b/temporalio/nexus/_temporal_client.py @@ -99,13 +99,20 @@ def _to_nexus_result( class TemporalNexusClient(ABC): """Nexus-aware wrapper around a Temporal Client. - Use :py:func:`temporalio.nexus.client` for other workflow interactions, such as - queries and signals. - .. warning:: This API is experimental and unstable. """ + @property + @abstractmethod + def client(self) -> temporalio.client.Client: + """The underlying Temporal Client + + .. warning:: + This API is experimental and unstable. + """ + ... + # Overload for no-param workflow @overload async def start_workflow( @@ -585,11 +592,16 @@ def __init__(self) -> None: self._temporal_context = _TemporalStartOperationContext.get() self._started_async = False + @property + def client(self) -> temporalio.client.Client: + """Return the Temporal client for the active Nexus operation.""" + return self._temporal_context.client + @contextmanager def _reserve_async_start(self) -> Iterator[None]: if self._started_async: raise HandlerError( - "Only one async operation can be started per operation handler invocation. Use temporalio.nexus.client() for additional workflow interactions", + "Only one async operation can be started per operation handler invocation. Use TemporalNexusClient.client for additional workflow interactions", type=HandlerErrorType.BAD_REQUEST, ) diff --git a/tests/nexus/test_temporal_operation.py b/tests/nexus/test_temporal_operation.py index d470b58e59..030db6f6c9 100644 --- a/tests/nexus/test_temporal_operation.py +++ b/tests/nexus/test_temporal_operation.py @@ -336,10 +336,10 @@ async def bad_update_stage_op( async def query_op( self, _ctx: nexus.TemporalStartOperationContext, - _client: nexus.TemporalNexusClient, + client: nexus.TemporalNexusClient, input: str, ) -> nexus.TemporalOperationResult[bool]: - handle = nexus.client().get_workflow_handle(input) + handle = client.client.get_workflow_handle(input) result = await handle.query(BlockingWorkflow.query_done) return nexus.TemporalOperationResult.sync(result) @@ -434,20 +434,18 @@ async def mixed_start( async def double_signal( self, _ctx: nexus.TemporalStartOperationContext, - _client: nexus.TemporalNexusClient, + client: nexus.TemporalNexusClient, input: str, ) -> nexus.TemporalOperationResult[None]: - await nexus.client().start_workflow( + 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 ( - nexus.client() - .get_workflow_handle(f"{input}-other") - .signal(SignalCountWorkflow.my_signal) + await client.client.get_workflow_handle(f"{input}-other").signal( + SignalCountWorkflow.my_signal ) raise RuntimeError("Expected second signal to be rejected") @@ -455,10 +453,10 @@ async def double_signal( async def concurrent_signal( self, _ctx: nexus.TemporalStartOperationContext, - _client: nexus.TemporalNexusClient, + client: nexus.TemporalNexusClient, input: str, ) -> nexus.TemporalOperationResult[None]: - handle = nexus.client().get_workflow_handle(input) + handle = client.client.get_workflow_handle(input) self.concurrent_signal_results = list( await asyncio.gather( handle.signal(SignalCountWorkflow.my_signal), @@ -478,10 +476,10 @@ async def unlimited_signals(self, _ctx: StartOperationContext, input: str) -> No async def signal_after_failed_signal( self, _ctx: nexus.TemporalStartOperationContext, - _client: nexus.TemporalNexusClient, + client: nexus.TemporalNexusClient, input: str, ) -> nexus.TemporalOperationResult[None]: - handle = nexus.client().get_workflow_handle(input) + handle = client.client.get_workflow_handle(input) try: await handle.signal(SignalCountWorkflow.my_signal) except temporalio.service.RPCError: @@ -493,10 +491,10 @@ async def signal_after_failed_signal( async def signal_handler_retry( self, ctx: nexus.TemporalStartOperationContext, - _client: nexus.TemporalNexusClient, + client: nexus.TemporalNexusClient, input: str, ) -> nexus.TemporalOperationResult[None]: - await nexus.client().start_workflow( + await client.client.start_workflow( SignalCountWorkflow.run, id=input, task_queue=nexus.info().task_queue,