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; a second raises a `BAD_REQUEST` error, unless the first signal failed. Signals from other Nexus operation handlers are not limited.

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.

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
         )

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:

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: