Make Nexus bypass-path signal/signalWithStart request IDs redelivery-safe without colliding

temporal-sdk/src/main/java/io/temporal/internal/client/RootWorkflowClientInvoker.java

@@ -132,10 +132,23 @@ public WorkflowSignalOutput signal(WorkflowSignalInput input) {
             .setHeader(HeaderUtils.toHeaderGrpc(input.getHeader(), null));
 
     // If this signal is being issued from inside a Nexus operation handler, forward the inbound
-    // Nexus task links so the SignalWorkflowExecution history event links back to the caller.
+    // Nexus task links so the SignalWorkflowExecution history event links back to the caller, and
+    // derive a redelivery-safe request ID. We deliberately do NOT reuse the ambient
+    // nexusContext.getRequestId() verbatim here the way RootActivityClientInvoker does for
+    // activity starts: SignalWorkflowExecutionRequest's request_id is a pure dedup key with no
+    // awareness of signal name, payload, or target, so if a single Nexus operation handler
+    // invocation issues more than one signal-class call to the same workflow, reusing the same
+    // raw ambient ID for both would make the server treat the second call as a duplicate of the
+    // first and silently drop it. nextSignalRequestId() hands out a distinct-but-redelivery-stable
+    // ID per signal-class call instead. See InternalNexusOperationContext.nextSignalRequestId().
     boolean inNexusContext = CurrentNexusOperationContext.isNexusContext();
     if (inNexusContext) {
-      request.addAllLinks(CurrentNexusOperationContext.get().getRequestLinks());
+      InternalNexusOperationContext nexusContext = CurrentNexusOperationContext.get();
+      request.addAllLinks(nexusContext.getRequestLinks());
+      String signalRequestId = nexusContext.nextSignalRequestId();
+      if (signalRequestId != null) {
+        request.setRequestId(signalRequestId);
+      }
     }
 
     DataConverter dataConverterWitSignalContext =
@@ -176,10 +189,17 @@ public WorkflowSignalWithStartOutput signalWithStart(WorkflowSignalWithStartInpu
             startRequest, input.getSignalName(), signalInput.orElse(null));
     // If this signalWithStart is being issued from inside a Nexus operation handler, forward
     // the inbound Nexus task links so both the WorkflowExecutionStarted and
-    // WorkflowExecutionSignaled events on the callee link back to the caller.
+    // WorkflowExecutionSignaled events on the callee link back to the caller, and derive a
+    // redelivery-safe request ID the same way signal() does above -- see the comment there for why
+    // the raw ambient nexusContext.getRequestId() must not be reused verbatim.
     boolean inNexusContext = CurrentNexusOperationContext.isNexusContext();
     if (inNexusContext) {
-      requestBuilder.addAllLinks(CurrentNexusOperationContext.get().getRequestLinks());
+      InternalNexusOperationContext nexusContext = CurrentNexusOperationContext.get();
+      requestBuilder.addAllLinks(nexusContext.getRequestLinks());
+      String signalRequestId = nexusContext.nextSignalRequestId();
+      if (signalRequestId != null) {
+        requestBuilder.setRequestId(signalRequestId);
+      }
     }
     SignalWithStartWorkflowExecutionRequest request = requestBuilder.build();
     SignalWithStartWorkflowExecutionResponse response = genericClient.signalWithStart(request);

temporal-sdk/src/main/java/io/temporal/internal/nexus/InternalNexusOperationContext.java

@@ -9,6 +9,7 @@
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
+import java.util.concurrent.atomic.AtomicInteger;
 import javax.annotation.Nonnull;
 
 public class InternalNexusOperationContext {
@@ -33,6 +34,16 @@ public class InternalNexusOperationContext {
   // nexusOperationMetadata, which is scoped to the single backing start because it carries
   // completion-callback semantics.
   private String requestId;
+  // Counter used to derive distinct-but-stable request IDs for signal-class RPCs (signal,
+  // signalWithStart) issued during this invocation. Unlike activity starts, a signal has no
+  // per-call unique identifier of its own (server-side dedup for
+  // Signal/SignalWithStartWorkflowExecutionRequest is keyed purely on request_id, with no
+  // awareness of signal name, payload, or target), so reusing the raw ambient requestId verbatim
+  // for more than one signal-class call in the same invocation would make the server treat the
+  // second call as a duplicate of the first and silently drop it. See nextSignalRequestId(). A
+  // handler may issue RPCs from multiple threads (see responseLinksLock below), so this must be
+  // thread-safe.
+  private final AtomicInteger signalRequestIdSequence = new AtomicInteger();
   // Links returned by outbound RPCs the operation handler issues (such as
   // SignalWorkflowExecutionResponse.link or SignalWithStartWorkflowExecutionResponse.signal_link).
   // One entry per outbound RPC that returned a link. Drained
@@ -122,6 +133,30 @@ public String getRequestId() {
     return requestId;
   }
 
+  /**
+   * Returns a request ID for a signal-class RPC (signal / signalWithStart) issued during this
+   * invocation.
+   *
+   * <p>The returned ID is stable across a Nexus task redelivery for the Nth such call issued by
+   * this invocation (assuming the handler reissues the same sequence of calls on retry -- the same
+   * determinism assumption the ambient requestId/requestLinks design already relies on), which
+   * makes redelivered signal-class calls redelivery-safe against the server's request-ID based
+   * dedup. Unlike {@link #getRequestId()}, repeated calls within the same invocation return
+   * distinct values, so two different signal-class calls issued by one invocation (e.g. a
+   * signalWithStart followed by a plain signal to the same workflow) never collide on the server's
+   * dedup key.
+   *
+   * @return a derived, per-call request ID, or {@code null} if no ambient requestId is set (outside
+   *     a Nexus context, or a bare context not populated by {@code NexusTaskHandlerImpl}),
+   *     signaling callers to fall back to a fresh random ID.
+   */
+  public String nextSignalRequestId() {
+    if (requestId == null || requestId.isEmpty()) {
+      return null;
+    }
+    return requestId + "-" + signalRequestIdSequence.getAndIncrement();
+  }
+
   public void setStartWorkflowResponseLink(Link link) {
     this.startWorkflowResponseLink = link;
   }

temporal-sdk/src/test/java/io/temporal/internal/client/RootWorkflowClientInvokerTest.java

@@ -0,0 +1,243 @@
+package io.temporal.internal.client;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import com.uber.m3.tally.NoopScope;
+import io.temporal.api.common.v1.WorkflowExecution;
+import io.temporal.api.workflowservice.v1.SignalWithStartWorkflowExecutionRequest;
+import io.temporal.api.workflowservice.v1.SignalWithStartWorkflowExecutionResponse;
+import io.temporal.api.workflowservice.v1.SignalWorkflowExecutionRequest;
+import io.temporal.api.workflowservice.v1.SignalWorkflowExecutionResponse;
+import io.temporal.client.WorkflowClient;
+import io.temporal.client.WorkflowClientOptions;
+import io.temporal.client.WorkflowOptions;
+import io.temporal.common.interceptors.Header;
+import io.temporal.common.interceptors.WorkflowClientCallsInterceptor.WorkflowSignalInput;
+import io.temporal.common.interceptors.WorkflowClientCallsInterceptor.WorkflowSignalWithStartInput;
+import io.temporal.common.interceptors.WorkflowClientCallsInterceptor.WorkflowStartInput;
+import io.temporal.internal.client.external.GenericWorkflowClient;
+import io.temporal.internal.nexus.CurrentNexusOperationContext;
+import io.temporal.internal.nexus.InternalNexusOperationContext;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.ArgumentCaptor;
+
+/**
+ * Unit tests for signal-class request-ID derivation by {@link RootWorkflowClientInvoker}, in
+ * particular the redelivery-safety / collision-avoidance behavior of {@code signal()} and {@code
+ * signalWithStart()} when issued from inside a Nexus operation handler.
+ */
+public class RootWorkflowClientInvokerTest {
+
+  private static final String NAMESPACE = "test-namespace";
+
+  private GenericWorkflowClient genericClient;
+  private RootWorkflowClientInvoker invoker;
+  private InternalNexusOperationContext nexusContext;
+
+  @Before
+  public void setUp() {
+    genericClient = mock(GenericWorkflowClient.class);
+    when(genericClient.signal(any(SignalWorkflowExecutionRequest.class)))
+        .thenReturn(SignalWorkflowExecutionResponse.newBuilder().build());
+    when(genericClient.signalWithStart(any(SignalWithStartWorkflowExecutionRequest.class)))
+        .thenReturn(
+            SignalWithStartWorkflowExecutionResponse.newBuilder().setRunId("run-id").build());
+    invoker =
+        new RootWorkflowClientInvoker(
+            genericClient,
+            WorkflowClientOptions.newBuilder()
+                .setNamespace(NAMESPACE)
+                .setIdentity("test-identity")
+                .validateAndBuildWithDefaults(),
+            new WorkerFactoryRegistry());
+    nexusContext =
+        new InternalNexusOperationContext(
+            NAMESPACE,
+            "test-task-queue",
+            "test-endpoint",
+            new NoopScope(),
+            mock(WorkflowClient.class));
+    CurrentNexusOperationContext.set(nexusContext);
+  }
+
+  @After
+  public void tearDown() {
+    CurrentNexusOperationContext.unset();
+  }
+
+  @Test
+  public void signalInNexusContextDerivesFromAmbientRequestIdRatherThanReusingItVerbatim() {
+    nexusContext.setRequestId("ambient-nexus-request-id");
+
+    invoker.signal(newSignalInput());
+
+    ArgumentCaptor<SignalWorkflowExecutionRequest> captor =
+        ArgumentCaptor.forClass(SignalWorkflowExecutionRequest.class);
+    verify(genericClient).signal(captor.capture());
+    String requestId = captor.getValue().getRequestId();
+    Assert.assertFalse(requestId.isEmpty());
+    Assert.assertNotEquals("ambient-nexus-request-id", requestId);
+    Assert.assertTrue(requestId.startsWith("ambient-nexus-request-id"));
+  }
+
+  @Test
+  public void twoSignalsInSameInvocationGetDistinctRequestIds() {
+    nexusContext.setRequestId("ambient-nexus-request-id");
+
+    invoker.signal(newSignalInput());
+    invoker.signal(newSignalInput());
+
+    ArgumentCaptor<SignalWorkflowExecutionRequest> captor =
+        ArgumentCaptor.forClass(SignalWorkflowExecutionRequest.class);
+    verify(genericClient, org.mockito.Mockito.times(2)).signal(captor.capture());
+    String firstRequestId = captor.getAllValues().get(0).getRequestId();
+    String secondRequestId = captor.getAllValues().get(1).getRequestId();
+    Assert.assertNotEquals(firstRequestId, secondRequestId);
+  }
+
+  @Test
+  public void signalThenSignalWithStartInSameInvocationGetDistinctRequestIds() {
+    nexusContext.setRequestId("ambient-nexus-request-id");
+
+    invoker.signal(newSignalInput());
+    invoker.signalWithStart(newSignalWithStartInput());
+
+    ArgumentCaptor<SignalWorkflowExecutionRequest> signalCaptor =
+        ArgumentCaptor.forClass(SignalWorkflowExecutionRequest.class);
+    verify(genericClient).signal(signalCaptor.capture());
+    ArgumentCaptor<SignalWithStartWorkflowExecutionRequest> signalWithStartCaptor =
+        ArgumentCaptor.forClass(SignalWithStartWorkflowExecutionRequest.class);
+    verify(genericClient).signalWithStart(signalWithStartCaptor.capture());
+
+    Assert.assertNotEquals(
+        signalCaptor.getValue().getRequestId(), signalWithStartCaptor.getValue().getRequestId());
+  }
+
+  @Test
+  public void sameSequenceOfCallsOnRedeliveredContextYieldsSameRequestIds() {
+    // Simulate two redelivery attempts of the same Nexus task: NexusTaskHandlerImpl.handle()
+    // creates a fresh InternalNexusOperationContext per attempt, but the server redelivers the
+    // same task, so both attempts get the same ambient requestId.
+    InternalNexusOperationContext attempt1 =
+        new InternalNexusOperationContext(
+            NAMESPACE, "tq", "endpoint", new NoopScope(), mock(WorkflowClient.class));
+    attempt1.setRequestId("redelivered-request-id");
+    InternalNexusOperationContext attempt2 =
+        new InternalNexusOperationContext(
+            NAMESPACE, "tq", "endpoint", new NoopScope(), mock(WorkflowClient.class));
+    attempt2.setRequestId("redelivered-request-id");
+
+    // Simulate the handler issuing the same two signal-class calls, in the same order, on each
+    // attempt.
+    String attempt1First = attempt1.nextSignalRequestId();
+    String attempt1Second = attempt1.nextSignalRequestId();
+    String attempt2First = attempt2.nextSignalRequestId();
+    String attempt2Second = attempt2.nextSignalRequestId();
+
+    Assert.assertEquals(attempt1First, attempt2First);
+    Assert.assertEquals(attempt1Second, attempt2Second);
+    Assert.assertNotEquals(attempt1First, attempt1Second);
+  }
+
+  @Test
+  public void signalOutsideNexusContextFallsBackToFreshRandomRequestIdEachCall() {
+    CurrentNexusOperationContext.unset();
+
+    invoker.signal(newSignalInput());
+    invoker.signal(newSignalInput());
+
+    ArgumentCaptor<SignalWorkflowExecutionRequest> captor =
+        ArgumentCaptor.forClass(SignalWorkflowExecutionRequest.class);
+    verify(genericClient, org.mockito.Mockito.times(2)).signal(captor.capture());
+    String firstRequestId = captor.getAllValues().get(0).getRequestId();
+    String secondRequestId = captor.getAllValues().get(1).getRequestId();
+    Assert.assertFalse(firstRequestId.isEmpty());
+    Assert.assertFalse(secondRequestId.isEmpty());
+    Assert.assertNotEquals(firstRequestId, secondRequestId);
+  }
+
+  @Test
+  public void signalInNexusContextWithoutAmbientRequestIdFallsBackToFreshRandomRequestId() {
+    // Nexus context is set (e.g. inside an operation handler), but no ambient requestId was ever
+    // populated (bare context not populated by NexusTaskHandlerImpl).
+    invoker.signal(newSignalInput());
+
+    ArgumentCaptor<SignalWorkflowExecutionRequest> captor =
+        ArgumentCaptor.forClass(SignalWorkflowExecutionRequest.class);
+    verify(genericClient).signal(captor.capture());
+    Assert.assertFalse(captor.getValue().getRequestId().isEmpty());
+  }
+
+  @Test
+  public void
+      signalWithStartInNexusContextDerivesFromAmbientRequestIdRatherThanReusingItVerbatim() {
+    nexusContext.setRequestId("ambient-nexus-request-id");
+
+    invoker.signalWithStart(newSignalWithStartInput());
+
+    ArgumentCaptor<SignalWithStartWorkflowExecutionRequest> captor =
+        ArgumentCaptor.forClass(SignalWithStartWorkflowExecutionRequest.class);
+    verify(genericClient).signalWithStart(captor.capture());
+    String requestId = captor.getValue().getRequestId();
+    Assert.assertFalse(requestId.isEmpty());
+    Assert.assertNotEquals("ambient-nexus-request-id", requestId);
+    Assert.assertTrue(requestId.startsWith("ambient-nexus-request-id"));
+  }
+
+  @Test
+  public void twoSignalWithStartsInSameInvocationGetDistinctRequestIds() {
+    nexusContext.setRequestId("ambient-nexus-request-id");
+
+    invoker.signalWithStart(newSignalWithStartInput());
+    invoker.signalWithStart(newSignalWithStartInput());
+
+    ArgumentCaptor<SignalWithStartWorkflowExecutionRequest> captor =
+        ArgumentCaptor.forClass(SignalWithStartWorkflowExecutionRequest.class);
+    verify(genericClient, org.mockito.Mockito.times(2)).signalWithStart(captor.capture());
+    String firstRequestId = captor.getAllValues().get(0).getRequestId();
+    String secondRequestId = captor.getAllValues().get(1).getRequestId();
+    Assert.assertNotEquals(firstRequestId, secondRequestId);
+  }
+
+  @Test
+  public void signalWithStartOutsideNexusContextFallsBackToFreshRandomRequestIdEachCall() {
+    CurrentNexusOperationContext.unset();
+
+    invoker.signalWithStart(newSignalWithStartInput());
+    invoker.signalWithStart(newSignalWithStartInput());
+
+    ArgumentCaptor<SignalWithStartWorkflowExecutionRequest> captor =
+        ArgumentCaptor.forClass(SignalWithStartWorkflowExecutionRequest.class);
+    verify(genericClient, org.mockito.Mockito.times(2)).signalWithStart(captor.capture());
+    String firstRequestId = captor.getAllValues().get(0).getRequestId();
+    String secondRequestId = captor.getAllValues().get(1).getRequestId();
+    Assert.assertFalse(firstRequestId.isEmpty());
+    Assert.assertFalse(secondRequestId.isEmpty());
+    Assert.assertNotEquals(firstRequestId, secondRequestId);
+  }
+
+  private static WorkflowSignalInput newSignalInput() {
+    return new WorkflowSignalInput(
+        WorkflowExecution.newBuilder().setWorkflowId("callee-workflow-id").build(),
+        "test-signal",
+        Header.empty(),
+        new Object[0]);
+  }
+
+  private static WorkflowSignalWithStartInput newSignalWithStartInput() {
+    WorkflowStartInput startInput =
+        new WorkflowStartInput(
+            "callee-workflow-id",
+            "TestWorkflow",
+            Header.empty(),
+            new Object[0],
+            WorkflowOptions.newBuilder().build());
+    return new WorkflowSignalWithStartInput(startInput, "test-signal", new Object[0]);
+  }
+}