Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 8 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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,
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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();
}
}
}

Expand Down Expand Up @@ -182,8 +194,13 @@ public <R> 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(
Expand All @@ -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);
Expand All @@ -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();
}
}
}

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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();

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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());
Expand All @@ -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();
}
Expand All @@ -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);
Expand All @@ -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.
Expand Down
Original file line number Diff line number Diff line change
@@ -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.
Expand All @@ -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");
Expand Down
Original file line number Diff line number Diff line change
@@ -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;
Expand All @@ -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;

Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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;
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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.
*
* <p>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();

/**
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -88,6 +88,7 @@ public final OperationStartResult<R> start(
OperationContext ctx, OperationStartDetails details, T input)
throws OperationException, HandlerException {
InternalNexusOperationContext nexusCtx = CurrentNexusOperationContext.get();
nexusCtx.enableSignalDeduplication();
TemporalNexusClient client =
new TemporalNexusClientImpl(nexusCtx.getWorkflowClient(), ctx, details);

Expand Down
Loading
Loading