From 32eb497c3b741ccda184536f80c579a7dce8b0f9 Mon Sep 17 00:00:00 2001 From: Alice Lin Date: Thu, 29 Oct 2026 16:42:10 -0700 Subject: [PATCH] Dedup retried Nexus signals --- CHANGELOG.md | 8 + .../io/temporal/client/WorkflowStubImpl.java | 22 ++ .../client/RootActivityClientInvoker.java | 3 +- .../client/RootWorkflowClientInvoker.java | 32 ++- .../nexus/CurrentNexusOperationContext.java | 7 + .../nexus/InternalNexusOperationContext.java | 52 ++++ .../temporal/nexus/TemporalNexusClient.java | 13 +- .../nexus/TemporalOperationHandler.java | 1 + .../client/NexusSignalDeduplicationTest.java | 229 ++++++++++++++++++ ...GenericHandlerSignalDeduplicationTest.java | 218 +++++++++++++++++ 10 files changed, 572 insertions(+), 13 deletions(-) create mode 100644 temporal-sdk/src/test/java/io/temporal/client/NexusSignalDeduplicationTest.java create mode 100644 temporal-sdk/src/test/java/io/temporal/workflow/nexus/GenericHandlerSignalDeduplicationTest.java diff --git a/CHANGELOG.md b/CHANGELOG.md index a390056d46..70cbcecfae 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -23,6 +23,11 @@ to docs, or any other relevant information. - `WorkflowOutboundCallsInterceptor` has new `sleep(Duration, TimerOptions)` and `await(Duration, TimerOptions, String, Supplier)` methods. Implementations that don't extend `WorkflowOutboundCallsInterceptorBase` need to add them, and Base subclasses override them to see calls with options. +- **Experimental**: A `@TemporalOperation` handler can now send only one signal per operation invocation; `signal` + and `signalWithStart` count toward the same limit. While the first signal is in flight, or after it succeeds, a + second signal to any workflow throws a `BAD_REQUEST` `HandlerException`. If the first signal fails, the handler can + signal again. `signalWithStart` from such a handler also requires an explicit workflow ID, since a defaulted ID + differs on each retried Nexus task and would start another workflow. ### Added - Applications can use the experimental `@TransferTypeConvertible` annotation and @@ -60,6 +65,9 @@ to docs, or any other relevant information. `ActivityCompletionClient.withContext(NexusSerializationContext)`. ### Changed +- **Experimental**: `signal` and `signalWithStart` called from a `@TemporalOperation` handler now reuse the + operation's request ID, so a retried Nexus task no longer delivers the same signal twice. Signals sent from other + Nexus operation handlers keep an independent request ID per call and are not deduplicated. - Moved the Spring AI integration to [temporalio/ai-integrations](https://github.com/temporalio/ai-integrations/tree/main/java/spring-ai). `io.temporal:temporal-spring-ai` now publishes a one-time relocation POM to `io.temporal:spring-ai:0.1.0`; the integration is versioned independently of the SDK. The standalone release requires Java 17+, Spring AI 2, diff --git a/temporal-sdk/src/main/java/io/temporal/client/WorkflowStubImpl.java b/temporal-sdk/src/main/java/io/temporal/client/WorkflowStubImpl.java index 69018ce65f..c1bf97d6ae 100644 --- a/temporal-sdk/src/main/java/io/temporal/client/WorkflowStubImpl.java +++ b/temporal-sdk/src/main/java/io/temporal/client/WorkflowStubImpl.java @@ -12,6 +12,8 @@ import io.temporal.common.interceptors.WorkflowClientCallsInterceptor; import io.temporal.failure.CanceledFailure; import io.temporal.internal.client.LazyWorkflowUpdateHandleImpl; +import io.temporal.internal.nexus.CurrentNexusOperationContext; +import io.temporal.internal.nexus.InternalNexusOperationContext; import io.temporal.serviceclient.CheckedExceptionWrapper; import io.temporal.serviceclient.StatusUtils; import java.lang.reflect.Type; @@ -77,13 +79,23 @@ class WorkflowStubImpl implements WorkflowStub { public void signal(String signalName, Object... args) { checkStarted(); WorkflowExecution targetExecution = currentExecutionCheckLegacy(); + InternalNexusOperationContext nexusContext = CurrentNexusOperationContext.getOrNull(); + if (nexusContext != null) { + nexusContext.claimSignal(); + } + boolean sent = false; try { workflowClientInvoker.signal( new WorkflowClientCallsInterceptor.WorkflowSignalInput( targetExecution, signalName, Header.empty(), args)); + sent = true; } catch (Exception e) { Throwable throwable = throwAsWorkflowFailureException(e, targetExecution); throw new WorkflowServiceException(targetExecution, workflowType.orElse(null), throwable); + } finally { + if (!sent && nexusContext != null) { + nexusContext.releaseSignal(); + } } } @@ -182,8 +194,13 @@ public R executeUpdateWithStart( private WorkflowExecution signalWithStartWithOptions( WorkflowOptions options, String signalName, Object[] signalArgs, Object[] startArgs) { checkExecutionIsNotStarted(); + InternalNexusOperationContext nexusContext = CurrentNexusOperationContext.getOrNull(); + if (nexusContext != null) { + nexusContext.claimSignalWithStart(options.getWorkflowId()); + } String workflowId = getWorkflowIdForStart(options); WorkflowExecution workflowExecution = null; + boolean sent = false; try { WorkflowClientCallsInterceptor.WorkflowSignalWithStartOutput workflowStartOutput = workflowClientInvoker.signalWithStart( @@ -194,6 +211,7 @@ private WorkflowExecution signalWithStartWithOptions( signalArgs)); workflowExecution = workflowStartOutput.getWorkflowStartOutput().getWorkflowExecution(); populateExecutionAfterStart(workflowExecution); + sent = true; return workflowExecution; } catch (StatusRuntimeException e) { throw wrapStartException(workflowId, workflowType.orElse(null), e); @@ -206,6 +224,10 @@ private WorkflowExecution signalWithStartWithOptions( workflowExecution = WorkflowExecution.newBuilder().setWorkflowId(workflowId).build(); } throw new WorkflowServiceException(workflowExecution, workflowType.orElse(null), e); + } finally { + if (!sent && nexusContext != null) { + nexusContext.releaseSignal(); + } } } diff --git a/temporal-sdk/src/main/java/io/temporal/internal/client/RootActivityClientInvoker.java b/temporal-sdk/src/main/java/io/temporal/internal/client/RootActivityClientInvoker.java index 6c8f801cf6..b4c97fbfce 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/client/RootActivityClientInvoker.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/client/RootActivityClientInvoker.java @@ -85,8 +85,7 @@ public StartActivityOutput startActivity(StartActivityInput input) { new ActivitySerializationContext( namespace, null, null, input.getActivityType(), options.getTaskQueue(), false)); - InternalNexusOperationContext nexusContext = - CurrentNexusOperationContext.isNexusContext() ? CurrentNexusOperationContext.get() : null; + InternalNexusOperationContext nexusContext = CurrentNexusOperationContext.getOrNull(); NexusOperationMetadata nexusOperationMetadata = nexusContext == null ? null : nexusContext.getNexusOperationMetadata(); 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 147a7bb7a9..7a8499f1ea 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 @@ -227,9 +227,16 @@ public WorkflowSignalOutput signal(WorkflowSignalInput input) { // 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. - boolean inNexusContext = CurrentNexusOperationContext.isNexusContext(); - if (inNexusContext) { - request.addAllLinks(CurrentNexusOperationContext.get().getRequestLinks()); + InternalNexusOperationContext nexusContext = CurrentNexusOperationContext.getOrNull(); + if (nexusContext != null) { + request.addAllLinks(nexusContext.getRequestLinks()); + // From a Temporal operation handler, reuse the operation's request ID so a retried Nexus task + // dedupes against the first attempt. WorkflowStubImpl limits such signals to one per + // invocation, since the server would drop a second one as a duplicate. + String requestId = nexusContext.getSignalRequestId(); + if (requestId != null) { + request.setRequestId(requestId); + } } DataConverter dataConverterWitSignalContext = workflowConverter(input.getWorkflowExecution()); @@ -244,8 +251,8 @@ public WorkflowSignalOutput signal(WorkflowSignalInput input) { SignalWorkflowExecutionResponse response = genericClient.signal(request.build()); // Server >=1.31 with EnableCHASMSignalBacklinks returns a response link pointing at the signal // event; older servers leave it unset. Propagate when present. - if (inNexusContext && response.hasLink()) { - CurrentNexusOperationContext.get().addResponseLink(response.getLink()); + if (nexusContext != null && response.hasLink()) { + nexusContext.addResponseLink(response.getLink()); } return new WorkflowSignalOutput(); } @@ -268,9 +275,14 @@ public WorkflowSignalWithStartOutput signalWithStart(WorkflowSignalWithStartInpu // 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. - boolean inNexusContext = CurrentNexusOperationContext.isNexusContext(); - if (inNexusContext) { - requestBuilder.addAllLinks(CurrentNexusOperationContext.get().getRequestLinks()); + InternalNexusOperationContext nexusContext = CurrentNexusOperationContext.getOrNull(); + if (nexusContext != null) { + requestBuilder.addAllLinks(nexusContext.getRequestLinks()); + // Reuse the operation's request ID, as in signal(). + String requestId = nexusContext.getSignalRequestId(); + if (requestId != null) { + requestBuilder.setRequestId(requestId); + } } SignalWithStartWorkflowExecutionRequest request = requestBuilder.build(); SignalWithStartWorkflowExecutionResponse response = genericClient.signalWithStart(request); @@ -281,8 +293,8 @@ public WorkflowSignalWithStartOutput signalWithStart(WorkflowSignalWithStartInpu .build(); // Server >=1.31 with EnableCHASMSignalBacklinks returns a response link pointing at the signal // event; older servers leave it unset. Propagate when present. - if (inNexusContext && response.hasSignalLink()) { - CurrentNexusOperationContext.get().addResponseLink(response.getSignalLink()); + if (nexusContext != null && response.hasSignalLink()) { + nexusContext.addResponseLink(response.getSignalLink()); } // TODO currently SignalWithStartWorkflowExecutionResponse doesn't have eagerWorkflowTask. // We should wire it when it's implemented server-side. diff --git a/temporal-sdk/src/main/java/io/temporal/internal/nexus/CurrentNexusOperationContext.java b/temporal-sdk/src/main/java/io/temporal/internal/nexus/CurrentNexusOperationContext.java index 273cab5303..87ec327c15 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/nexus/CurrentNexusOperationContext.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/nexus/CurrentNexusOperationContext.java @@ -1,5 +1,7 @@ package io.temporal.internal.nexus; +import javax.annotation.Nullable; + /** * Thread local store of the context object passed to a nexus operation implementation. Not to be * used directly. @@ -21,6 +23,11 @@ public static InternalNexusOperationContext get() { return CURRENT.get(); } + /** The current context, or {@code null} outside a Nexus operation handler. */ + public static @Nullable InternalNexusOperationContext getOrNull() { + return CURRENT.get(); + } + public static void set(InternalNexusOperationContext context) { if (context == null) { throw new IllegalArgumentException("null context"); 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 91a0e398cb..b137db7cd9 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 @@ -1,6 +1,8 @@ package io.temporal.internal.nexus; +import com.google.common.base.Strings; import com.uber.m3.tally.Scope; +import io.nexusrpc.handler.HandlerException; import io.temporal.api.common.v1.Link; import io.temporal.client.WorkflowClient; import io.temporal.common.interceptors.NexusOperationOutboundCallsInterceptor; @@ -10,6 +12,7 @@ import java.util.ArrayList; import java.util.Collections; import java.util.List; +import java.util.concurrent.atomic.AtomicBoolean; import javax.annotation.Nonnull; import javax.annotation.Nullable; @@ -51,6 +54,12 @@ public class InternalNexusOperationContext { // service and operation names are known, which is only after the request variant has been // inspected, so it is null while the task is being dispatched. private NexusSerializationContext serializationContext; + // Set while a TemporalOperationHandler start runs. Signals and signalWithStarts issued then reuse + // requestId so a retried Nexus task dedupes against the first attempt. Because they share the ID, + // a second signal could be dropped by the server as a duplicate, so signalClaimed limits the + // invocation to one. + private boolean signalDeduplication; + private final AtomicBoolean signalClaimed = new AtomicBoolean(); public InternalNexusOperationContext( String namespace, @@ -144,6 +153,49 @@ public String getRequestId() { return requestId; } + /** Makes signals reuse {@link #getRequestId} and limits them to one; see {@link #claimSignal}. */ + public void enableSignalDeduplication() { + this.signalDeduplication = true; + } + + /** The request ID signals must reuse; {@code null} outside a Temporal operation start. */ + public @Nullable String getSignalRequestId() { + return signalDeduplication && !Strings.isNullOrEmpty(requestId) ? requestId : null; + } + + /** + * Claims the invocation's single signal. No-op unless {@link #enableSignalDeduplication} was + * called. Callers must {@link #releaseSignal} if the signal fails, so the handler can send it + * again. + */ + public void claimSignal() { + if (signalDeduplication && !signalClaimed.compareAndSet(false, true)) { + throw new HandlerException( + HandlerException.ErrorType.BAD_REQUEST, + new IllegalStateException("Only one signal can be sent per Nexus operation invocation.")); + } + } + + /** + * Like {@link #claimSignal}, but also rejects an empty workflow ID: it would be defaulted to a + * fresh UUID on each retried Nexus task, so the server could not dedupe and would start another + * workflow. + */ + public void claimSignalWithStart(@Nullable String workflowId) { + if (signalDeduplication && Strings.isNullOrEmpty(workflowId)) { + throw new HandlerException( + HandlerException.ErrorType.BAD_REQUEST, + new IllegalArgumentException( + "signalWithStart from a Nexus operation handler requires a workflow ID.")); + } + claimSignal(); + } + + /** Frees the signal claimed by {@link #claimSignal} after the signal failed. */ + public void releaseSignal() { + signalClaimed.set(false); + } + public void setStartWorkflowResponseLink(Link link) { this.startWorkflowResponseLink = link; } diff --git a/temporal-sdk/src/main/java/io/temporal/nexus/TemporalNexusClient.java b/temporal-sdk/src/main/java/io/temporal/nexus/TemporalNexusClient.java index 8b70f6f515..6566c29f71 100644 --- a/temporal-sdk/src/main/java/io/temporal/nexus/TemporalNexusClient.java +++ b/temporal-sdk/src/main/java/io/temporal/nexus/TemporalNexusClient.java @@ -50,7 +50,18 @@ @Experimental public interface TemporalNexusClient { - /** Returns the underlying {@link WorkflowClient} for advanced use cases. */ + /** + * Returns the underlying {@link WorkflowClient} for advanced use cases. + * + *

Signals sent from the operation handler carry the 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 invocation; a second throws a {@link + * io.nexusrpc.handler.HandlerException.ErrorType#BAD_REQUEST} {@link + * io.nexusrpc.handler.HandlerException} rather than being silently dropped by the server. For the + * same reason {@code signalWithStart} requires an explicit workflow ID. This applies to every + * signal sent on the handler's thread, including through {@link + * NexusOperationContext#getWorkflowClient()}. + */ WorkflowClient getWorkflowClient(); /** diff --git a/temporal-sdk/src/main/java/io/temporal/nexus/TemporalOperationHandler.java b/temporal-sdk/src/main/java/io/temporal/nexus/TemporalOperationHandler.java index 9cb88229a4..f4266cb67d 100644 --- a/temporal-sdk/src/main/java/io/temporal/nexus/TemporalOperationHandler.java +++ b/temporal-sdk/src/main/java/io/temporal/nexus/TemporalOperationHandler.java @@ -88,6 +88,7 @@ public final OperationStartResult start( OperationContext ctx, OperationStartDetails details, T input) throws OperationException, HandlerException { InternalNexusOperationContext nexusCtx = CurrentNexusOperationContext.get(); + nexusCtx.enableSignalDeduplication(); TemporalNexusClient client = new TemporalNexusClientImpl(nexusCtx.getWorkflowClient(), ctx, details); diff --git a/temporal-sdk/src/test/java/io/temporal/client/NexusSignalDeduplicationTest.java b/temporal-sdk/src/test/java/io/temporal/client/NexusSignalDeduplicationTest.java new file mode 100644 index 0000000000..22f5aa3812 --- /dev/null +++ b/temporal-sdk/src/test/java/io/temporal/client/NexusSignalDeduplicationTest.java @@ -0,0 +1,229 @@ +package io.temporal.client; + +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.clearInvocations; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.mockingDetails; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import com.uber.m3.tally.NoopScope; +import io.grpc.Status; +import io.nexusrpc.handler.HandlerException; +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.internal.client.RootWorkflowClientInvoker; +import io.temporal.internal.client.WorkerFactoryRegistry; +import io.temporal.internal.client.external.GenericWorkflowClient; +import io.temporal.internal.nexus.CurrentNexusOperationContext; +import io.temporal.internal.nexus.InternalNexusOperationContext; +import java.util.Optional; +import org.junit.After; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; +import org.mockito.ArgumentCaptor; + +/** + * Signals sent while a Temporal operation handler runs reuse the operation's request ID, so a + * retried Nexus task dedupes against the first attempt. Because they share that ID, only one signal + * is allowed per invocation. + */ +public class NexusSignalDeduplicationTest { + private static final String REQUEST_ID = "nexus-op-request-id"; + private static final String WORKFLOW_ID = "wf-target"; + + private GenericWorkflowClient genericClient; + private WorkflowClientOptions options; + private RootWorkflowClientInvoker invoker; + private InternalNexusOperationContext nexusContext; + + @Before + public void setUp() { + genericClient = mock(GenericWorkflowClient.class); + when(genericClient.signal(any())) + .thenReturn(SignalWorkflowExecutionResponse.getDefaultInstance()); + when(genericClient.signalWithStart(any())) + .thenReturn(SignalWithStartWorkflowExecutionResponse.newBuilder().setRunId("run").build()); + options = + WorkflowClientOptions.newBuilder() + .setNamespace("test-namespace") + .validateAndBuildWithDefaults(); + invoker = + new RootWorkflowClientInvoker( + genericClient, + options.getNamespace(), + options.getIdentity(), + options.getQueryRejectCondition(), + options.getContextPropagators(), + options.getDataConverter(), + new WorkerFactoryRegistry()); + nexusContext = + new InternalNexusOperationContext( + options.getNamespace(), "tq", "endpoint", new NoopScope(), mock(WorkflowClient.class)); + CurrentNexusOperationContext.set(nexusContext); + } + + @After + public void tearDown() { + CurrentNexusOperationContext.unset(); + } + + private void enableSignalDeduplication() { + nexusContext.setRequestId(REQUEST_ID); + nexusContext.enableSignalDeduplication(); + } + + private void signal(String workflowId) { + new WorkflowStubImpl( + options, + invoker, + Optional.empty(), + WorkflowExecution.newBuilder().setWorkflowId(workflowId).build(), + false, + null) + .signal("test-signal", "payload"); + } + + private void signalWithStart(String workflowId) { + new WorkflowStubImpl( + options, + invoker, + "TestWorkflow", + WorkflowOptions.newBuilder().setWorkflowId(workflowId).setTaskQueue("tq").build()) + .signalWithStart("test-signal", new Object[] {"payload"}, new Object[] {}); + } + + private String sentSignalRequestId() { + ArgumentCaptor captor = + ArgumentCaptor.forClass(SignalWorkflowExecutionRequest.class); + verify(genericClient).signal(captor.capture()); + return captor.getValue().getRequestId(); + } + + private String sentSignalWithStartRequestId() { + ArgumentCaptor captor = + ArgumentCaptor.forClass(SignalWithStartWorkflowExecutionRequest.class); + verify(genericClient).signalWithStart(captor.capture()); + return captor.getValue().getRequestId(); + } + + private static void assertOneSignalRejection(Runnable second) { + HandlerException e = Assert.assertThrows(HandlerException.class, second::run); + Assert.assertEquals(HandlerException.ErrorType.BAD_REQUEST, e.getErrorType()); + Assert.assertEquals( + "Only one signal can be sent per Nexus operation invocation.", e.getCause().getMessage()); + } + + @Test + public void signalUsesOperationRequestId() { + enableSignalDeduplication(); + signal(WORKFLOW_ID); + Assert.assertEquals(REQUEST_ID, sentSignalRequestId()); + } + + @Test + public void signalWithStartUsesOperationRequestId() { + enableSignalDeduplication(); + signalWithStart(WORKFLOW_ID); + Assert.assertEquals(REQUEST_ID, sentSignalWithStartRequestId()); + } + + /** + * In a Nexus handler other than a Temporal operation start, signals keep generating their own + * request ID and aren't limited to one. + */ + @Test + public void signalsWithoutDeduplicationKeepOwnRequestId() { + nexusContext.setRequestId(REQUEST_ID); + signal(WORKFLOW_ID); + signalWithStart(WORKFLOW_ID); + Assert.assertNotEquals(REQUEST_ID, sentSignalRequestId()); + Assert.assertNotEquals(REQUEST_ID, sentSignalWithStartRequestId()); + } + + /** + * Any second signal or signalWithStart is rejected, whatever workflow it targets, since the + * server could dedupe it against the first. + */ + @Test + public void onlyOneSignalPerInvocation() { + Runnable[][] cases = { + {() -> signal(WORKFLOW_ID), () -> signal(WORKFLOW_ID)}, + {() -> signal(WORKFLOW_ID), () -> signal("other-wf")}, + {() -> signal(WORKFLOW_ID), () -> signalWithStart(WORKFLOW_ID)}, + {() -> signalWithStart(WORKFLOW_ID), () -> signal(WORKFLOW_ID)}, + {() -> signalWithStart(WORKFLOW_ID), () -> signalWithStart(WORKFLOW_ID)}, + }; + enableSignalDeduplication(); + for (Runnable[] c : cases) { + nexusContext.releaseSignal(); + clearInvocations(genericClient); + c[0].run(); + assertOneSignalRejection(c[1]); + // The rejected call must not reach the server. + Assert.assertEquals(1, mockingDetails(genericClient).getInvocations().size()); + } + } + + /** + * A failed signal frees the slot so the handler can resend it, and the resend reuses the request + * ID. + */ + @Test + public void failedSignalDoesNotConsumeSlot() { + enableSignalDeduplication(); + when(genericClient.signal(any())) + .thenThrow(Status.UNAVAILABLE.withDescription("try again").asRuntimeException()) + .thenReturn(SignalWorkflowExecutionResponse.getDefaultInstance()); + + Assert.assertThrows(WorkflowServiceException.class, () -> signal(WORKFLOW_ID)); + signal(WORKFLOW_ID); + + ArgumentCaptor captor = + ArgumentCaptor.forClass(SignalWorkflowExecutionRequest.class); + verify(genericClient, times(2)).signal(captor.capture()); + for (SignalWorkflowExecutionRequest request : captor.getAllValues()) { + Assert.assertEquals(REQUEST_ID, request.getRequestId()); + } + } + + @Test + public void failedSignalWithStartDoesNotConsumeSlot() { + enableSignalDeduplication(); + when(genericClient.signalWithStart(any())) + .thenThrow(Status.UNAVAILABLE.withDescription("try again").asRuntimeException()) + .thenReturn(SignalWithStartWorkflowExecutionResponse.newBuilder().setRunId("run").build()); + + Assert.assertThrows(WorkflowServiceException.class, () -> signalWithStart(WORKFLOW_ID)); + signalWithStart(WORKFLOW_ID); + + ArgumentCaptor captor = + ArgumentCaptor.forClass(SignalWithStartWorkflowExecutionRequest.class); + verify(genericClient, times(2)).signalWithStart(captor.capture()); + for (SignalWithStartWorkflowExecutionRequest request : captor.getAllValues()) { + Assert.assertEquals(REQUEST_ID, request.getRequestId()); + } + } + + /** + * A missing workflow ID is defaulted to a fresh UUID per call, so a retried Nexus task would + * start a second workflow rather than dedupe against the first. + */ + @Test + public void signalWithStartRequiresWorkflowId() { + enableSignalDeduplication(); + + HandlerException e = Assert.assertThrows(HandlerException.class, () -> signalWithStart(null)); + Assert.assertEquals(HandlerException.ErrorType.BAD_REQUEST, e.getErrorType()); + Assert.assertEquals( + "signalWithStart from a Nexus operation handler requires a workflow ID.", + e.getCause().getMessage()); + verify(genericClient, never()).signalWithStart(any()); + } +} diff --git a/temporal-sdk/src/test/java/io/temporal/workflow/nexus/GenericHandlerSignalDeduplicationTest.java b/temporal-sdk/src/test/java/io/temporal/workflow/nexus/GenericHandlerSignalDeduplicationTest.java new file mode 100644 index 0000000000..379fbea2a1 --- /dev/null +++ b/temporal-sdk/src/test/java/io/temporal/workflow/nexus/GenericHandlerSignalDeduplicationTest.java @@ -0,0 +1,218 @@ +package io.temporal.workflow.nexus; + +import static org.junit.Assume.assumeTrue; + +import io.nexusrpc.Operation; +import io.nexusrpc.Service; +import io.nexusrpc.handler.HandlerException; +import io.nexusrpc.handler.ServiceImpl; +import io.temporal.client.WorkflowClient; +import io.temporal.client.WorkflowOptions; +import io.temporal.client.WorkflowStub; +import io.temporal.nexus.Nexus; +import io.temporal.nexus.TemporalNexusClient; +import io.temporal.nexus.TemporalOperation; +import io.temporal.nexus.TemporalOperationResult; +import io.temporal.nexus.TemporalOperationStartContext; +import io.temporal.testing.CloudTestExclusion.RequiresCloudProvisioning; +import io.temporal.testing.CloudTestExclusionNote; +import io.temporal.testing.internal.SDKTestWorkflowRule; +import io.temporal.workflow.*; +import java.time.Duration; +import java.util.List; +import java.util.Map; +import java.util.UUID; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.CopyOnWriteArrayList; +import org.junit.Assert; +import org.junit.Rule; +import org.junit.Test; +import org.junit.experimental.categories.Category; + +/** + * Signals sent through {@link TemporalNexusClient#getWorkflowClient()} reuse the operation's + * request ID, so only one signal is allowed per operation invocation and a retried Nexus task does + * not deliver the signal twice. + */ +@CloudTestExclusionNote("Cloud CI does not provision the Nexus endpoint required by this test.") +@Category(RequiresCloudProvisioning.class) +public class GenericHandlerSignalDeduplicationTest { + private static final String ONE_SIGNAL_MESSAGE = + "Only one signal can be sent per Nexus operation invocation."; + private static final Map> retryRequestIds = new ConcurrentHashMap<>(); + + @Rule + public SDKTestWorkflowRule testWorkflowRule = + SDKTestWorkflowRule.newBuilder() + .setWorkflowTypes(CallerWorkflowImpl.class, CountSignalsWorkflowImpl.class) + .setNexusServiceImplementation(new TestNexusServiceImpl()) + .build(); + + @Test + public void secondSignalRejected() { + String targetId = "double-signal-" + UUID.randomUUID(); + CountSignalsWorkflow target = + testWorkflowRule + .getWorkflowClient() + .newWorkflowStub( + CountSignalsWorkflow.class, + WorkflowOptions.newBuilder() + .setWorkflowId(targetId) + .setTaskQueue(testWorkflowRule.getTaskQueue()) + .build()); + WorkflowClient.start(target::execute); + + Assert.assertEquals("signaled-" + targetId, newCaller().execute("doubleSignal", targetId)); + + target.done(); + Assert.assertEquals(1, WorkflowStub.fromTyped(target).getResult(Integer.class).intValue()); + } + + @Test + public void signalDedupedAcrossHandlerRetry() { + // The in-memory test server does not dedupe signals on request ID. + assumeTrue(SDKTestWorkflowRule.useExternalService); + String targetId = "signal-retry-" + UUID.randomUUID(); + + Assert.assertEquals("signaled-" + targetId, newCaller().execute("signalRetry", targetId)); + + List requestIds = retryRequestIds.get(targetId); + Assert.assertEquals( + "handler should fail once and then be redelivered once", 2, requestIds.size()); + Assert.assertEquals(requestIds.get(0), requestIds.get(1)); + + CountSignalsWorkflow target = + testWorkflowRule.getWorkflowClient().newWorkflowStub(CountSignalsWorkflow.class, targetId); + target.done(); + Assert.assertEquals( + "signal must be deduped on the operation's request ID across handler redelivery", + 1, + WorkflowStub.fromTyped(target).getResult(Integer.class).intValue()); + } + + private CallerWorkflow newCaller() { + return testWorkflowRule.newWorkflowStubTimeoutOptions(CallerWorkflow.class); + } + + @WorkflowInterface + public interface CallerWorkflow { + @WorkflowMethod + String execute(String operation, String targetId); + } + + public static class CallerWorkflowImpl implements CallerWorkflow { + @Override + public String execute(String operation, String targetId) { + TestNexusService service = + Workflow.newNexusServiceStub( + TestNexusService.class, + NexusServiceOptions.newBuilder() + .setOperationOptions( + NexusOperationOptions.newBuilder() + .setScheduleToCloseTimeout(Duration.ofSeconds(30)) + .build()) + .build()); + switch (operation) { + case "doubleSignal": + return service.doubleSignal(targetId); + case "signalRetry": + return service.signalRetry(targetId); + default: + throw new IllegalArgumentException("unknown operation " + operation); + } + } + } + + @WorkflowInterface + public interface CountSignalsWorkflow { + @WorkflowMethod + int execute(); + + @SignalMethod + void mySignal(String value); + + @SignalMethod + void done(); + } + + /** Returns how many {@code mySignal} deliveries reached the workflow before {@code done}. */ + public static class CountSignalsWorkflowImpl implements CountSignalsWorkflow { + private int count; + private boolean done; + + @Override + public int execute() { + Workflow.await(() -> done); + return count; + } + + @Override + public void mySignal(String value) { + count++; + } + + @Override + public void done() { + done = true; + } + } + + @Service + public interface TestNexusService { + @Operation + String doubleSignal(String targetId); + + @Operation + String signalRetry(String targetId); + } + + @ServiceImpl(service = TestNexusService.class) + public static class TestNexusServiceImpl { + @TemporalOperation + public TemporalOperationResult doubleSignal( + TemporalOperationStartContext context, TemporalNexusClient client, String targetId) { + client.getWorkflowClient().newUntypedWorkflowStub(targetId).signal("mySignal", targetId); + // The first signal uses up the invocation's only signal, so a second one is rejected even + // though it targets a different workflow. + try { + client + .getWorkflowClient() + .newUntypedWorkflowStub(targetId + "-other") + .signal("mySignal", targetId); + } catch (HandlerException e) { + if (e.getErrorType() == HandlerException.ErrorType.BAD_REQUEST + && ONE_SIGNAL_MESSAGE.equals(e.getCause().getMessage())) { + return TemporalOperationResult.sync("signaled-" + targetId); + } + throw e; + } + throw new IllegalStateException("expected second signal to be rejected"); + } + + @TemporalOperation + public TemporalOperationResult signalRetry( + TemporalOperationStartContext context, TemporalNexusClient client, String targetId) { + client + .getWorkflowClient() + .newUntypedWorkflowStub( + "CountSignalsWorkflow", + WorkflowOptions.newBuilder() + .setWorkflowId(targetId) + .setTaskQueue(Nexus.getOperationContext().getInfo().getTaskQueue()) + .build()) + .signalWithStart("mySignal", new Object[] {targetId}, new Object[] {}); + + List requestIds = + retryRequestIds.computeIfAbsent(targetId, k -> new CopyOnWriteArrayList<>()); + requestIds.add(context.getRequestId()); + if (requestIds.size() == 1) { + throw new HandlerException( + HandlerException.ErrorType.INTERNAL, + "force handler redelivery after signalling", + null, + HandlerException.RetryBehavior.RETRYABLE); + } + return TemporalOperationResult.sync("signaled-" + targetId); + } + } +}