Dedup retried Nexus signals

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.

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.

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]

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:

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)

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]: