From 16f16b3fe201af3ccb31012410b05d5b931ac7e0 Mon Sep 17 00:00:00 2001 From: Quinn Klassen Date: Sat, 16 May 2026 09:25:56 -0700 Subject: [PATCH 1/2] Add support activities to Nexus operations --- .../temporal/client/ActivityClientImpl.java | 8 +- .../ActivityClientCallsInterceptor.java | 66 ++++ .../client/ActivityClientInternal.java | 16 + .../client/NexusStartActivityRequest.java | 82 ++++ .../client/NexusStartActivityResponse.java | 35 ++ .../client/RootActivityClientInvoker.java | 63 +++- .../internal/common/InternalUtils.java | 74 ++-- .../nexus/NexusStartActivityHelper.java | 67 ++++ .../internal/nexus/OperationToken.java | 51 +++ .../internal/nexus/OperationTokenType.java | 4 +- .../internal/nexus/OperationTokenUtil.java | 71 +++- .../nexus/CancelActivityExecutionInput.java | 38 ++ .../temporal/nexus/TemporalNexusClient.java | 355 ++++++++++++++++++ .../nexus/TemporalNexusClientImpl.java | 305 ++++++++++++++- .../nexus/TemporalOperationHandler.java | 56 ++- .../internal/nexus/WorkflowRunTokenTest.java | 86 +++++ .../nexus/TemporalNexusClientImplTest.java | 248 ++++++++++++ .../nexus/AsyncActivityOperationTest.java | 217 +++++++++++ .../CancelActivityAsyncOperationTest.java | 264 +++++++++++++ 19 files changed, 2054 insertions(+), 52 deletions(-) create mode 100644 temporal-sdk/src/main/java/io/temporal/internal/client/ActivityClientInternal.java create mode 100644 temporal-sdk/src/main/java/io/temporal/internal/client/NexusStartActivityRequest.java create mode 100644 temporal-sdk/src/main/java/io/temporal/internal/client/NexusStartActivityResponse.java create mode 100644 temporal-sdk/src/main/java/io/temporal/internal/nexus/NexusStartActivityHelper.java create mode 100644 temporal-sdk/src/main/java/io/temporal/nexus/CancelActivityExecutionInput.java create mode 100644 temporal-sdk/src/test/java/io/temporal/nexus/TemporalNexusClientImplTest.java create mode 100644 temporal-sdk/src/test/java/io/temporal/workflow/nexus/AsyncActivityOperationTest.java create mode 100644 temporal-sdk/src/test/java/io/temporal/workflow/nexus/CancelActivityAsyncOperationTest.java diff --git a/temporal-sdk/src/main/java/io/temporal/client/ActivityClientImpl.java b/temporal-sdk/src/main/java/io/temporal/client/ActivityClientImpl.java index efcc9df698..77bf2bd794 100644 --- a/temporal-sdk/src/main/java/io/temporal/client/ActivityClientImpl.java +++ b/temporal-sdk/src/main/java/io/temporal/client/ActivityClientImpl.java @@ -6,6 +6,7 @@ import io.temporal.common.interceptors.ActivityClientCallsInterceptor; import io.temporal.common.interceptors.ActivityClientInterceptor; import io.temporal.common.interceptors.Header; +import io.temporal.internal.client.ActivityClientInternal; import io.temporal.internal.client.ActivityHandleImpl; import io.temporal.internal.client.RootActivityClientInvoker; import io.temporal.internal.client.external.GenericWorkflowClientImpl; @@ -27,7 +28,7 @@ * Implementation of {@link ActivityClient} that delegates calls through the activity interceptor * chain and ultimately to the Temporal service. */ -class ActivityClientImpl implements ActivityClient { +class ActivityClientImpl implements ActivityClient, ActivityClientInternal { private final WorkflowServiceStubs stubs; private final ActivityClientOptions options; @@ -56,6 +57,11 @@ private static ActivityClientCallsInterceptor initializeClientInvoker( return invoker; } + @Override + public ActivityClientCallsInterceptor getInvoker() { + return invoker; + } + // ---- Interface-based start (Proc variants) ---- @Override diff --git a/temporal-sdk/src/main/java/io/temporal/common/interceptors/ActivityClientCallsInterceptor.java b/temporal-sdk/src/main/java/io/temporal/common/interceptors/ActivityClientCallsInterceptor.java index 16a34dc285..5e278969d5 100644 --- a/temporal-sdk/src/main/java/io/temporal/common/interceptors/ActivityClientCallsInterceptor.java +++ b/temporal-sdk/src/main/java/io/temporal/common/interceptors/ActivityClientCallsInterceptor.java @@ -9,6 +9,8 @@ import io.temporal.common.Experimental; import java.lang.reflect.Type; import java.util.List; +import java.util.Map; +import java.util.Objects; import java.util.concurrent.CompletableFuture; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; @@ -116,19 +118,56 @@ GetActivityResultOutput getActivityResult(GetActivityResultInput input CompletableFuture> getActivityResultAsync( GetActivityResultInput input); + /** + * Nexus completion-callback metadata propagated through {@link StartActivityInput} when an + * activity is scheduled from a Nexus operation handler. + */ + @Experimental + final class CompletionCallback { + private final String url; + private final Map headers; + + public CompletionCallback(String url, Map headers) { + this.url = Objects.requireNonNull(url); + this.headers = Objects.requireNonNull(headers); + } + + public String getUrl() { + return url; + } + + public Map getHeaders() { + return headers; + } + } + @Experimental final class StartActivityInput { private final String activityType; private final List args; private final StartActivityOptions options; private final Header header; + private final @Nullable CompletionCallback completionCallback; + private final @Nullable List links; public StartActivityInput( String activityType, List args, StartActivityOptions options, Header header) { + this(activityType, args, options, header, null, null); + } + + public StartActivityInput( + String activityType, + List args, + StartActivityOptions options, + Header header, + @Nullable CompletionCallback completionCallback, + @Nullable List links) { this.activityType = activityType; this.args = args; this.options = options; this.header = header; + this.completionCallback = completionCallback; + this.links = links; } public String getActivityType() { @@ -146,6 +185,16 @@ public StartActivityOptions getOptions() { public Header getHeader() { return header; } + + @Nullable + public CompletionCallback getCompletionCallback() { + return completionCallback; + } + + @Nullable + public List getLinks() { + return links; + } } @Experimental @@ -153,9 +202,21 @@ final class StartActivityOutput { private final String activityId; private final @Nullable String activityRunId; + /** + * Set by the invoker when the start request included a Nexus completion callback; null + * otherwise. Internal use; do not depend on this in user-facing code. + */ + private final @Nullable String nexusOperationToken; + public StartActivityOutput(String activityId, @Nullable String activityRunId) { + this(activityId, activityRunId, null); + } + + public StartActivityOutput( + String activityId, @Nullable String activityRunId, @Nullable String nexusOperationToken) { this.activityId = activityId; this.activityRunId = activityRunId; + this.nexusOperationToken = nexusOperationToken; } public String getActivityId() { @@ -166,6 +227,11 @@ public String getActivityId() { public String getActivityRunId() { return activityRunId; } + + @Nullable + public String getNexusOperationToken() { + return nexusOperationToken; + } } @Experimental diff --git a/temporal-sdk/src/main/java/io/temporal/internal/client/ActivityClientInternal.java b/temporal-sdk/src/main/java/io/temporal/internal/client/ActivityClientInternal.java new file mode 100644 index 0000000000..1652da00d2 --- /dev/null +++ b/temporal-sdk/src/main/java/io/temporal/internal/client/ActivityClientInternal.java @@ -0,0 +1,16 @@ +package io.temporal.internal.client; + +import io.temporal.common.interceptors.ActivityClientCallsInterceptor; + +/** + * Internal-only view of an {@code ActivityClient} that exposes the configured interceptor chain. + * + *

Lives in {@code io.temporal.internal.client} so that other internal SDK packages (e.g. {@code + * io.temporal.nexus}) can route a fully-constructed {@link + * ActivityClientCallsInterceptor.StartActivityInput} through the chain without bypassing + * user-registered interceptors or the metrics-tagged scope, and without forcing the concrete impl + * class to be public. + */ +public interface ActivityClientInternal { + ActivityClientCallsInterceptor getInvoker(); +} diff --git a/temporal-sdk/src/main/java/io/temporal/internal/client/NexusStartActivityRequest.java b/temporal-sdk/src/main/java/io/temporal/internal/client/NexusStartActivityRequest.java new file mode 100644 index 0000000000..4de1626408 --- /dev/null +++ b/temporal-sdk/src/main/java/io/temporal/internal/client/NexusStartActivityRequest.java @@ -0,0 +1,82 @@ +package io.temporal.internal.client; + +import io.nexusrpc.Link; +import io.temporal.client.StartActivityOptions; +import io.temporal.common.Experimental; +import io.temporal.common.interceptors.Header; +import java.util.List; +import java.util.Map; + +/** + * Request used to start an activity from a Nexus operation handler. Mirrors {@link + * NexusStartWorkflowRequest} but carries the activity-specific scheduling payload. + */ +@Experimental +public final class NexusStartActivityRequest { + private final String requestId; + private final String callbackUrl; + private final Map callbackHeaders; + private final String taskQueue; + private final List links; + private final String activityType; + private final List args; + private final StartActivityOptions options; + private final Header header; + + public NexusStartActivityRequest( + String requestId, + String callbackUrl, + Map callbackHeaders, + String taskQueue, + List links, + String activityType, + List args, + StartActivityOptions options, + Header header) { + this.requestId = requestId; + this.callbackUrl = callbackUrl; + this.callbackHeaders = callbackHeaders; + this.taskQueue = taskQueue; + this.links = links; + this.activityType = activityType; + this.args = args; + this.options = options; + this.header = header; + } + + public String getRequestId() { + return requestId; + } + + public String getCallbackUrl() { + return callbackUrl; + } + + public Map getCallbackHeaders() { + return callbackHeaders; + } + + public String getTaskQueue() { + return taskQueue; + } + + public List getLinks() { + return links; + } + + public String getActivityType() { + return activityType; + } + + public List getArgs() { + return args; + } + + public StartActivityOptions getOptions() { + return options; + } + + public Header getHeader() { + return header; + } +} diff --git a/temporal-sdk/src/main/java/io/temporal/internal/client/NexusStartActivityResponse.java b/temporal-sdk/src/main/java/io/temporal/internal/client/NexusStartActivityResponse.java new file mode 100644 index 0000000000..6971ee550e --- /dev/null +++ b/temporal-sdk/src/main/java/io/temporal/internal/client/NexusStartActivityResponse.java @@ -0,0 +1,35 @@ +package io.temporal.internal.client; + +import io.temporal.common.Experimental; +import javax.annotation.Nullable; + +/** + * Response returned from starting an activity via {@link NexusStartActivityRequest}. Mirrors {@link + * NexusStartWorkflowResponse}. + */ +@Experimental +public final class NexusStartActivityResponse { + private final String activityId; + private final @Nullable String runId; + private final String operationToken; + + public NexusStartActivityResponse( + String activityId, @Nullable String runId, String operationToken) { + this.activityId = activityId; + this.runId = runId; + this.operationToken = operationToken; + } + + public String getActivityId() { + return activityId; + } + + @Nullable + public String getRunId() { + return runId; + } + + public String getOperationToken() { + return operationToken; + } +} 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 96a9e70f5a..f9678c1996 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 @@ -9,6 +9,8 @@ import io.grpc.StatusRuntimeException; import io.temporal.api.activity.v1.ActivityExecutionOutcome; import io.temporal.api.common.v1.ActivityType; +import io.temporal.api.common.v1.Callback; +import io.temporal.api.common.v1.Link; import io.temporal.api.common.v1.Payloads; import io.temporal.api.errordetails.v1.ActivityExecutionAlreadyStartedFailure; import io.temporal.api.sdk.v1.UserMetadata; @@ -19,6 +21,8 @@ import io.temporal.common.interceptors.ActivityClientCallsInterceptor; import io.temporal.internal.client.external.GenericWorkflowClient; import io.temporal.internal.common.HeaderUtils; +import io.temporal.internal.common.InternalUtils; +import io.temporal.internal.common.LinkConverter; import io.temporal.internal.common.ProtoConverters; import io.temporal.internal.common.ProtobufTimeUtils; import io.temporal.internal.common.SearchAttributesUtil; @@ -28,7 +32,10 @@ import java.util.concurrent.CompletableFuture; import java.util.concurrent.CompletionException; import java.util.concurrent.TimeoutException; +import java.util.stream.Collectors; import java.util.stream.StreamSupport; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; /** * Terminus of the activity interceptor chain. Implements all activity RPCs against the Temporal @@ -36,6 +43,8 @@ */ public class RootActivityClientInvoker implements ActivityClientCallsInterceptor { + private static final Logger log = LoggerFactory.getLogger(RootActivityClientInvoker.class); + private final GenericWorkflowClient genericClient; private final ActivityClientOptions clientOptions; @@ -104,6 +113,58 @@ public StartActivityOutput startActivity(StartActivityInput input) { io.temporal.api.common.v1.Header grpcHeader = HeaderUtils.toHeaderGrpc(input.getHeader(), null); request.setHeader(grpcHeader); + // Hoisted so it can be returned in StartActivityOutput when a Nexus callback is present. + String nexusOperationToken = null; + if (input.getCompletionCallback() != null) { + List protoLinks = null; + if (input.getLinks() != null) { + protoLinks = + input.getLinks().stream() + .map( + link -> { + if (io.temporal.api.common.v1.Link.WorkflowEvent.getDescriptor() + .getFullName() + .equals(link.getType())) { + io.temporal.api.nexus.v1.Link nexusLink = + io.temporal.api.nexus.v1.Link.newBuilder() + .setType(link.getType()) + .setUrl(link.getUri().toString()) + .build(); + return LinkConverter.nexusLinkToWorkflowEvent(nexusLink); + } else { + log.warn("ignoring unsupported link data type: {}", link.getType()); + return null; + } + }) + .filter(Objects::nonNull) + .collect(Collectors.toList()); + if (!protoLinks.isEmpty()) { + request.addAllLinks(protoLinks); + } else { + protoLinks = null; + } + } + // Generate the operation token from the user-supplied activity ID and namespace so the + // dual OPERATION_ID + OPERATION_TOKEN headers can be injected before the start RPC fires. + try { + nexusOperationToken = + io.temporal.internal.nexus.OperationTokenUtil.generateActivityExecutionOperationToken( + options.getId(), clientOptions.getNamespace()); + } catch (com.fasterxml.jackson.core.JsonProcessingException e) { + throw new io.nexusrpc.handler.HandlerException( + io.nexusrpc.handler.HandlerException.ErrorType.BAD_REQUEST, + "failed to generate activity operation token", + e); + } + Callback cb = + InternalUtils.buildNexusCallback( + input.getCompletionCallback().getUrl(), + input.getCompletionCallback().getHeaders(), + nexusOperationToken, + protoLinks); + request.addCompletionCallbacks(cb); + } + StartActivityExecutionResponse response; try { response = genericClient.startActivity(request.build()); @@ -121,7 +182,7 @@ public StartActivityOutput startActivity(StartActivityInput input) { } String runId = response.getRunId().isEmpty() ? null : response.getRunId(); - return new StartActivityOutput(options.getId(), runId); + return new StartActivityOutput(options.getId(), runId, nexusOperationToken); } @Override diff --git a/temporal-sdk/src/main/java/io/temporal/internal/common/InternalUtils.java b/temporal-sdk/src/main/java/io/temporal/internal/common/InternalUtils.java index 72987e39c2..08679ca103 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/common/InternalUtils.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/common/InternalUtils.java @@ -104,35 +104,10 @@ public static NexusWorkflowStarter createNexusBoundStub( // If a callback URL is provided, pass it as a completion callback. if (!Strings.isNullOrEmpty(request.getCallbackUrl())) { - // Add the Nexus operation ID to the headers if it is not already present to support - // fabricating - // a NexusOperationStarted event if the completion is received before the response to a - // StartOperation request. - Map headers = - request.getCallbackHeaders().entrySet().stream() - .collect( - Collectors.toMap( - (k) -> k.getKey().toLowerCase(), - Map.Entry::getValue, - (a, b) -> a, - () -> new TreeMap<>(String.CASE_INSENSITIVE_ORDER))); - if (!headers.containsKey(Header.OPERATION_ID)) { - headers.put(Header.OPERATION_ID.toLowerCase(), operationToken); - } - if (!headers.containsKey(Header.OPERATION_TOKEN)) { - headers.put(Header.OPERATION_TOKEN.toLowerCase(), operationToken); - } - Callback.Builder cbBuilder = - Callback.newBuilder() - .setNexus( - Callback.Nexus.newBuilder() - .setUrl(request.getCallbackUrl()) - .putAllHeader(headers) - .build()); - if (links != null) { - cbBuilder.addAllLinks(links); - } - nexusWorkflowOptions.setCompletionCallbacks(Collections.singletonList(cbBuilder.build())); + Callback cb = + buildNexusCallback( + request.getCallbackUrl(), request.getCallbackHeaders(), operationToken, links); + nexusWorkflowOptions.setCompletionCallbacks(Collections.singletonList(cb)); } if (options.getTaskQueue() == null) { @@ -157,6 +132,47 @@ public static boolean isWorkflowStreamReservedName(String name) { return name.startsWith(WORKFLOW_STREAM_RESERVED_PREFIX); } + /** + * Builds a {@link Callback} for use as a Nexus completion callback. Injects both the legacy + * {@code Nexus-Operation-Id} and the newer {@code Nexus-Operation-Token} headers + * (case-insensitive lookup) when not already present so the server can fabricate + * operation-started events if the completion is received before the response to a StartOperation + * request. + * + *

Shared by the workflow start path ({@link #createNexusBoundStub}) and the activity start + * path ({@code RootActivityClientInvoker.startActivity}). The dual {@code OPERATION_ID} + {@code + * OPERATION_TOKEN} headers must be injected before the start RPC is issued. + */ + @SuppressWarnings("deprecation") // Check the OPERATION_ID header for backwards compatibility + public static Callback buildNexusCallback( + String callbackUrl, + Map callbackHeaders, + String operationToken, + List protoLinks) { + Map headers = + callbackHeaders.entrySet().stream() + .collect( + Collectors.toMap( + (k) -> k.getKey().toLowerCase(), + Map.Entry::getValue, + (a, b) -> a, + () -> new TreeMap<>(String.CASE_INSENSITIVE_ORDER))); + if (!headers.containsKey(Header.OPERATION_ID)) { + headers.put(Header.OPERATION_ID.toLowerCase(), operationToken); + } + if (!headers.containsKey(Header.OPERATION_TOKEN)) { + headers.put(Header.OPERATION_TOKEN.toLowerCase(), operationToken); + } + Callback.Builder cbBuilder = + Callback.newBuilder() + .setNexus( + Callback.Nexus.newBuilder().setUrl(callbackUrl).putAllHeader(headers).build()); + if (protoLinks != null) { + cbBuilder.addAllLinks(protoLinks); + } + return cbBuilder.build(); + } + /** Check the method name for reserved prefixes or names. */ public static void checkMethodName(POJOWorkflowMethodMetadata methodMetadata) { boolean workflowStreamExempt = diff --git a/temporal-sdk/src/main/java/io/temporal/internal/nexus/NexusStartActivityHelper.java b/temporal-sdk/src/main/java/io/temporal/internal/nexus/NexusStartActivityHelper.java new file mode 100644 index 0000000000..3e6fd2cea6 --- /dev/null +++ b/temporal-sdk/src/main/java/io/temporal/internal/nexus/NexusStartActivityHelper.java @@ -0,0 +1,67 @@ +package io.temporal.internal.nexus; + +import io.nexusrpc.handler.OperationContext; +import io.nexusrpc.handler.OperationStartDetails; +import io.temporal.client.StartActivityOptions; +import io.temporal.common.Experimental; +import io.temporal.common.interceptors.Header; +import io.temporal.internal.client.NexusStartActivityRequest; +import io.temporal.internal.client.NexusStartActivityResponse; +import java.util.List; +import java.util.function.Function; + +/** + * Shared helper for starting an activity from a Nexus operation and (in future) attaching links to + * the operation context. Mirrors {@link NexusStartWorkflowHelper} for activities. + */ +@Experimental +public class NexusStartActivityHelper { + + /** + * Starts an activity via the provided invoker function and returns the response. + * + *

The link-attachment block is intentionally a no-op in this revision; see the TODO inside. + * + * @param ctx the operation context (link attachment is deferred — see TODO) + * @param details the operation start details containing requestId, callback, links + * @param activityType the activity type name + * @param args the activity arguments + * @param options the activity scheduling options (must include task queue, ID) + * @param header the propagated header + * @param invoker function that starts the activity given a {@link NexusStartActivityRequest} + * @return the {@link NexusStartActivityResponse} containing the activity ID and operation token + */ + public static NexusStartActivityResponse startActivityAndAttachLinks( + OperationContext ctx, + OperationStartDetails details, + String activityType, + List args, + StartActivityOptions options, + Header header, + Function invoker) { + InternalNexusOperationContext nexusCtx = CurrentNexusOperationContext.get(); + + NexusStartActivityRequest nexusRequest = + new NexusStartActivityRequest( + details.getRequestId(), + details.getCallbackUrl(), + details.getCallbackHeaders(), + nexusCtx.getTaskQueue(), + details.getLinks(), + activityType, + args, + options, + header); + + NexusStartActivityResponse response = invoker.apply(nexusRequest); + + // TODO: Attach activity-event link when server-side activity link support is available. + // No StartActivityResponseLink analog exists and there is no verified activity-event link path + // in LinkConverter. Do NOT copy the synthetic workflow-event link fabrication from + // NexusStartWorkflowHelper; Nexus operations function correctly without diagnostic links. + + return response; + } + + private NexusStartActivityHelper() {} +} diff --git a/temporal-sdk/src/main/java/io/temporal/internal/nexus/OperationToken.java b/temporal-sdk/src/main/java/io/temporal/internal/nexus/OperationToken.java index 4bd5635e93..a5f7ec9972 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/nexus/OperationToken.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/nexus/OperationToken.java @@ -1,5 +1,6 @@ package io.temporal.internal.nexus; +import com.fasterxml.jackson.annotation.JsonCreator; import com.fasterxml.jackson.annotation.JsonInclude; import com.fasterxml.jackson.annotation.JsonProperty; @@ -16,16 +17,30 @@ public class OperationToken { private final String namespace; @JsonProperty("wid") + @JsonInclude(JsonInclude.Include.NON_NULL) private final String workflowId; + @JsonProperty("aid") + @JsonInclude(JsonInclude.Include.NON_NULL) + private final String activityId; + + @JsonProperty("rid") + @JsonInclude(JsonInclude.Include.NON_NULL) + private final String runId; + + @JsonCreator public OperationToken( @JsonProperty("t") Integer type, @JsonProperty("ns") String namespace, @JsonProperty("wid") String workflowId, + @JsonProperty("aid") String activityId, + @JsonProperty("rid") String runId, @JsonProperty("v") Integer version) { this.type = OperationTokenType.fromValue(type); this.namespace = namespace; this.workflowId = workflowId; + this.activityId = activityId; + this.runId = runId; this.version = version; } @@ -33,6 +48,27 @@ public OperationToken(OperationTokenType type, String namespace, String workflow this.type = type; this.namespace = namespace; this.workflowId = workflowId; + this.activityId = null; + this.runId = null; + this.version = null; + } + + public OperationToken( + OperationTokenType type, String namespace, String workflowId, String activityId) { + this(type, namespace, workflowId, activityId, null); + } + + public OperationToken( + OperationTokenType type, + String namespace, + String workflowId, + String activityId, + String runId) { + this.type = type; + this.namespace = namespace; + this.workflowId = workflowId; + this.activityId = activityId; + this.runId = runId; this.version = null; } @@ -51,4 +87,19 @@ public String getNamespace() { public String getWorkflowId() { return workflowId; } + + public String getActivityId() { + return activityId; + } + + /** + * Returns the activity run ID embedded in the token, or {@code null} if absent. + * + *

Run ID is only present on activity-execution tokens that were generated AFTER the start + * activity RPC completed (so the run ID was known). Tokens written into the Nexus operation-token + * callback header are generated before that point and therefore do not carry a run ID. + */ + public String getRunId() { + return runId; + } } diff --git a/temporal-sdk/src/main/java/io/temporal/internal/nexus/OperationTokenType.java b/temporal-sdk/src/main/java/io/temporal/internal/nexus/OperationTokenType.java index 11aa57a81e..952d784bef 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/nexus/OperationTokenType.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/nexus/OperationTokenType.java @@ -5,7 +5,9 @@ public enum OperationTokenType { UNKNOWN(0), - WORKFLOW_RUN(1); + WORKFLOW_RUN(1), + // Values 2 and 3 are reserved for future token types (e.g. update-workflow, get-workflow-result). + ACTIVITY_EXECUTION(4); private final int value; diff --git a/temporal-sdk/src/main/java/io/temporal/internal/nexus/OperationTokenUtil.java b/temporal-sdk/src/main/java/io/temporal/internal/nexus/OperationTokenUtil.java index 737a84aad4..447eb04f19 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/nexus/OperationTokenUtil.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/nexus/OperationTokenUtil.java @@ -31,8 +31,20 @@ public static OperationToken loadOperationToken(String operationToken) { if (token.getVersion() != null && token.getVersion() != 0) { throw new IllegalArgumentException("Invalid operation token: unexpected version field"); } - if (Strings.isNullOrEmpty(token.getWorkflowId())) { - throw new IllegalArgumentException("Invalid operation token: missing workflow ID (wid)"); + switch (token.getType()) { + case WORKFLOW_RUN: + if (Strings.isNullOrEmpty(token.getWorkflowId())) { + throw new IllegalArgumentException("Invalid operation token: missing workflow ID (wid)"); + } + break; + case ACTIVITY_EXECUTION: + if (Strings.isNullOrEmpty(token.getActivityId())) { + throw new IllegalArgumentException("Invalid operation token: missing activity ID (aid)"); + } + break; + default: + throw new IllegalArgumentException( + "Invalid operation token: unknown operation token type: " + token.getType()); } return token; } @@ -61,6 +73,31 @@ public static String loadWorkflowIdFromOperationToken(String operationToken) { return loadWorkflowRunOperationToken(operationToken).getWorkflowId(); } + /** + * Load an activity execution operation token, asserting that the token type is {@link + * OperationTokenType#ACTIVITY_EXECUTION}. + * + * @throws IllegalArgumentException if the operation token is invalid or not an activity execution + * token + */ + public static OperationToken loadActivityExecutionOperationToken(String operationToken) { + OperationToken token = loadOperationToken(operationToken); + if (!token.getType().equals(OperationTokenType.ACTIVITY_EXECUTION)) { + throw new IllegalArgumentException( + "Invalid activity execution token: incorrect operation token type: " + token.getType()); + } + return token; + } + + /** + * Extract the activity ID from an activity execution operation token. + * + * @throws IllegalArgumentException if the operation token is invalid + */ + public static String loadActivityIdFromOperationToken(String operationToken) { + return loadActivityExecutionOperationToken(operationToken).getActivityId(); + } + /** Generate a workflow run operation token from a workflow ID and namespace. */ public static String generateWorkflowRunOperationToken(String workflowId, String namespace) throws JsonProcessingException { @@ -70,5 +107,35 @@ public static String generateWorkflowRunOperationToken(String workflowId, String return encoder.encodeToString(json.getBytes()); } + /** + * Generate an activity execution operation token from an activity ID and namespace. + * + *

This overload omits the run ID. Use it when writing the token into the Nexus operation-token + * callback header — that token is generated before the start RPC completes, so the run ID is not + * yet known. + */ + public static String generateActivityExecutionOperationToken(String activityId, String namespace) + throws JsonProcessingException { + return generateActivityExecutionOperationToken(activityId, null, namespace); + } + + /** + * Generate an activity execution operation token from an activity ID, run ID, and namespace. The + * {@code runId} is included only when non-null. + * + *

This overload is used for the operation token returned to the Nexus caller from a start + * operation — at that point the start RPC has completed and the run ID is known. The header token + * written into the activity completion callback must NOT carry a run ID; use {@link + * #generateActivityExecutionOperationToken(String, String)} for that path. + */ + public static String generateActivityExecutionOperationToken( + String activityId, String runId, String namespace) throws JsonProcessingException { + String json = + ow.writeValueAsString( + new OperationToken( + OperationTokenType.ACTIVITY_EXECUTION, namespace, null, activityId, runId)); + return encoder.encodeToString(json.getBytes()); + } + private OperationTokenUtil() {} } diff --git a/temporal-sdk/src/main/java/io/temporal/nexus/CancelActivityExecutionInput.java b/temporal-sdk/src/main/java/io/temporal/nexus/CancelActivityExecutionInput.java new file mode 100644 index 0000000000..6794033817 --- /dev/null +++ b/temporal-sdk/src/main/java/io/temporal/nexus/CancelActivityExecutionInput.java @@ -0,0 +1,38 @@ +package io.temporal.nexus; + +import io.temporal.common.Experimental; +import java.util.Objects; +import javax.annotation.Nullable; + +/** + * Input to {@link TemporalOperationHandler#cancelActivityExecution} describing the activity + * execution to cancel. + */ +@Experimental +public final class CancelActivityExecutionInput { + + private final String activityId; + private final @Nullable String runId; + + public CancelActivityExecutionInput(String activityId, @Nullable String runId) { + this.activityId = Objects.requireNonNull(activityId); + this.runId = runId; + } + + /** Returns the activity ID extracted from the operation token. */ + public String getActivityId() { + return activityId; + } + + /** + * Returns the activity run ID extracted from the operation token, or {@code null} if absent. + * + *

Run ID is only present on operation tokens that were generated by this SDK AFTER the start + * activity RPC completed. Tokens originating from the activity completion callback header do not + * carry a run ID. + */ + @Nullable + public String getRunId() { + return runId; + } +} 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 4eed1fe350..58bf25bb66 100644 --- a/temporal-sdk/src/main/java/io/temporal/nexus/TemporalNexusClient.java +++ b/temporal-sdk/src/main/java/io/temporal/nexus/TemporalNexusClient.java @@ -1,5 +1,6 @@ package io.temporal.nexus; +import io.temporal.client.StartActivityOptions; import io.temporal.client.WorkflowClient; import io.temporal.client.WorkflowOptions; import io.temporal.common.Experimental; @@ -507,4 +508,358 @@ TemporalOperationResult startWorkflow( Type resultType, WorkflowOptions options, Object... args); + + // ---------- Activity overloads ---------- + + /** + * Starts a zero-argument activity that returns a value. + * + *

Example: + * + * 

{@code
+   * client.startActivity(MyActivity.class, MyActivity::run, options)
+   * }
+ * + * @param activityInterface the activity interface class + * @param activityMethod unbound method reference to the activity method + * @param options activity start options (must include taskQueue) + * @param the activity interface type + * @param the activity return type + * @return an async {@link TemporalOperationResult} with the activity-execution operation token + */ + TemporalOperationResult startActivity( + Class activityInterface, + Functions.Func1 activityMethod, + StartActivityOptions options); + + /** + * Starts a one-argument activity that returns a value. + * + * @param activityInterface the activity interface class + * @param activityMethod unbound method reference to the activity method + * @param arg1 first activity argument + * @param options activity start options (must include taskQueue) + * @param the activity interface type + * @param the type of the first activity argument + * @param the activity return type + * @return an async {@link TemporalOperationResult} with the activity-execution operation token + */ + TemporalOperationResult startActivity( + Class activityInterface, + Functions.Func2 activityMethod, + A1 arg1, + StartActivityOptions options); + + /** + * Starts a two-argument activity that returns a value. + * + * @param activityInterface the activity interface class + * @param activityMethod unbound method reference to the activity method + * @param arg1 first activity argument + * @param arg2 second activity argument + * @param options activity start options (must include taskQueue) + * @param the activity interface type + * @param the type of the first activity argument + * @param the type of the second activity argument + * @param the activity return type + * @return an async {@link TemporalOperationResult} with the activity-execution operation token + */ + TemporalOperationResult startActivity( + Class activityInterface, + Functions.Func3 activityMethod, + A1 arg1, + A2 arg2, + StartActivityOptions options); + + /** + * Starts a three-argument activity that returns a value. + * + * @param activityInterface the activity interface class + * @param activityMethod unbound method reference to the activity method + * @param arg1 first activity argument + * @param arg2 second activity argument + * @param arg3 third activity argument + * @param options activity start options (must include taskQueue) + * @param the activity interface type + * @param the type of the first activity argument + * @param the type of the second activity argument + * @param the type of the third activity argument + * @param the activity return type + * @return an async {@link TemporalOperationResult} with the activity-execution operation token + */ + TemporalOperationResult startActivity( + Class activityInterface, + Functions.Func4 activityMethod, + A1 arg1, + A2 arg2, + A3 arg3, + StartActivityOptions options); + + /** + * Starts a four-argument activity that returns a value. + * + * @param activityInterface the activity interface class + * @param activityMethod unbound method reference to the activity method + * @param arg1 first activity argument + * @param arg2 second activity argument + * @param arg3 third activity argument + * @param arg4 fourth activity argument + * @param options activity start options (must include taskQueue) + * @param the activity interface type + * @param the type of the first activity argument + * @param the type of the second activity argument + * @param the type of the third activity argument + * @param the type of the fourth activity argument + * @param the activity return type + * @return an async {@link TemporalOperationResult} with the activity-execution operation token + */ + TemporalOperationResult startActivity( + Class activityInterface, + Functions.Func5 activityMethod, + A1 arg1, + A2 arg2, + A3 arg3, + A4 arg4, + StartActivityOptions options); + + /** + * Starts a five-argument activity that returns a value. + * + * @param activityInterface the activity interface class + * @param activityMethod unbound method reference to the activity method + * @param arg1 first activity argument + * @param arg2 second activity argument + * @param arg3 third activity argument + * @param arg4 fourth activity argument + * @param arg5 fifth activity argument + * @param options activity start options (must include taskQueue) + * @param the activity interface type + * @param the type of the first activity argument + * @param the type of the second activity argument + * @param the type of the third activity argument + * @param the type of the fourth activity argument + * @param the type of the fifth activity argument + * @param the activity return type + * @return an async {@link TemporalOperationResult} with the activity-execution operation token + */ + TemporalOperationResult startActivity( + Class activityInterface, + Functions.Func6 activityMethod, + A1 arg1, + A2 arg2, + A3 arg3, + A4 arg4, + A5 arg5, + StartActivityOptions options); + + /** + * Starts a six-argument activity that returns a value. + * + * @param activityInterface the activity interface class + * @param activityMethod unbound method reference to the activity method + * @param arg1 first activity argument + * @param arg2 second activity argument + * @param arg3 third activity argument + * @param arg4 fourth activity argument + * @param arg5 fifth activity argument + * @param arg6 sixth activity argument + * @param options activity start options (must include taskQueue) + * @param the activity interface type + * @param the type of the first activity argument + * @param the type of the second activity argument + * @param the type of the third activity argument + * @param the type of the fourth activity argument + * @param the type of the fifth activity argument + * @param the type of the sixth activity argument + * @param the activity return type + * @return an async {@link TemporalOperationResult} with the activity-execution operation token + */ + TemporalOperationResult startActivity( + Class activityInterface, + Functions.Func7 activityMethod, + A1 arg1, + A2 arg2, + A3 arg3, + A4 arg4, + A5 arg5, + A6 arg6, + StartActivityOptions options); + + /** + * Starts a zero-argument activity with no return value. + * + * @param activityInterface the activity interface class + * @param activityMethod unbound method reference to the activity method + * @param options activity start options (must include taskQueue) + * @param the activity interface type + * @return an async {@link TemporalOperationResult} with the activity-execution operation token + */ + TemporalOperationResult startActivity( + Class activityInterface, Functions.Proc1 activityMethod, StartActivityOptions options); + + /** + * Starts a one-argument activity with no return value. + * + * @param activityInterface the activity interface class + * @param activityMethod unbound method reference to the activity method + * @param arg1 first activity argument + * @param options activity start options (must include taskQueue) + * @param the activity interface type + * @param the type of the first activity argument + * @return an async {@link TemporalOperationResult} with the activity-execution operation token + */ + TemporalOperationResult startActivity( + Class activityInterface, + Functions.Proc2 activityMethod, + A1 arg1, + StartActivityOptions options); + + /** + * Starts a two-argument activity with no return value. + * + * @param activityInterface the activity interface class + * @param activityMethod unbound method reference to the activity method + * @param arg1 first activity argument + * @param arg2 second activity argument + * @param options activity start options (must include taskQueue) + * @param the activity interface type + * @param the type of the first activity argument + * @param the type of the second activity argument + * @return an async {@link TemporalOperationResult} with the activity-execution operation token + */ + TemporalOperationResult startActivity( + Class activityInterface, + Functions.Proc3 activityMethod, + A1 arg1, + A2 arg2, + StartActivityOptions options); + + /** + * Starts a three-argument activity with no return value. + * + * @param activityInterface the activity interface class + * @param activityMethod unbound method reference to the activity method + * @param arg1 first activity argument + * @param arg2 second activity argument + * @param arg3 third activity argument + * @param options activity start options (must include taskQueue) + * @param the activity interface type + * @param the type of the first activity argument + * @param the type of the second activity argument + * @param the type of the third activity argument + * @return an async {@link TemporalOperationResult} with the activity-execution operation token + */ + TemporalOperationResult startActivity( + Class activityInterface, + Functions.Proc4 activityMethod, + A1 arg1, + A2 arg2, + A3 arg3, + StartActivityOptions options); + + /** + * Starts a four-argument activity with no return value. + * + * @param activityInterface the activity interface class + * @param activityMethod unbound method reference to the activity method + * @param arg1 first activity argument + * @param arg2 second activity argument + * @param arg3 third activity argument + * @param arg4 fourth activity argument + * @param options activity start options (must include taskQueue) + * @param the activity interface type + * @param the type of the first activity argument + * @param the type of the second activity argument + * @param the type of the third activity argument + * @param the type of the fourth activity argument + * @return an async {@link TemporalOperationResult} with the activity-execution operation token + */ + TemporalOperationResult startActivity( + Class activityInterface, + Functions.Proc5 activityMethod, + A1 arg1, + A2 arg2, + A3 arg3, + A4 arg4, + StartActivityOptions options); + + /** + * Starts a five-argument activity with no return value. + * + * @param activityInterface the activity interface class + * @param activityMethod unbound method reference to the activity method + * @param arg1 first activity argument + * @param arg2 second activity argument + * @param arg3 third activity argument + * @param arg4 fourth activity argument + * @param arg5 fifth activity argument + * @param options activity start options (must include taskQueue) + * @param the activity interface type + * @param the type of the first activity argument + * @param the type of the second activity argument + * @param the type of the third activity argument + * @param the type of the fourth activity argument + * @param the type of the fifth activity argument + * @return an async {@link TemporalOperationResult} with the activity-execution operation token + */ + TemporalOperationResult startActivity( + Class activityInterface, + Functions.Proc6 activityMethod, + A1 arg1, + A2 arg2, + A3 arg3, + A4 arg4, + A5 arg5, + StartActivityOptions options); + + /** + * Starts a six-argument activity with no return value. + * + * @param activityInterface the activity interface class + * @param activityMethod unbound method reference to the activity method + * @param arg1 first activity argument + * @param arg2 second activity argument + * @param arg3 third activity argument + * @param arg4 fourth activity argument + * @param arg5 fifth activity argument + * @param arg6 sixth activity argument + * @param options activity start options (must include taskQueue) + * @param the activity interface type + * @param the type of the first activity argument + * @param the type of the second activity argument + * @param the type of the third activity argument + * @param the type of the fourth activity argument + * @param the type of the fifth activity argument + * @param the type of the sixth activity argument + * @return an async {@link TemporalOperationResult} with the activity-execution operation token + */ + TemporalOperationResult startActivity( + Class activityInterface, + Functions.Proc7 activityMethod, + A1 arg1, + A2 arg2, + A3 arg3, + A4 arg4, + A5 arg5, + A6 arg6, + StartActivityOptions options); + + /** + * Starts an activity using an untyped activity type name. + * + *

Example: + * + * 

{@code
+   * client.startActivity("MyActivity", String.class, options, input)
+   * }
+ * + * @param activityType the activity type name string + * @param resultClass the expected result class + * @param options activity start options (must include taskQueue) + * @param args activity arguments + * @param the activity return type + * @return an async {@link TemporalOperationResult} with the activity-execution operation token + */ + TemporalOperationResult startActivity( + String activityType, Class resultClass, StartActivityOptions options, Object... args); } diff --git a/temporal-sdk/src/main/java/io/temporal/nexus/TemporalNexusClientImpl.java b/temporal-sdk/src/main/java/io/temporal/nexus/TemporalNexusClientImpl.java index 43e29e18f5..d63b0dd434 100644 --- a/temporal-sdk/src/main/java/io/temporal/nexus/TemporalNexusClientImpl.java +++ b/temporal-sdk/src/main/java/io/temporal/nexus/TemporalNexusClientImpl.java @@ -1,16 +1,31 @@ package io.temporal.nexus; +import com.fasterxml.jackson.core.JsonProcessingException; import io.nexusrpc.handler.HandlerException; import io.nexusrpc.handler.OperationContext; import io.nexusrpc.handler.OperationStartDetails; +import io.temporal.client.ActivityClient; +import io.temporal.client.ActivityClientOptions; +import io.temporal.client.StartActivityOptions; import io.temporal.client.WorkflowClient; import io.temporal.client.WorkflowOptions; import io.temporal.client.WorkflowStub; import io.temporal.common.Experimental; +import io.temporal.common.interceptors.ActivityClientCallsInterceptor; +import io.temporal.common.interceptors.Header; +import io.temporal.internal.client.ActivityClientInternal; +import io.temporal.internal.client.NexusStartActivityResponse; import io.temporal.internal.client.NexusStartWorkflowResponse; +import io.temporal.internal.nexus.NexusStartActivityHelper; import io.temporal.internal.nexus.NexusStartWorkflowHelper; +import io.temporal.internal.nexus.OperationTokenUtil; +import io.temporal.internal.util.MethodExtractor; import io.temporal.workflow.Functions; +import java.lang.reflect.Method; import java.lang.reflect.Type; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; import java.util.Objects; import java.util.concurrent.atomic.AtomicBoolean; @@ -244,13 +259,7 @@ public TemporalOperationResult startWorkflow( } private TemporalOperationResult invokeAndReturn(WorkflowHandle handle) { - if (!asyncOperationStarted.compareAndSet(false, true)) { - throw new HandlerException( - HandlerException.ErrorType.BAD_REQUEST, - new IllegalStateException( - "Only one async operation can be started per operation handler invocation. " - + "Use getWorkflowClient() for additional workflow interactions.")); - } + markAsyncOperationStarted(); try { NexusStartWorkflowResponse response = NexusStartWorkflowHelper.startWorkflowAndAttachLinks( @@ -265,4 +274,286 @@ private TemporalOperationResult invokeAndReturn(WorkflowHandle handle) throw t; } } + + // ---------- Activity overloads (Func returning) ---------- + + @Override + public TemporalOperationResult startActivity( + Class activityInterface, + Functions.Func1 activityMethod, + StartActivityOptions options) { + Method method = MethodExtractor.extract(activityInterface, activityMethod); + String activityType = MethodExtractor.activityTypeName(activityInterface, method); + return startActivityImpl(activityType, Collections.emptyList(), options); + } + + @Override + public TemporalOperationResult startActivity( + Class activityInterface, + Functions.Func2 activityMethod, + A1 arg1, + StartActivityOptions options) { + Method method = MethodExtractor.extract(activityInterface, activityMethod); + String activityType = MethodExtractor.activityTypeName(activityInterface, method); + return startActivityImpl(activityType, Collections.singletonList(arg1), options); + } + + @Override + public TemporalOperationResult startActivity( + Class activityInterface, + Functions.Func3 activityMethod, + A1 arg1, + A2 arg2, + StartActivityOptions options) { + Method method = MethodExtractor.extract(activityInterface, activityMethod); + String activityType = MethodExtractor.activityTypeName(activityInterface, method); + return startActivityImpl(activityType, Arrays.asList(arg1, arg2), options); + } + + @Override + public TemporalOperationResult startActivity( + Class activityInterface, + Functions.Func4 activityMethod, + A1 arg1, + A2 arg2, + A3 arg3, + StartActivityOptions options) { + Method method = MethodExtractor.extract(activityInterface, activityMethod); + String activityType = MethodExtractor.activityTypeName(activityInterface, method); + return startActivityImpl(activityType, Arrays.asList(arg1, arg2, arg3), options); + } + + @Override + public TemporalOperationResult startActivity( + Class activityInterface, + Functions.Func5 activityMethod, + A1 arg1, + A2 arg2, + A3 arg3, + A4 arg4, + StartActivityOptions options) { + Method method = MethodExtractor.extract(activityInterface, activityMethod); + String activityType = MethodExtractor.activityTypeName(activityInterface, method); + return startActivityImpl(activityType, Arrays.asList(arg1, arg2, arg3, arg4), options); + } + + @Override + public TemporalOperationResult startActivity( + Class activityInterface, + Functions.Func6 activityMethod, + A1 arg1, + A2 arg2, + A3 arg3, + A4 arg4, + A5 arg5, + StartActivityOptions options) { + Method method = MethodExtractor.extract(activityInterface, activityMethod); + String activityType = MethodExtractor.activityTypeName(activityInterface, method); + return startActivityImpl(activityType, Arrays.asList(arg1, arg2, arg3, arg4, arg5), options); + } + + @Override + public TemporalOperationResult startActivity( + Class activityInterface, + Functions.Func7 activityMethod, + A1 arg1, + A2 arg2, + A3 arg3, + A4 arg4, + A5 arg5, + A6 arg6, + StartActivityOptions options) { + Method method = MethodExtractor.extract(activityInterface, activityMethod); + String activityType = MethodExtractor.activityTypeName(activityInterface, method); + return startActivityImpl( + activityType, Arrays.asList(arg1, arg2, arg3, arg4, arg5, arg6), options); + } + + // ---------- Activity overloads (Proc void) ---------- + + @Override + public TemporalOperationResult startActivity( + Class activityInterface, Functions.Proc1 activityMethod, StartActivityOptions options) { + Method method = MethodExtractor.extract(activityInterface, activityMethod); + String activityType = MethodExtractor.activityTypeName(activityInterface, method); + return startActivityImpl(activityType, Collections.emptyList(), options); + } + + @Override + public TemporalOperationResult startActivity( + Class activityInterface, + Functions.Proc2 activityMethod, + A1 arg1, + StartActivityOptions options) { + Method method = MethodExtractor.extract(activityInterface, activityMethod); + String activityType = MethodExtractor.activityTypeName(activityInterface, method); + return startActivityImpl(activityType, Collections.singletonList(arg1), options); + } + + @Override + public TemporalOperationResult startActivity( + Class activityInterface, + Functions.Proc3 activityMethod, + A1 arg1, + A2 arg2, + StartActivityOptions options) { + Method method = MethodExtractor.extract(activityInterface, activityMethod); + String activityType = MethodExtractor.activityTypeName(activityInterface, method); + return startActivityImpl(activityType, Arrays.asList(arg1, arg2), options); + } + + @Override + public TemporalOperationResult startActivity( + Class activityInterface, + Functions.Proc4 activityMethod, + A1 arg1, + A2 arg2, + A3 arg3, + StartActivityOptions options) { + Method method = MethodExtractor.extract(activityInterface, activityMethod); + String activityType = MethodExtractor.activityTypeName(activityInterface, method); + return startActivityImpl(activityType, Arrays.asList(arg1, arg2, arg3), options); + } + + @Override + public TemporalOperationResult startActivity( + Class activityInterface, + Functions.Proc5 activityMethod, + A1 arg1, + A2 arg2, + A3 arg3, + A4 arg4, + StartActivityOptions options) { + Method method = MethodExtractor.extract(activityInterface, activityMethod); + String activityType = MethodExtractor.activityTypeName(activityInterface, method); + return startActivityImpl(activityType, Arrays.asList(arg1, arg2, arg3, arg4), options); + } + + @Override + public TemporalOperationResult startActivity( + Class activityInterface, + Functions.Proc6 activityMethod, + A1 arg1, + A2 arg2, + A3 arg3, + A4 arg4, + A5 arg5, + StartActivityOptions options) { + Method method = MethodExtractor.extract(activityInterface, activityMethod); + String activityType = MethodExtractor.activityTypeName(activityInterface, method); + return startActivityImpl(activityType, Arrays.asList(arg1, arg2, arg3, arg4, arg5), options); + } + + @Override + public TemporalOperationResult startActivity( + Class activityInterface, + Functions.Proc7 activityMethod, + A1 arg1, + A2 arg2, + A3 arg3, + A4 arg4, + A5 arg5, + A6 arg6, + StartActivityOptions options) { + Method method = MethodExtractor.extract(activityInterface, activityMethod); + String activityType = MethodExtractor.activityTypeName(activityInterface, method); + return startActivityImpl( + activityType, Arrays.asList(arg1, arg2, arg3, arg4, arg5, arg6), options); + } + + // ---------- Activity untyped ---------- + + @Override + public TemporalOperationResult startActivity( + String activityType, Class resultClass, StartActivityOptions options, Object... args) { + List argList = args == null ? Collections.emptyList() : Arrays.asList(args); + return startActivityImpl(activityType, argList, options); + } + + private TemporalOperationResult startActivityImpl( + String activityType, List args, StartActivityOptions options) { + markAsyncOperationStarted(); + try { + NexusStartActivityResponse response = + NexusStartActivityHelper.startActivityAndAttachLinks( + operationContext, + operationStartDetails, + activityType, + args, + options, + Header.empty(), + request -> { + ActivityClientCallsInterceptor.CompletionCallback cb = + (request.getCallbackUrl() == null || request.getCallbackUrl().isEmpty()) + ? null + : new ActivityClientCallsInterceptor.CompletionCallback( + request.getCallbackUrl(), request.getCallbackHeaders()); + ActivityClientCallsInterceptor.StartActivityInput input = + new ActivityClientCallsInterceptor.StartActivityInput( + request.getActivityType(), + request.getArgs(), + request.getOptions(), + request.getHeader(), + cb, + request.getLinks()); + // Build an ActivityClient that mirrors the surrounding WorkflowClient's options + // so that the metrics-tagged scope built by the impl is preserved. setIdentity is + // load-bearing because the cancel RPC reads it from clientOptions.getIdentity(). + ActivityClient activityClient = + ActivityClient.newInstance( + client.getWorkflowServiceStubs(), + ActivityClientOptions.newBuilder() + .setNamespace(client.getOptions().getNamespace()) + .setDataConverter(client.getOptions().getDataConverter()) + .setIdentity(client.getOptions().getIdentity()) + .build()); + ActivityClientCallsInterceptor.StartActivityOutput out = + ((ActivityClientInternal) activityClient).getInvoker().startActivity(input); + // The invoker generated and injected the runId-free token into the callback + // headers before the start RPC fired. The header token cannot include a run ID + // because the run ID isn't known until after the start RPC returns. The operation + // token returned to the Nexus caller can — and should — include it, so it's + // regenerated here from the same activity ID + the run ID the start RPC produced. + String headerToken = out.getNexusOperationToken(); + if (headerToken == null) { + throw new HandlerException( + HandlerException.ErrorType.INTERNAL, + "invoker did not return a Nexus operation token for activity start with callback", + new IllegalStateException( + "nexusOperationToken is null on StartActivityOutput when CompletionCallback was set")); + } + String returnToken; + try { + returnToken = + OperationTokenUtil.generateActivityExecutionOperationToken( + out.getActivityId(), + out.getActivityRunId(), + client.getOptions().getNamespace()); + } catch (JsonProcessingException e) { + throw new HandlerException( + HandlerException.ErrorType.INTERNAL, + "failed to generate activity operation token", + e); + } + return new NexusStartActivityResponse( + out.getActivityId(), out.getActivityRunId(), returnToken); + }); + return TemporalOperationResult.async(response.getOperationToken()); + } catch (Throwable t) { + // Reset on failure so that if the activity start throws, the handler can retry without + // being blocked by the guard. + asyncOperationStarted.set(false); + throw t; + } + } + + private void markAsyncOperationStarted() { + if (!asyncOperationStarted.compareAndSet(false, true)) { + throw new HandlerException( + HandlerException.ErrorType.BAD_REQUEST, + new IllegalStateException( + "Only one async operation can be started per operation handler invocation. " + + "Use getWorkflowClient() for additional workflow interactions.")); + } + } } 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 6a01d11fc6..584d702229 100644 --- a/temporal-sdk/src/main/java/io/temporal/nexus/TemporalOperationHandler.java +++ b/temporal-sdk/src/main/java/io/temporal/nexus/TemporalOperationHandler.java @@ -1,17 +1,19 @@ package io.temporal.nexus; import io.nexusrpc.handler.*; +import io.temporal.client.ActivityClient; +import io.temporal.client.ActivityClientOptions; import io.temporal.client.WorkflowClient; import io.temporal.common.Experimental; import io.temporal.internal.nexus.CurrentNexusOperationContext; import io.temporal.internal.nexus.InternalNexusOperationContext; import io.temporal.internal.nexus.OperationToken; -import io.temporal.internal.nexus.OperationTokenType; import io.temporal.internal.nexus.OperationTokenUtil; /** * Generic Nexus operation handler backed by Temporal. Implements {@link OperationHandler} and - * provides a composable way to map Temporal operations (start workflow, etc.) to Nexus operations. + * provides a composable way to map Temporal operations (start workflow, start activity, etc.) to + * Nexus operations. * *

Usage example: * @@ -30,8 +32,10 @@ * } * *

This class supports subclassing to customize cancel behavior. Override {@link - * #cancelWorkflowRun} to change how workflow-run cancellations are handled. The {@link #start} and - * {@link #cancel} methods should not be overridden — they contain the core dispatch logic. + * #cancelWorkflowRun} to change how workflow-run (token type {@code t:1}) cancellations are + * handled, or {@link #cancelActivityExecution} to change how activity-execution (token type {@code + * t:4}) cancellations are handled. The {@link #start} and {@link #cancel} methods should not be + * overridden — they contain the core dispatch logic. * * @param the input type * @param the result type @@ -59,7 +63,7 @@ protected TemporalOperationHandler(StartHandler startHandler) { /** * Creates a {@link TemporalOperationHandler} from a start handler. Subclass and override {@link - * #cancelWorkflowRun} to customize cancel behavior. + * #cancelWorkflowRun} or {@link #cancelActivityExecution} to customize cancel behavior. * * @param startHandler the handler to invoke on start operation requests * @return an operation handler backed by the given start handler @@ -100,12 +104,19 @@ public final void cancel(OperationContext ctx, OperationCancelDetails details) { } TemporalOperationCancelContext cancelContext = new TemporalOperationCancelContext(ctx, details); - if (token.getType() == OperationTokenType.WORKFLOW_RUN) { - cancelWorkflowRun(cancelContext, new CancelWorkflowRunInput(token.getWorkflowId())); - } else { - throw new HandlerException( - HandlerException.ErrorType.BAD_REQUEST, - new IllegalArgumentException("unsupported operation token type: " + token.getType())); + switch (token.getType()) { + case WORKFLOW_RUN: + cancelWorkflowRun(cancelContext, new CancelWorkflowRunInput(token.getWorkflowId())); + break; + case ACTIVITY_EXECUTION: + cancelActivityExecution( + cancelContext, + new CancelActivityExecutionInput(token.getActivityId(), token.getRunId())); + break; + default: + throw new HandlerException( + HandlerException.ErrorType.BAD_REQUEST, + new IllegalArgumentException("unsupported operation token type: " + token.getType())); } } @@ -123,4 +134,27 @@ protected void cancelWorkflowRun( WorkflowClient client = CurrentNexusOperationContext.get().getWorkflowClient(); client.newUntypedWorkflowStub(input.getWorkflowId()).cancel(); } + + /** + * Called when a cancel request is received for an activity-execution token (type=4). Override to + * customize cancel behavior. + * + *

Default behavior: requests cancellation of the underlying standalone activity execution. + * + * @param context the cancel context + * @param input describes the activity execution to cancel + */ + protected void cancelActivityExecution( + TemporalOperationCancelContext context, CancelActivityExecutionInput input) { + WorkflowClient wc = CurrentNexusOperationContext.get().getWorkflowClient(); + ActivityClient ac = + ActivityClient.newInstance( + wc.getWorkflowServiceStubs(), + ActivityClientOptions.newBuilder() + .setNamespace(wc.getOptions().getNamespace()) + .setDataConverter(wc.getOptions().getDataConverter()) + .setIdentity(wc.getOptions().getIdentity()) + .build()); + ac.getHandle(input.getActivityId(), input.getRunId()).cancel(); + } } diff --git a/temporal-sdk/src/test/java/io/temporal/internal/nexus/WorkflowRunTokenTest.java b/temporal-sdk/src/test/java/io/temporal/internal/nexus/WorkflowRunTokenTest.java index 1f22fe8c2e..ad55f7a3ab 100644 --- a/temporal-sdk/src/test/java/io/temporal/internal/nexus/WorkflowRunTokenTest.java +++ b/temporal-sdk/src/test/java/io/temporal/internal/nexus/WorkflowRunTokenTest.java @@ -92,6 +92,92 @@ public void loadWorkflowIdFromGoOperationToken() { Assert.assertEquals(OperationTokenType.WORKFLOW_RUN, token.getType()); } + @Test + public void roundTripActivityExecutionToken() throws JsonProcessingException { + String encoded = OperationTokenUtil.generateActivityExecutionOperationToken("act-1", "ns"); + OperationToken token = OperationTokenUtil.loadOperationToken(encoded); + Assert.assertEquals(OperationTokenType.ACTIVITY_EXECUTION, token.getType()); + Assert.assertEquals("act-1", token.getActivityId()); + Assert.assertEquals("ns", token.getNamespace()); + Assert.assertNull(token.getWorkflowId()); + Assert.assertNull(token.getRunId()); + Assert.assertNull(token.getVersion()); + + // Also exercise the symmetric activityId loader. + Assert.assertEquals("act-1", OperationTokenUtil.loadActivityIdFromOperationToken(encoded)); + } + + @Test + public void roundTripActivityExecutionTokenWithRunId() throws JsonProcessingException { + String encoded = + OperationTokenUtil.generateActivityExecutionOperationToken("act-1", "run-1", "ns"); + OperationToken token = OperationTokenUtil.loadOperationToken(encoded); + Assert.assertEquals(OperationTokenType.ACTIVITY_EXECUTION, token.getType()); + Assert.assertEquals("act-1", token.getActivityId()); + Assert.assertEquals("run-1", token.getRunId()); + Assert.assertEquals("ns", token.getNamespace()); + Assert.assertNull(token.getWorkflowId()); + Assert.assertNull(token.getVersion()); + } + + @Test + public void activityExecutionTokenOmitsRunIdWhenNull() throws JsonProcessingException { + // The header-token path passes runId=null and must produce a payload byte-identical to the + // two-arg overload — the runId-aware overload is the *only* one used in code now. + String withRunId = + OperationTokenUtil.generateActivityExecutionOperationToken("act-1", null, "ns"); + String withoutRunId = OperationTokenUtil.generateActivityExecutionOperationToken("act-1", "ns"); + Assert.assertEquals(withoutRunId, withRunId); + } + + @Test + public void workflowRunTokenBytesByteIdenticalSnapshot() throws JsonProcessingException { + String encoded = OperationTokenUtil.generateWorkflowRunOperationToken("wf-1", "ns"); + // The encoded token must remain byte-identical for cross-SDK compatibility. + // Snapshot captured from the current (pre-aid) SDK output. + Assert.assertEquals("eyJ0IjoxLCJucyI6Im5zIiwid2lkIjoid2YtMSJ9", encoded); + } + + @Test + public void malformedActivityTokenRejected() { + String malformed = "{\"t\":4,\"ns\":\"ns\",\"aid\":\"\"}"; + IllegalArgumentException ex = + Assert.assertThrows( + IllegalArgumentException.class, + () -> + OperationTokenUtil.loadOperationToken( + encoder.encodeToString(malformed.getBytes()))); + Assert.assertTrue( + "expected 'missing activity ID' in message but got: " + ex.getMessage(), + ex.getMessage().contains("missing activity ID")); + } + + @Test + public void unknownOperationTokenTypeRejected() { + String unknown = "{\"t\":7,\"ns\":\"ns\",\"wid\":\"x\"}"; + IllegalArgumentException ex = + Assert.assertThrows( + IllegalArgumentException.class, + () -> + OperationTokenUtil.loadOperationToken(encoder.encodeToString(unknown.getBytes()))); + Assert.assertTrue( + "expected 'unknown operation token type' in message but got: " + ex.getMessage(), + ex.getMessage().contains("unknown operation token type")); + } + + @Test + public void loadWorkflowRunRejectsActivityToken() throws JsonProcessingException { + String activityToken = + OperationTokenUtil.generateActivityExecutionOperationToken("act-1", "ns"); + IllegalArgumentException ex = + Assert.assertThrows( + IllegalArgumentException.class, + () -> OperationTokenUtil.loadWorkflowRunOperationToken(activityToken)); + Assert.assertTrue( + "expected 'incorrect operation token type' in message but got: " + ex.getMessage(), + ex.getMessage().contains("incorrect operation token type")); + } + @Test public void loadWorkflowIdFromBadOperationToken() { // Bad token, empty json diff --git a/temporal-sdk/src/test/java/io/temporal/nexus/TemporalNexusClientImplTest.java b/temporal-sdk/src/test/java/io/temporal/nexus/TemporalNexusClientImplTest.java new file mode 100644 index 0000000000..020124f6ff --- /dev/null +++ b/temporal-sdk/src/test/java/io/temporal/nexus/TemporalNexusClientImplTest.java @@ -0,0 +1,248 @@ +package io.temporal.nexus; + +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import com.uber.m3.tally.NoopScope; +import io.nexusrpc.handler.HandlerException; +import io.nexusrpc.handler.OperationContext; +import io.nexusrpc.handler.OperationStartDetails; +import io.temporal.client.StartActivityOptions; +import io.temporal.client.WorkflowClient; +import io.temporal.client.WorkflowClientOptions; +import io.temporal.client.WorkflowOptions; +import io.temporal.internal.nexus.CurrentNexusOperationContext; +import io.temporal.internal.nexus.InternalNexusOperationContext; +import java.time.Duration; +import org.junit.After; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; + +/** + * Pure unit tests for {@link TemporalNexusClientImpl#markAsyncOperationStarted()} semantics. These + * run without a Temporal server (no {@link io.temporal.testing.internal.SDKTestWorkflowRule}). + * + *

The {@code markAsyncOperationStarted} guard fires as the very first statement in both {@code + * startActivityImpl} and {@code invokeAndReturn}, so the second call always throws {@link + * HandlerException}({@link HandlerException.ErrorType#BAD_REQUEST}) regardless of whether the first + * call's downstream RPC succeeded. + */ +public class TemporalNexusClientImplTest { + + private static final String NAMESPACE = "test-namespace"; + private static final String TASK_QUEUE = "test-task-queue"; + private static final String ENDPOINT = "test-endpoint"; + + private TemporalNexusClientImpl client; + + @Before + public void setUp() { + // Mock WorkflowClient so constructor doesn't need a live server. + WorkflowClient workflowClient = mock(WorkflowClient.class); + WorkflowClientOptions clientOptions = mock(WorkflowClientOptions.class); + when(workflowClient.getOptions()).thenReturn(clientOptions); + when(clientOptions.getNamespace()).thenReturn(NAMESPACE); + + OperationContext operationContext = mock(OperationContext.class); + when(operationContext.getService()).thenReturn("TestService"); + when(operationContext.getOperation()).thenReturn("testOperation"); + + OperationStartDetails operationStartDetails = + OperationStartDetails.newBuilder() + .setCallbackUrl("http://localhost/callback") + .setRequestId("test-request-id") + .build(); + + client = new TemporalNexusClientImpl(workflowClient, operationContext, operationStartDetails); + + // Set up the thread-local nexus context required by NexusStartActivityHelper and + // NexusStartWorkflowHelper deep in the call stack. + InternalNexusOperationContext internalCtx = + new InternalNexusOperationContext( + NAMESPACE, TASK_QUEUE, ENDPOINT, new NoopScope(), workflowClient); + CurrentNexusOperationContext.set(internalCtx); + } + + @After + public void tearDown() { + CurrentNexusOperationContext.unset(); + } + + // ---------- Activity double-start ---------- + + @Test + public void doubleStartActivity_secondCallThrowsBadRequest() { + StartActivityOptions options = + StartActivityOptions.newBuilder() + .setId("act-1") + .setTaskQueue(TASK_QUEUE) + .setStartToCloseTimeout(Duration.ofSeconds(10)) + .build(); + + // First call: markAsyncOperationStarted() succeeds (sets flag), RPC may throw — we don't care. + try { + client.startActivity(TestActivity.class, TestActivity::doSomething, options); + } catch (HandlerException e) { + // If a HandlerException leaks out of the first call it must NOT be BAD_REQUEST + // from markAsyncOperationStarted — that would mean the flag was already set before setUp. + Assert.assertNotEquals( + "First startActivity must not fail with BAD_REQUEST (markAsyncOperationStarted guard)", + HandlerException.ErrorType.BAD_REQUEST, + e.getErrorType()); + } catch (Exception ignored) { + // Any other exception from the RPC layer is expected; the important thing is + // markAsyncOperationStarted already ran and set asyncOperationStarted = true. + } + + // Second call: markAsyncOperationStarted() sees the flag and must throw BAD_REQUEST + // immediately. + HandlerException ex = + Assert.assertThrows( + HandlerException.class, + () -> + client.startActivity( + TestActivity.class, + TestActivity::doSomething, + StartActivityOptions.newBuilder() + .setId("act-2") + .setTaskQueue(TASK_QUEUE) + .setStartToCloseTimeout(Duration.ofSeconds(10)) + .build())); + + Assert.assertEquals(HandlerException.ErrorType.BAD_REQUEST, ex.getErrorType()); + Assert.assertTrue( + "Message should contain 'Only one async operation'", + ex.getCause() instanceof IllegalStateException + && ex.getCause() + .getMessage() + .startsWith("Only one async operation can be started per operation handler")); + } + + // ---------- Workflow double-start ---------- + + @Test + public void doubleStartWorkflow_secondCallThrowsBadRequest() { + WorkflowOptions options = + WorkflowOptions.newBuilder().setWorkflowId("wf-1").setTaskQueue(TASK_QUEUE).build(); + + // First call: markAsyncOperationStarted() succeeds, downstream RPC may throw — we don't care. + try { + client.startWorkflow(BlockingWorkflow.class, BlockingWorkflow::execute, "input", options); + } catch (HandlerException e) { + Assert.assertNotEquals( + "First startWorkflow must not fail with BAD_REQUEST (markAsyncOperationStarted guard)", + HandlerException.ErrorType.BAD_REQUEST, + e.getErrorType()); + } catch (Exception ignored) { + // Expected: no live server. + } + + // Second call must throw BAD_REQUEST immediately. + HandlerException ex = + Assert.assertThrows( + HandlerException.class, + () -> + client.startWorkflow( + BlockingWorkflow.class, + BlockingWorkflow::execute, + "input", + WorkflowOptions.newBuilder() + .setWorkflowId("wf-2") + .setTaskQueue(TASK_QUEUE) + .build())); + + Assert.assertEquals(HandlerException.ErrorType.BAD_REQUEST, ex.getErrorType()); + Assert.assertTrue( + "Message should contain 'Only one async operation'", + ex.getCause() instanceof IllegalStateException + && ex.getCause() + .getMessage() + .startsWith("Only one async operation can be started per operation handler")); + } + + // ---------- Mixed: workflow then activity ---------- + + @Test + public void startWorkflowThenActivity_activityThrowsBadRequest() { + WorkflowOptions wfOptions = + WorkflowOptions.newBuilder().setWorkflowId("wf-mixed-1").setTaskQueue(TASK_QUEUE).build(); + + try { + client.startWorkflow(BlockingWorkflow.class, BlockingWorkflow::execute, "in", wfOptions); + } catch (HandlerException e) { + Assert.assertNotEquals(HandlerException.ErrorType.BAD_REQUEST, e.getErrorType()); + } catch (Exception ignored) { + } + + HandlerException ex = + Assert.assertThrows( + HandlerException.class, + () -> + client.startActivity( + TestActivity.class, + TestActivity::doSomething, + StartActivityOptions.newBuilder() + .setId("act-mixed-1") + .setTaskQueue(TASK_QUEUE) + .setStartToCloseTimeout(Duration.ofSeconds(10)) + .build())); + + Assert.assertEquals(HandlerException.ErrorType.BAD_REQUEST, ex.getErrorType()); + } + + // ---------- Mixed: activity then workflow ---------- + + @Test + public void startActivityThenWorkflow_workflowThrowsBadRequest() { + StartActivityOptions actOptions = + StartActivityOptions.newBuilder() + .setId("act-mixed-2") + .setTaskQueue(TASK_QUEUE) + .setStartToCloseTimeout(Duration.ofSeconds(10)) + .build(); + + try { + client.startActivity(TestActivity.class, TestActivity::doSomething, actOptions); + } catch (HandlerException e) { + Assert.assertNotEquals(HandlerException.ErrorType.BAD_REQUEST, e.getErrorType()); + } catch (Exception ignored) { + } + + HandlerException ex = + Assert.assertThrows( + HandlerException.class, + () -> + client.startWorkflow( + BlockingWorkflow.class, + BlockingWorkflow::execute, + "in", + WorkflowOptions.newBuilder() + .setWorkflowId("wf-mixed-2") + .setTaskQueue(TASK_QUEUE) + .build())); + + Assert.assertEquals(HandlerException.ErrorType.BAD_REQUEST, ex.getErrorType()); + } + + // ---------- Minimal stubs ---------- + + @io.temporal.activity.ActivityInterface + public interface TestActivity { + @io.temporal.activity.ActivityMethod + void doSomething(); + } + + @io.temporal.workflow.WorkflowInterface + public interface BlockingWorkflow { + @io.temporal.workflow.WorkflowMethod + String execute(String input); + } + + public static class BlockingWorkflowImpl implements BlockingWorkflow { + @Override + public String execute(String input) { + return input; + } + } +} diff --git a/temporal-sdk/src/test/java/io/temporal/workflow/nexus/AsyncActivityOperationTest.java b/temporal-sdk/src/test/java/io/temporal/workflow/nexus/AsyncActivityOperationTest.java new file mode 100644 index 0000000000..455ea9f3d6 --- /dev/null +++ b/temporal-sdk/src/test/java/io/temporal/workflow/nexus/AsyncActivityOperationTest.java @@ -0,0 +1,217 @@ +package io.temporal.workflow.nexus; + +import static org.junit.Assume.assumeTrue; + +import io.nexusrpc.handler.HandlerException; +import io.nexusrpc.handler.OperationHandler; +import io.nexusrpc.handler.OperationImpl; +import io.nexusrpc.handler.ServiceImpl; +import io.temporal.activity.ActivityInterface; +import io.temporal.activity.ActivityMethod; +import io.temporal.client.StartActivityOptions; +import io.temporal.client.WorkflowFailedException; +import io.temporal.failure.ApplicationFailure; +import io.temporal.failure.NexusOperationFailure; +import io.temporal.internal.nexus.OperationToken; +import io.temporal.internal.nexus.OperationTokenType; +import io.temporal.internal.nexus.OperationTokenUtil; +import io.temporal.nexus.TemporalOperationHandler; +import io.temporal.testing.internal.SDKTestWorkflowRule; +import io.temporal.workflow.*; +import io.temporal.workflow.shared.TestNexusServices; +import io.temporal.workflow.shared.TestWorkflows; +import java.time.Duration; +import org.junit.Assert; +import org.junit.Rule; +import org.junit.Test; + +public class AsyncActivityOperationTest extends BaseNexusTest { + + @Rule + public SDKTestWorkflowRule testWorkflowRule = + SDKTestWorkflowRule.newBuilder() + .setWorkflowTypes(TestNexus.class, DoubleStartNexus.class) + .setActivityImplementations(new TestActivityImpl()) + .setNexusServiceImplementation( + new TestNexusServiceImpl(), new DoubleStartNexusServiceImpl()) + .build(); + + @Override + protected SDKTestWorkflowRule getTestWorkflowRule() { + return testWorkflowRule; + } + + @Test + public void testActivityOperationEndToEnd() { + // The in-process test-server does not implement the StartActivityExecution RPC; the + // standalone-activity Nexus path requires a real server. Unit-only token assertions stay + // active in testActivityOperationTokenRoundTrip below. + assumeTrue(SDKTestWorkflowRule.useExternalService); + // Combined (a) caller workflow receives "hello " + input + // (b) opToken loads as ACTIVITY_EXECUTION with aid="act-" + requestId + TestWorkflows.TestWorkflow1 workflowStub = + testWorkflowRule.newWorkflowStubTimeoutOptions(TestWorkflows.TestWorkflow1.class); + String result = workflowStub.execute("world"); + Assert.assertEquals("hello world", result); + } + + @Test + public void testActivityOperationTokenRoundTrip() throws Exception { + // (c) generateActivityExecutionOperationToken then loadOperationToken round-trips equal + // aid + ns. + String token = OperationTokenUtil.generateActivityExecutionOperationToken("act-123", "test-ns"); + OperationToken loaded = OperationTokenUtil.loadOperationToken(token); + Assert.assertEquals(OperationTokenType.ACTIVITY_EXECUTION, loaded.getType()); + Assert.assertEquals("act-123", loaded.getActivityId()); + Assert.assertEquals("test-ns", loaded.getNamespace()); + Assert.assertNull(loaded.getWorkflowId()); + } + + @Test + public void testDoubleStartActivityThrows() { + // The first start RPC requires StartActivityExecution which is not implemented by the + // in-process test server; gate this on a real server so the guard at the second call can be + // observed. + assumeTrue(SDKTestWorkflowRule.useExternalService); + // (d) Calling startActivity twice in one handler invocation throws + // HandlerException(BAD_REQUEST). + TestDoubleStartWorkflow workflowStub = + testWorkflowRule.newWorkflowStubTimeoutOptions(TestDoubleStartWorkflow.class); + WorkflowFailedException e = + Assert.assertThrows(WorkflowFailedException.class, () -> workflowStub.execute("anything")); + Assert.assertTrue(e.getCause() instanceof NexusOperationFailure); + NexusOperationFailure nexusFailure = (NexusOperationFailure) e.getCause(); + Assert.assertTrue(nexusFailure.getCause() instanceof HandlerException); + HandlerException handlerException = (HandlerException) nexusFailure.getCause(); + Assert.assertTrue(handlerException.getCause() instanceof ApplicationFailure); + ApplicationFailure appFailure = (ApplicationFailure) handlerException.getCause(); + Assert.assertEquals("java.lang.IllegalStateException", appFailure.getType()); + Assert.assertTrue( + appFailure + .getOriginalMessage() + .startsWith("Only one async operation can be started per operation handler")); + } + + @ActivityInterface + public interface TestActivity { + @ActivityMethod + String process(String input); + } + + public static class TestActivityImpl implements TestActivity { + @Override + public String process(String input) { + return "hello " + input; + } + } + + public static class TestNexus implements TestWorkflows.TestWorkflow1 { + @Override + public String execute(String input) { + NexusOperationOptions options = + NexusOperationOptions.newBuilder() + .setScheduleToCloseTimeout(Duration.ofSeconds(20)) + .build(); + NexusServiceOptions serviceOptions = + NexusServiceOptions.newBuilder() + .setEndpoint(getEndpointName()) + .setOperationOptions(options) + .build(); + TestNexusServices.TestNexusService1 serviceStub = + Workflow.newNexusServiceStub(TestNexusServices.TestNexusService1.class, serviceOptions); + + // Drive a handle so we can inspect the operation token for assertion (b). + NexusOperationHandle handle = + Workflow.startNexusOperation(serviceStub::operation, input); + NexusOperationExecution exec = handle.getExecution().get(); + Assert.assertTrue("Operation token should be present", exec.getOperationToken().isPresent()); + OperationToken token = OperationTokenUtil.loadOperationToken(exec.getOperationToken().get()); + Assert.assertEquals(OperationTokenType.ACTIVITY_EXECUTION, token.getType()); + Assert.assertTrue( + "activityId should start with 'act-' (got " + token.getActivityId() + ")", + token.getActivityId() != null && token.getActivityId().startsWith("act-")); + + return handle.getResult().get(); + } + } + + @ServiceImpl(service = TestNexusServices.TestNexusService1.class) + public class TestNexusServiceImpl { + @OperationImpl + public OperationHandler operation() { + return TemporalOperationHandler.create( + (context, client, input) -> + client.startActivity( + TestActivity.class, + TestActivity::process, + input, + StartActivityOptions.newBuilder() + .setId("act-" + context.getRequestId()) + .setTaskQueue(testWorkflowRule.getTaskQueue()) + .setStartToCloseTimeout(Duration.ofSeconds(10)) + .build())); + } + } + + // ---------- Double-start test scaffolding ---------- + + public static class DoubleStartNexus implements TestDoubleStartWorkflow { + @Override + public String execute(String input) { + NexusOperationOptions options = + NexusOperationOptions.newBuilder() + .setScheduleToCloseTimeout(Duration.ofSeconds(10)) + .build(); + NexusServiceOptions serviceOptions = + NexusServiceOptions.newBuilder() + .setEndpoint(getEndpointName()) + .setOperationOptions(options) + .build(); + DoubleStartNexusService stub = + Workflow.newNexusServiceStub(DoubleStartNexusService.class, serviceOptions); + return stub.operation(input); + } + } + + @WorkflowInterface + public interface TestDoubleStartWorkflow { + @WorkflowMethod + String execute(String input); + } + + @io.nexusrpc.Service + public interface DoubleStartNexusService { + @io.nexusrpc.Operation + String operation(String input); + } + + @ServiceImpl(service = DoubleStartNexusService.class) + public class DoubleStartNexusServiceImpl { + @OperationImpl + public OperationHandler operation() { + return TemporalOperationHandler.create( + (context, client, input) -> { + // First start should succeed. + client.startActivity( + TestActivity.class, + TestActivity::process, + input, + StartActivityOptions.newBuilder() + .setId("act-first-" + context.getRequestId()) + .setTaskQueue(testWorkflowRule.getTaskQueue()) + .setStartToCloseTimeout(Duration.ofSeconds(10)) + .build()); + // Second start must throw HandlerException(BAD_REQUEST). + return client.startActivity( + TestActivity.class, + TestActivity::process, + input, + StartActivityOptions.newBuilder() + .setId("act-second-" + context.getRequestId()) + .setTaskQueue(testWorkflowRule.getTaskQueue()) + .setStartToCloseTimeout(Duration.ofSeconds(10)) + .build()); + }); + } + } +} diff --git a/temporal-sdk/src/test/java/io/temporal/workflow/nexus/CancelActivityAsyncOperationTest.java b/temporal-sdk/src/test/java/io/temporal/workflow/nexus/CancelActivityAsyncOperationTest.java new file mode 100644 index 0000000000..45a2b78794 --- /dev/null +++ b/temporal-sdk/src/test/java/io/temporal/workflow/nexus/CancelActivityAsyncOperationTest.java @@ -0,0 +1,264 @@ +package io.temporal.workflow.nexus; + +import static org.junit.Assume.assumeTrue; + +import io.nexusrpc.handler.OperationHandler; +import io.nexusrpc.handler.OperationImpl; +import io.nexusrpc.handler.ServiceImpl; +import io.temporal.activity.Activity; +import io.temporal.activity.ActivityInterface; +import io.temporal.activity.ActivityMethod; +import io.temporal.client.ActivityCompletionException; +import io.temporal.client.StartActivityOptions; +import io.temporal.client.WorkflowFailedException; +import io.temporal.client.WorkflowStub; +import io.temporal.common.RetryOptions; +import io.temporal.failure.CanceledFailure; +import io.temporal.failure.NexusOperationFailure; +import io.temporal.nexus.CancelActivityExecutionInput; +import io.temporal.nexus.TemporalOperationCancelContext; +import io.temporal.nexus.TemporalOperationHandler; +import io.temporal.testing.internal.SDKTestWorkflowRule; +import io.temporal.workflow.*; +import java.time.Duration; +import java.util.concurrent.atomic.AtomicBoolean; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Rule; +import org.junit.Test; + +public class CancelActivityAsyncOperationTest extends BaseNexusTest { + + static final AtomicBoolean cancelled = new AtomicBoolean(false); + static final AtomicBoolean customCancelInvoked = new AtomicBoolean(false); + + @Rule + public SDKTestWorkflowRule testWorkflowRule = + SDKTestWorkflowRule.newBuilder() + .setWorkflowTypes(DefaultCancelNexus.class, OverriddenCancelNexus.class) + .setActivityImplementations(new HeartbeatingActivityImpl()) + .setNexusServiceImplementation( + new DefaultCancelNexusServiceImpl(), new OverriddenCancelNexusServiceImpl()) + .build(); + + @Override + protected SDKTestWorkflowRule getTestWorkflowRule() { + return testWorkflowRule; + } + + @Before + public void resetState() { + cancelled.set(false); + customCancelInvoked.set(false); + } + + @Test(timeout = 60_000) + public void testDefaultActivityCancel() { + // Standalone-activity Nexus path requires a real Temporal server; the in-process test server + // does not implement StartActivityExecution. + assumeTrue(SDKTestWorkflowRule.useExternalService); + WorkflowStub stub = + testWorkflowRule.newUntypedWorkflowStubTimeoutOptions("DefaultCancelCaller"); + stub.start(testWorkflowRule.getTaskQueue()); + // Either succeeds with "ok" or throws a workflow failure; we only care about the cancel path. + try { + stub.getResult(String.class); + } catch (WorkflowFailedException ignored) { + // Acceptable: caller workflow may surface the cancel as failure. + } + // History contains EVENT_TYPE_NEXUS_OPERATION_CANCEL_REQUESTED on the caller workflow. + testWorkflowRule.assertHistoryEvent( + stub.getExecution().getWorkflowId(), + io.temporal.api.enums.v1.EventType.EVENT_TYPE_NEXUS_OPERATION_CANCEL_REQUESTED); + // Activity worker observed cancel. + Assert.assertTrue("activity should have observed cancel", cancelled.get()); + } + + @Test(timeout = 60_000) + public void testOverriddenActivityCancel() { + // Same constraint as testDefaultActivityCancel. + assumeTrue(SDKTestWorkflowRule.useExternalService); + WorkflowStub stub = + testWorkflowRule.newUntypedWorkflowStubTimeoutOptions("OverriddenCancelCaller"); + stub.start(testWorkflowRule.getTaskQueue()); + try { + stub.getResult(String.class); + } catch (WorkflowFailedException ignored) { + // Acceptable. + } + // The overriding handler should have been called. + Assert.assertTrue("override should have been invoked", customCancelInvoked.get()); + // Default cancel was not invoked: activity ran until startToCloseTimeout fired + // (no ActivityCompletionException raised on the worker). + Assert.assertFalse("default cancel should not have been invoked", cancelled.get()); + } + + @ActivityInterface + public interface HeartbeatingActivity { + @ActivityMethod + String process(String input); + } + + public static class HeartbeatingActivityImpl implements HeartbeatingActivity { + @Override + public String process(String input) { + while (true) { + try { + Activity.getExecutionContext().heartbeat(null); + } catch (ActivityCompletionException ex) { + cancelled.set(true); + throw ex; + } + try { + Thread.sleep(100); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + break; + } + } + return "done"; + } + } + + @WorkflowInterface + public interface DefaultCancelCaller { + @WorkflowMethod + String execute(String taskQueue); + } + + @WorkflowInterface + public interface OverriddenCancelCaller { + @WorkflowMethod + String execute(String taskQueue); + } + + @io.nexusrpc.Service + public interface DefaultCancelNexusService { + @io.nexusrpc.Operation + String operation(String taskQueue); + } + + @io.nexusrpc.Service + public interface OverrideCancelNexusService { + @io.nexusrpc.Operation + String operation(String taskQueue); + } + + // ---- Default cancel ---- + + public static class DefaultCancelNexus implements DefaultCancelCaller { + @Override + public String execute(String taskQueue) { + NexusOperationOptions options = + NexusOperationOptions.newBuilder() + .setScheduleToCloseTimeout(Duration.ofSeconds(15)) + .build(); + NexusServiceOptions serviceOptions = + NexusServiceOptions.newBuilder() + .setEndpoint(getEndpointName()) + .setOperationOptions(options) + .build(); + DefaultCancelNexusService stub = + Workflow.newNexusServiceStub(DefaultCancelNexusService.class, serviceOptions); + try { + Workflow.newCancellationScope( + () -> { + NexusOperationHandle handle = + Workflow.startNexusOperation(stub::operation, taskQueue); + handle.getExecution().get(); + CancellationScope.current().cancel(); + handle.getResult().get(); + }) + .run(); + } catch (NexusOperationFailure failure) { + if (!(failure.getCause() instanceof CanceledFailure)) { + throw failure; + } + } + return "ok"; + } + } + + @ServiceImpl(service = DefaultCancelNexusService.class) + public class DefaultCancelNexusServiceImpl { + @OperationImpl + public OperationHandler operation() { + return TemporalOperationHandler.create( + (context, client, input) -> + client.startActivity( + HeartbeatingActivity.class, + HeartbeatingActivity::process, + input, + StartActivityOptions.newBuilder() + .setId("act-" + context.getRequestId()) + .setTaskQueue(input) + .setStartToCloseTimeout(Duration.ofSeconds(60)) + .setHeartbeatTimeout(Duration.ofSeconds(2)) + .setRetryOptions(RetryOptions.newBuilder().setMaximumAttempts(1).build()) + .build())); + } + } + + // ---- Overridden cancel ---- + + public static class OverriddenCancelNexus implements OverriddenCancelCaller { + @Override + public String execute(String taskQueue) { + NexusOperationOptions options = + NexusOperationOptions.newBuilder() + .setScheduleToCloseTimeout(Duration.ofSeconds(15)) + .build(); + NexusServiceOptions serviceOptions = + NexusServiceOptions.newBuilder() + .setEndpoint(getEndpointName()) + .setOperationOptions(options) + .build(); + OverrideCancelNexusService stub = + Workflow.newNexusServiceStub(OverrideCancelNexusService.class, serviceOptions); + try { + Workflow.newCancellationScope( + () -> { + NexusOperationHandle handle = + Workflow.startNexusOperation(stub::operation, taskQueue); + handle.getExecution().get(); + CancellationScope.current().cancel(); + handle.getResult().get(); + }) + .run(); + } catch (NexusOperationFailure failure) { + if (!(failure.getCause() instanceof CanceledFailure)) { + throw failure; + } + } + return "ok"; + } + } + + @ServiceImpl(service = OverrideCancelNexusService.class) + public class OverriddenCancelNexusServiceImpl { + @OperationImpl + public OperationHandler operation() { + return new TemporalOperationHandler( + (context, client, input) -> + client.startActivity( + HeartbeatingActivity.class, + HeartbeatingActivity::process, + input, + StartActivityOptions.newBuilder() + .setId("act-override-" + context.getRequestId()) + .setTaskQueue(input) + .setStartToCloseTimeout(Duration.ofSeconds(5)) + .setHeartbeatTimeout(Duration.ofSeconds(2)) + .setRetryOptions(RetryOptions.newBuilder().setMaximumAttempts(1).build()) + .build())) { + @Override + protected void cancelActivityExecution( + TemporalOperationCancelContext ctx, CancelActivityExecutionInput input) { + customCancelInvoked.set(true); + // Intentionally do NOT invoke default cancel; the activity will self-terminate via + // startToCloseTimeout. + } + }; + } + } +} From 441ee33b6c95fbe8201a18d1c6858e303034c7d2 Mon Sep 17 00:00:00 2001 From: Quinn Klassen Date: Thu, 23 Jul 2026 09:10:25 -0700 Subject: [PATCH 2/2] Align with Update approach --- .../ActivityClientCallsInterceptor.java | 66 --------- .../client/RootActivityClientInvoker.java | 62 +++----- .../nexus/InternalNexusOperationContext.java | 10 ++ .../nexus/NexusOperationMetadata.java | 22 +++ .../nexus/TemporalNexusClientImpl.java | 25 ++-- .../client/RootActivityClientInvokerTest.java | 139 ++++++++++++++++++ .../nexus/TemporalNexusClientImplTest.java | 102 ++++++++----- 7 files changed, 269 insertions(+), 157 deletions(-) create mode 100644 temporal-sdk/src/main/java/io/temporal/internal/nexus/NexusOperationMetadata.java create mode 100644 temporal-sdk/src/test/java/io/temporal/internal/client/RootActivityClientInvokerTest.java diff --git a/temporal-sdk/src/main/java/io/temporal/common/interceptors/ActivityClientCallsInterceptor.java b/temporal-sdk/src/main/java/io/temporal/common/interceptors/ActivityClientCallsInterceptor.java index 5e278969d5..16a34dc285 100644 --- a/temporal-sdk/src/main/java/io/temporal/common/interceptors/ActivityClientCallsInterceptor.java +++ b/temporal-sdk/src/main/java/io/temporal/common/interceptors/ActivityClientCallsInterceptor.java @@ -9,8 +9,6 @@ import io.temporal.common.Experimental; import java.lang.reflect.Type; import java.util.List; -import java.util.Map; -import java.util.Objects; import java.util.concurrent.CompletableFuture; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; @@ -118,56 +116,19 @@ GetActivityResultOutput getActivityResult(GetActivityResultInput input CompletableFuture> getActivityResultAsync( GetActivityResultInput input); - /** - * Nexus completion-callback metadata propagated through {@link StartActivityInput} when an - * activity is scheduled from a Nexus operation handler. - */ - @Experimental - final class CompletionCallback { - private final String url; - private final Map headers; - - public CompletionCallback(String url, Map headers) { - this.url = Objects.requireNonNull(url); - this.headers = Objects.requireNonNull(headers); - } - - public String getUrl() { - return url; - } - - public Map getHeaders() { - return headers; - } - } - @Experimental final class StartActivityInput { private final String activityType; private final List args; private final StartActivityOptions options; private final Header header; - private final @Nullable CompletionCallback completionCallback; - private final @Nullable List links; public StartActivityInput( String activityType, List args, StartActivityOptions options, Header header) { - this(activityType, args, options, header, null, null); - } - - public StartActivityInput( - String activityType, - List args, - StartActivityOptions options, - Header header, - @Nullable CompletionCallback completionCallback, - @Nullable List links) { this.activityType = activityType; this.args = args; this.options = options; this.header = header; - this.completionCallback = completionCallback; - this.links = links; } public String getActivityType() { @@ -185,16 +146,6 @@ public StartActivityOptions getOptions() { public Header getHeader() { return header; } - - @Nullable - public CompletionCallback getCompletionCallback() { - return completionCallback; - } - - @Nullable - public List getLinks() { - return links; - } } @Experimental @@ -202,21 +153,9 @@ final class StartActivityOutput { private final String activityId; private final @Nullable String activityRunId; - /** - * Set by the invoker when the start request included a Nexus completion callback; null - * otherwise. Internal use; do not depend on this in user-facing code. - */ - private final @Nullable String nexusOperationToken; - public StartActivityOutput(String activityId, @Nullable String activityRunId) { - this(activityId, activityRunId, null); - } - - public StartActivityOutput( - String activityId, @Nullable String activityRunId, @Nullable String nexusOperationToken) { this.activityId = activityId; this.activityRunId = activityRunId; - this.nexusOperationToken = nexusOperationToken; } public String getActivityId() { @@ -227,11 +166,6 @@ public String getActivityId() { public String getActivityRunId() { return activityRunId; } - - @Nullable - public String getNexusOperationToken() { - return nexusOperationToken; - } } @Experimental 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 f9678c1996..c93289e813 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 @@ -22,20 +22,19 @@ import io.temporal.internal.client.external.GenericWorkflowClient; import io.temporal.internal.common.HeaderUtils; import io.temporal.internal.common.InternalUtils; -import io.temporal.internal.common.LinkConverter; import io.temporal.internal.common.ProtoConverters; import io.temporal.internal.common.ProtobufTimeUtils; import io.temporal.internal.common.SearchAttributesUtil; +import io.temporal.internal.nexus.CurrentNexusOperationContext; +import io.temporal.internal.nexus.InternalNexusOperationContext; +import io.temporal.internal.nexus.NexusOperationMetadata; import io.temporal.serviceclient.StatusUtils; import java.lang.reflect.Type; import java.util.*; import java.util.concurrent.CompletableFuture; import java.util.concurrent.CompletionException; import java.util.concurrent.TimeoutException; -import java.util.stream.Collectors; import java.util.stream.StreamSupport; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; /** * Terminus of the activity interceptor chain. Implements all activity RPCs against the Temporal @@ -43,8 +42,6 @@ */ public class RootActivityClientInvoker implements ActivityClientCallsInterceptor { - private static final Logger log = LoggerFactory.getLogger(RootActivityClientInvoker.class); - private final GenericWorkflowClient genericClient; private final ActivityClientOptions clientOptions; @@ -58,12 +55,19 @@ public RootActivityClientInvoker( public StartActivityOutput startActivity(StartActivityInput input) { StartActivityOptions options = input.getOptions(); DataConverter dc = clientOptions.getDataConverter(); + InternalNexusOperationContext nexusContext = + CurrentNexusOperationContext.isNexusContext() ? CurrentNexusOperationContext.get() : null; + NexusOperationMetadata nexusOperationMetadata = + nexusContext == null ? null : nexusContext.getNexusOperationMetadata(); StartActivityExecutionRequest.Builder request = StartActivityExecutionRequest.newBuilder() .setNamespace(clientOptions.getNamespace()) .setIdentity(clientOptions.getIdentity()) - .setRequestId(UUID.randomUUID().toString()) + .setRequestId( + nexusOperationMetadata == null + ? UUID.randomUUID().toString() + : nexusOperationMetadata.requestId) .setActivityId(options.getId()) .setActivityType(ActivityType.newBuilder().setName(input.getActivityType()).build()) .setTaskQueue(TaskQueue.newBuilder().setName(options.getTaskQueue()).build()) @@ -113,41 +117,13 @@ public StartActivityOutput startActivity(StartActivityInput input) { io.temporal.api.common.v1.Header grpcHeader = HeaderUtils.toHeaderGrpc(input.getHeader(), null); request.setHeader(grpcHeader); - // Hoisted so it can be returned in StartActivityOutput when a Nexus callback is present. - String nexusOperationToken = null; - if (input.getCompletionCallback() != null) { - List protoLinks = null; - if (input.getLinks() != null) { - protoLinks = - input.getLinks().stream() - .map( - link -> { - if (io.temporal.api.common.v1.Link.WorkflowEvent.getDescriptor() - .getFullName() - .equals(link.getType())) { - io.temporal.api.nexus.v1.Link nexusLink = - io.temporal.api.nexus.v1.Link.newBuilder() - .setType(link.getType()) - .setUrl(link.getUri().toString()) - .build(); - return LinkConverter.nexusLinkToWorkflowEvent(nexusLink); - } else { - log.warn("ignoring unsupported link data type: {}", link.getType()); - return null; - } - }) - .filter(Objects::nonNull) - .collect(Collectors.toList()); - if (!protoLinks.isEmpty()) { - request.addAllLinks(protoLinks); - } else { - protoLinks = null; - } - } + if (nexusOperationMetadata != null) { + List protoLinks = nexusContext.getRequestLinks(); + request.addAllLinks(protoLinks); // Generate the operation token from the user-supplied activity ID and namespace so the // dual OPERATION_ID + OPERATION_TOKEN headers can be injected before the start RPC fires. try { - nexusOperationToken = + nexusOperationMetadata.operationToken = io.temporal.internal.nexus.OperationTokenUtil.generateActivityExecutionOperationToken( options.getId(), clientOptions.getNamespace()); } catch (com.fasterxml.jackson.core.JsonProcessingException e) { @@ -158,9 +134,9 @@ public StartActivityOutput startActivity(StartActivityInput input) { } Callback cb = InternalUtils.buildNexusCallback( - input.getCompletionCallback().getUrl(), - input.getCompletionCallback().getHeaders(), - nexusOperationToken, + nexusOperationMetadata.callbackUrl, + nexusOperationMetadata.callbackHeaders, + nexusOperationMetadata.operationToken, protoLinks); request.addCompletionCallbacks(cb); } @@ -182,7 +158,7 @@ public StartActivityOutput startActivity(StartActivityInput input) { } String runId = response.getRunId().isEmpty() ? null : response.getRunId(); - return new StartActivityOutput(options.getId(), runId, nexusOperationToken); + return new StartActivityOutput(options.getId(), runId); } @Override 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 0308683857..d712627e14 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 @@ -37,6 +37,7 @@ public class InternalNexusOperationContext { // by responseLinksLock and getResponseLinks() returns a defensive copy taken under the lock. private final Object responseLinksLock = new Object(); private final List responseLinks = new ArrayList<>(); + private NexusOperationMetadata nexusOperationMetadata; public InternalNexusOperationContext( String namespace, @@ -82,6 +83,15 @@ public NexusOperationContext getUserFacingContext() { return new NexusOperationContextImpl(); } + /** Sets metadata for the Temporal primitive backing the current Nexus operation. */ + public void setNexusOperationMetadata(NexusOperationMetadata metadata) { + this.nexusOperationMetadata = metadata; + } + + public NexusOperationMetadata getNexusOperationMetadata() { + return nexusOperationMetadata; + } + /** * Set the {@code common.v1.Link}s extracted from the inbound Nexus task so they can be attached * to RPCs issued by the operation handler. diff --git a/temporal-sdk/src/main/java/io/temporal/internal/nexus/NexusOperationMetadata.java b/temporal-sdk/src/main/java/io/temporal/internal/nexus/NexusOperationMetadata.java new file mode 100644 index 0000000000..99b601fff9 --- /dev/null +++ b/temporal-sdk/src/main/java/io/temporal/internal/nexus/NexusOperationMetadata.java @@ -0,0 +1,22 @@ +package io.temporal.internal.nexus; + +import io.temporal.common.Experimental; +import java.util.Map; + +/** Container for in-flight Nexus operation metadata. */ +@Experimental +public final class NexusOperationMetadata { + public final String requestId; + public final String callbackUrl; + public final Map callbackHeaders; + + public String operationToken; + public boolean operationCompleted; + + public NexusOperationMetadata( + String requestId, String callbackUrl, Map callbackHeaders) { + this.requestId = requestId; + this.callbackUrl = callbackUrl; + this.callbackHeaders = callbackHeaders; + } +} diff --git a/temporal-sdk/src/main/java/io/temporal/nexus/TemporalNexusClientImpl.java b/temporal-sdk/src/main/java/io/temporal/nexus/TemporalNexusClientImpl.java index d63b0dd434..79169cfe78 100644 --- a/temporal-sdk/src/main/java/io/temporal/nexus/TemporalNexusClientImpl.java +++ b/temporal-sdk/src/main/java/io/temporal/nexus/TemporalNexusClientImpl.java @@ -16,6 +16,9 @@ import io.temporal.internal.client.ActivityClientInternal; import io.temporal.internal.client.NexusStartActivityResponse; import io.temporal.internal.client.NexusStartWorkflowResponse; +import io.temporal.internal.nexus.CurrentNexusOperationContext; +import io.temporal.internal.nexus.InternalNexusOperationContext; +import io.temporal.internal.nexus.NexusOperationMetadata; import io.temporal.internal.nexus.NexusStartActivityHelper; import io.temporal.internal.nexus.NexusStartWorkflowHelper; import io.temporal.internal.nexus.OperationTokenUtil; @@ -473,7 +476,14 @@ public TemporalOperationResult startActivity( private TemporalOperationResult startActivityImpl( String activityType, List args, StartActivityOptions options) { markAsyncOperationStarted(); + InternalNexusOperationContext nexusContext = CurrentNexusOperationContext.get(); try { + NexusOperationMetadata nexusOperationMetadata = + new NexusOperationMetadata( + operationStartDetails.getRequestId(), + operationStartDetails.getCallbackUrl(), + operationStartDetails.getCallbackHeaders()); + nexusContext.setNexusOperationMetadata(nexusOperationMetadata); NexusStartActivityResponse response = NexusStartActivityHelper.startActivityAndAttachLinks( operationContext, @@ -483,19 +493,12 @@ private TemporalOperationResult startActivityImpl( options, Header.empty(), request -> { - ActivityClientCallsInterceptor.CompletionCallback cb = - (request.getCallbackUrl() == null || request.getCallbackUrl().isEmpty()) - ? null - : new ActivityClientCallsInterceptor.CompletionCallback( - request.getCallbackUrl(), request.getCallbackHeaders()); ActivityClientCallsInterceptor.StartActivityInput input = new ActivityClientCallsInterceptor.StartActivityInput( request.getActivityType(), request.getArgs(), request.getOptions(), - request.getHeader(), - cb, - request.getLinks()); + request.getHeader()); // Build an ActivityClient that mirrors the surrounding WorkflowClient's options // so that the metrics-tagged scope built by the impl is preserved. setIdentity is // load-bearing because the cancel RPC reads it from clientOptions.getIdentity(). @@ -514,13 +517,13 @@ private TemporalOperationResult startActivityImpl( // because the run ID isn't known until after the start RPC returns. The operation // token returned to the Nexus caller can — and should — include it, so it's // regenerated here from the same activity ID + the run ID the start RPC produced. - String headerToken = out.getNexusOperationToken(); + String headerToken = nexusOperationMetadata.operationToken; if (headerToken == null) { throw new HandlerException( HandlerException.ErrorType.INTERNAL, "invoker did not return a Nexus operation token for activity start with callback", new IllegalStateException( - "nexusOperationToken is null on StartActivityOutput when CompletionCallback was set")); + "operationToken is null on NexusOperationMetadata after activity start")); } String returnToken; try { @@ -544,6 +547,8 @@ private TemporalOperationResult startActivityImpl( // being blocked by the guard. asyncOperationStarted.set(false); throw t; + } finally { + nexusContext.setNexusOperationMetadata(null); } } diff --git a/temporal-sdk/src/test/java/io/temporal/internal/client/RootActivityClientInvokerTest.java b/temporal-sdk/src/test/java/io/temporal/internal/client/RootActivityClientInvokerTest.java new file mode 100644 index 0000000000..884145af21 --- /dev/null +++ b/temporal-sdk/src/test/java/io/temporal/internal/client/RootActivityClientInvokerTest.java @@ -0,0 +1,139 @@ +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.Link; +import io.temporal.api.enums.v1.EventType; +import io.temporal.api.workflowservice.v1.StartActivityExecutionRequest; +import io.temporal.api.workflowservice.v1.StartActivityExecutionResponse; +import io.temporal.client.ActivityClientOptions; +import io.temporal.client.StartActivityOptions; +import io.temporal.client.WorkflowClient; +import io.temporal.common.interceptors.ActivityClientCallsInterceptor.StartActivityInput; +import io.temporal.common.interceptors.Header; +import io.temporal.internal.client.external.GenericWorkflowClient; +import io.temporal.internal.nexus.CurrentNexusOperationContext; +import io.temporal.internal.nexus.InternalNexusOperationContext; +import io.temporal.internal.nexus.NexusOperationMetadata; +import java.time.Duration; +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; +import org.junit.After; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; +import org.mockito.ArgumentCaptor; + +/** Unit tests for Nexus metadata propagation by {@link RootActivityClientInvoker}. */ +public class RootActivityClientInvokerTest { + + private static final String NAMESPACE = "test-namespace"; + + private GenericWorkflowClient genericClient; + private RootActivityClientInvoker invoker; + private InternalNexusOperationContext nexusContext; + + @Before + public void setUp() { + genericClient = mock(GenericWorkflowClient.class); + when(genericClient.startActivity(any(StartActivityExecutionRequest.class))) + .thenReturn( + StartActivityExecutionResponse.newBuilder().setRunId("activity-run-id").build()); + invoker = + new RootActivityClientInvoker( + genericClient, + ActivityClientOptions.newBuilder() + .setNamespace(NAMESPACE) + .setIdentity("test-identity") + .build()); + 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 nexusMetadataAddsCallbackLinksAndRequestId() { + Map callbackHeaders = new HashMap<>(); + callbackHeaders.put("Custom-Header", "value"); + NexusOperationMetadata metadata = + new NexusOperationMetadata( + "nexus-request-id", "http://localhost/callback", callbackHeaders); + nexusContext.setNexusOperationMetadata(metadata); + Link link = workflowEventLink(); + nexusContext.setRequestLinks(Collections.singletonList(link)); + + invoker.startActivity(newStartActivityInput()); + + ArgumentCaptor captor = + ArgumentCaptor.forClass(StartActivityExecutionRequest.class); + verify(genericClient).startActivity(captor.capture()); + StartActivityExecutionRequest request = captor.getValue(); + Assert.assertEquals("nexus-request-id", request.getRequestId()); + Assert.assertEquals(Collections.singletonList(link), request.getLinksList()); + Assert.assertEquals(1, request.getCompletionCallbacksCount()); + Assert.assertEquals( + "http://localhost/callback", request.getCompletionCallbacks(0).getNexus().getUrl()); + Assert.assertEquals( + "value", request.getCompletionCallbacks(0).getNexus().getHeaderOrThrow("custom-header")); + Assert.assertNotNull(metadata.operationToken); + Assert.assertEquals( + metadata.operationToken, + request + .getCompletionCallbacks(0) + .getNexus() + .getHeaderOrThrow(io.nexusrpc.Header.OPERATION_TOKEN.toLowerCase())); + } + + @Test + public void nexusContextWithoutMetadataStartsOrdinaryActivity() { + nexusContext.setRequestLinks(Collections.singletonList(workflowEventLink())); + + invoker.startActivity(newStartActivityInput()); + + ArgumentCaptor captor = + ArgumentCaptor.forClass(StartActivityExecutionRequest.class); + verify(genericClient).startActivity(captor.capture()); + StartActivityExecutionRequest request = captor.getValue(); + Assert.assertFalse(request.getRequestId().isEmpty()); + Assert.assertEquals(0, request.getLinksCount()); + Assert.assertEquals(0, request.getCompletionCallbacksCount()); + } + + private static StartActivityInput newStartActivityInput() { + StartActivityOptions options = + StartActivityOptions.newBuilder() + .setId("activity-id") + .setTaskQueue("test-task-queue") + .setStartToCloseTimeout(Duration.ofSeconds(10)) + .build(); + return new StartActivityInput("TestActivity", Collections.emptyList(), options, Header.empty()); + } + + private static Link workflowEventLink() { + return Link.newBuilder() + .setWorkflowEvent( + Link.WorkflowEvent.newBuilder() + .setNamespace(NAMESPACE) + .setWorkflowId("caller-workflow-id") + .setRunId("caller-run-id") + .setEventRef( + Link.WorkflowEvent.EventReference.newBuilder() + .setEventType(EventType.EVENT_TYPE_NEXUS_OPERATION_SCHEDULED))) + .build(); + } +} diff --git a/temporal-sdk/src/test/java/io/temporal/nexus/TemporalNexusClientImplTest.java b/temporal-sdk/src/test/java/io/temporal/nexus/TemporalNexusClientImplTest.java index 020124f6ff..055ff68a8f 100644 --- a/temporal-sdk/src/test/java/io/temporal/nexus/TemporalNexusClientImplTest.java +++ b/temporal-sdk/src/test/java/io/temporal/nexus/TemporalNexusClientImplTest.java @@ -1,23 +1,37 @@ package io.temporal.nexus; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.mockStatic; import static org.mockito.Mockito.when; +import static org.mockito.Mockito.withSettings; import com.uber.m3.tally.NoopScope; import io.nexusrpc.handler.HandlerException; import io.nexusrpc.handler.OperationContext; import io.nexusrpc.handler.OperationStartDetails; +import io.temporal.api.common.v1.Link; +import io.temporal.api.common.v1.WorkflowExecution; +import io.temporal.client.ActivityClient; +import io.temporal.client.ActivityClientOptions; import io.temporal.client.StartActivityOptions; import io.temporal.client.WorkflowClient; import io.temporal.client.WorkflowClientOptions; import io.temporal.client.WorkflowOptions; +import io.temporal.common.interceptors.ActivityClientCallsInterceptor; +import io.temporal.internal.client.ActivityClientInternal; +import io.temporal.internal.client.NexusStartWorkflowRequest; +import io.temporal.internal.client.NexusStartWorkflowResponse; +import io.temporal.internal.client.WorkflowClientInternal; import io.temporal.internal.nexus.CurrentNexusOperationContext; import io.temporal.internal.nexus.InternalNexusOperationContext; +import io.temporal.serviceclient.WorkflowServiceStubs; +import io.temporal.workflow.Functions; import java.time.Duration; import org.junit.After; import org.junit.Assert; import org.junit.Before; import org.junit.Test; +import org.mockito.MockedStatic; /** * Pure unit tests for {@link TemporalNexusClientImpl#markAsyncOperationStarted()} semantics. These @@ -35,14 +49,33 @@ public class TemporalNexusClientImplTest { private static final String ENDPOINT = "test-endpoint"; private TemporalNexusClientImpl client; + private MockedStatic activityClientFactory; @Before public void setUp() { - // Mock WorkflowClient so constructor doesn't need a live server. WorkflowClient workflowClient = mock(WorkflowClient.class); WorkflowClientOptions clientOptions = mock(WorkflowClientOptions.class); when(workflowClient.getOptions()).thenReturn(clientOptions); when(clientOptions.getNamespace()).thenReturn(NAMESPACE); + when(clientOptions.getIdentity()).thenReturn("test-identity"); + when(workflowClient.getWorkflowServiceStubs()).thenReturn(mock(WorkflowServiceStubs.class)); + + WorkflowClientInternal workflowClientInternal = mock(WorkflowClientInternal.class); + when(workflowClient.getInternal()).thenReturn(workflowClientInternal); + when(workflowClientInternal.startNexus( + org.mockito.ArgumentMatchers.any(NexusStartWorkflowRequest.class), + org.mockito.ArgumentMatchers.any(Functions.Proc.class))) + .thenReturn( + new NexusStartWorkflowResponse( + WorkflowExecution.newBuilder() + .setWorkflowId("workflow-id") + .setRunId("workflow-run-id") + .build(), + "workflow-operation-token")); + when(workflowClient.newWorkflowStub( + org.mockito.ArgumentMatchers.eq(BlockingWorkflow.class), + org.mockito.ArgumentMatchers.any(WorkflowOptions.class))) + .thenReturn(mock(BlockingWorkflow.class)); OperationContext operationContext = mock(OperationContext.class); when(operationContext.getService()).thenReturn("TestService"); @@ -61,11 +94,37 @@ public void setUp() { InternalNexusOperationContext internalCtx = new InternalNexusOperationContext( NAMESPACE, TASK_QUEUE, ENDPOINT, new NoopScope(), workflowClient); + internalCtx.setStartWorkflowResponseLink(Link.getDefaultInstance()); CurrentNexusOperationContext.set(internalCtx); + + ActivityClient activityClient = + mock(ActivityClient.class, withSettings().extraInterfaces(ActivityClientInternal.class)); + ActivityClientCallsInterceptor activityInvoker = mock(ActivityClientCallsInterceptor.class); + when(((ActivityClientInternal) activityClient).getInvoker()).thenReturn(activityInvoker); + when(activityInvoker.startActivity( + org.mockito.ArgumentMatchers.any( + ActivityClientCallsInterceptor.StartActivityInput.class))) + .thenAnswer( + invocation -> { + CurrentNexusOperationContext.get().getNexusOperationMetadata().operationToken = + "activity-operation-token"; + ActivityClientCallsInterceptor.StartActivityInput input = invocation.getArgument(0); + return new ActivityClientCallsInterceptor.StartActivityOutput( + input.getOptions().getId(), "activity-run-id"); + }); + activityClientFactory = mockStatic(ActivityClient.class); + activityClientFactory + .when( + () -> + ActivityClient.newInstance( + org.mockito.ArgumentMatchers.any(WorkflowServiceStubs.class), + org.mockito.ArgumentMatchers.any(ActivityClientOptions.class))) + .thenReturn(activityClient); } @After public void tearDown() { + activityClientFactory.close(); CurrentNexusOperationContext.unset(); } @@ -80,20 +139,7 @@ public void doubleStartActivity_secondCallThrowsBadRequest() { .setStartToCloseTimeout(Duration.ofSeconds(10)) .build(); - // First call: markAsyncOperationStarted() succeeds (sets flag), RPC may throw — we don't care. - try { - client.startActivity(TestActivity.class, TestActivity::doSomething, options); - } catch (HandlerException e) { - // If a HandlerException leaks out of the first call it must NOT be BAD_REQUEST - // from markAsyncOperationStarted — that would mean the flag was already set before setUp. - Assert.assertNotEquals( - "First startActivity must not fail with BAD_REQUEST (markAsyncOperationStarted guard)", - HandlerException.ErrorType.BAD_REQUEST, - e.getErrorType()); - } catch (Exception ignored) { - // Any other exception from the RPC layer is expected; the important thing is - // markAsyncOperationStarted already ran and set asyncOperationStarted = true. - } + client.startActivity(TestActivity.class, TestActivity::doSomething, options); // Second call: markAsyncOperationStarted() sees the flag and must throw BAD_REQUEST // immediately. @@ -126,17 +172,7 @@ public void doubleStartWorkflow_secondCallThrowsBadRequest() { WorkflowOptions options = WorkflowOptions.newBuilder().setWorkflowId("wf-1").setTaskQueue(TASK_QUEUE).build(); - // First call: markAsyncOperationStarted() succeeds, downstream RPC may throw — we don't care. - try { - client.startWorkflow(BlockingWorkflow.class, BlockingWorkflow::execute, "input", options); - } catch (HandlerException e) { - Assert.assertNotEquals( - "First startWorkflow must not fail with BAD_REQUEST (markAsyncOperationStarted guard)", - HandlerException.ErrorType.BAD_REQUEST, - e.getErrorType()); - } catch (Exception ignored) { - // Expected: no live server. - } + client.startWorkflow(BlockingWorkflow.class, BlockingWorkflow::execute, "input", options); // Second call must throw BAD_REQUEST immediately. HandlerException ex = @@ -168,12 +204,7 @@ public void startWorkflowThenActivity_activityThrowsBadRequest() { WorkflowOptions wfOptions = WorkflowOptions.newBuilder().setWorkflowId("wf-mixed-1").setTaskQueue(TASK_QUEUE).build(); - try { - client.startWorkflow(BlockingWorkflow.class, BlockingWorkflow::execute, "in", wfOptions); - } catch (HandlerException e) { - Assert.assertNotEquals(HandlerException.ErrorType.BAD_REQUEST, e.getErrorType()); - } catch (Exception ignored) { - } + client.startWorkflow(BlockingWorkflow.class, BlockingWorkflow::execute, "in", wfOptions); HandlerException ex = Assert.assertThrows( @@ -202,12 +233,7 @@ public void startActivityThenWorkflow_workflowThrowsBadRequest() { .setStartToCloseTimeout(Duration.ofSeconds(10)) .build(); - try { - client.startActivity(TestActivity.class, TestActivity::doSomething, actOptions); - } catch (HandlerException e) { - Assert.assertNotEquals(HandlerException.ErrorType.BAD_REQUEST, e.getErrorType()); - } catch (Exception ignored) { - } + client.startActivity(TestActivity.class, TestActivity::doSomething, actOptions); HandlerException ex = Assert.assertThrows(