diff --git a/temporal-sdk/src/main/java/io/temporal/internal/client/RootWorkflowClientInvoker.java b/temporal-sdk/src/main/java/io/temporal/internal/client/RootWorkflowClientInvoker.java index 502c12e8e..294b16192 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/client/RootWorkflowClientInvoker.java +++ b/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); diff --git a/temporal-sdk/src/main/java/io/temporal/internal/nexus/InternalNexusOperationContext.java b/temporal-sdk/src/main/java/io/temporal/internal/nexus/InternalNexusOperationContext.java index 3c5a6b0af..ab034763b 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/nexus/InternalNexusOperationContext.java +++ b/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. + * + *

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; } diff --git a/temporal-sdk/src/test/java/io/temporal/internal/client/RootWorkflowClientInvokerTest.java b/temporal-sdk/src/test/java/io/temporal/internal/client/RootWorkflowClientInvokerTest.java new file mode 100644 index 000000000..ef0987828 --- /dev/null +++ b/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 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 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 signalCaptor = + ArgumentCaptor.forClass(SignalWorkflowExecutionRequest.class); + verify(genericClient).signal(signalCaptor.capture()); + ArgumentCaptor 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 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 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 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 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 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]); + } +}