From 7100ee4448359db7bd81aff6de31ca92ef8c0da5 Mon Sep 17 00:00:00 2001 From: tekkaya <86028633+tekkaya@users.noreply.github.com> Date: Fri, 28 Aug 2026 13:38:47 -0700 Subject: [PATCH] Make Nexus bypass-path signal/signalWithStart request IDs redelivery-safe without colliding RootWorkflowClientInvoker.signal/signalWithStart always generate a fresh UUID for the outgoing request_id, even when issued from inside a Nexus operation handler. That means a redelivered Nexus task can apply the same signal twice: the retried handler re-issues the call with a brand-new random ID, so the server has nothing to dedup against. The naive fix -- reuse the ambient Nexus request ID the same way RootActivityClientInvoker.startActivity already does for bypass-path activity starts -- is unsafe here. Confirmed by reading the server's dedup code directly (service/history/api/signalworkflow/api.go and signalwithstartworkflow/signal_with_start_workflow.go, both keying off mutableState.IsSignalRequested/AddSignalRequested in mutable_state_impl.go): request_id on Signal(WithStart)WorkflowExecutionRequest is a pure dedup key with no awareness of signal name, payload, or target. Unlike activities, which already carry their own unique activity_id so sharing one ambient request ID across multiple starts in one invocation never collides, a signal has no such identifier. Reusing the same raw ambient ID for two different signal-class calls to the same workflow in one invocation would make the server treat the second as a duplicate of the first and silently drop it -- never delivered, no error. InternalNexusOperationContext.nextSignalRequestId() derives a per-call ID instead: ambientRequestId + "-" + N, where N is a per-invocation counter. This keeps the Nth signal-class call's ID stable across a redelivery (each redelivery attempt gets a fresh InternalNexusOperationContext but the same ambient request ID, and the handler is expected to reissue the same sequence of calls), while giving distinct calls within one invocation distinct IDs so they can't collide. RootWorkflowClientInvokerTest covers: ID derived-but-not-verbatim for both signal and signalWithStart; two calls (including a mixed signal + signalWithStart pair) in one invocation getting distinct IDs; redelivery stability across two separate context instances sharing the same ambient ID; and the existing fallback-to-fresh-UUID behavior outside a Nexus context or with no ambient ID set. --- .../client/RootWorkflowClientInvoker.java | 28 +- .../nexus/InternalNexusOperationContext.java | 35 +++ .../client/RootWorkflowClientInvokerTest.java | 243 ++++++++++++++++++ 3 files changed, 302 insertions(+), 4 deletions(-) create mode 100644 temporal-sdk/src/test/java/io/temporal/internal/client/RootWorkflowClientInvokerTest.java 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 502c12e8ee..294b161924 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 3c5a6b0af8..ab034763b8 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 0000000000..ef0987828a --- /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]); + } +}