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 efcc9df69..77bf2bd79 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/internal/client/ActivityClientInternal.java b/temporal-sdk/src/main/java/io/temporal/internal/client/ActivityClientInternal.java new file mode 100644 index 000000000..1652da00d --- /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 000000000..4de162640 --- /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 000000000..6971ee550 --- /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 96a9e70f5..fd6ba281a 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,9 +21,13 @@ 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.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.*; @@ -49,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()) @@ -104,6 +117,30 @@ public StartActivityOutput startActivity(StartActivityInput input) { io.temporal.api.common.v1.Header grpcHeader = HeaderUtils.toHeaderGrpc(input.getHeader(), null); request.setHeader(grpcHeader); + 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 { + nexusOperationMetadata.operationToken = + 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( + nexusOperationMetadata.callbackUrl, + nexusOperationMetadata.callbackHeaders, + nexusOperationMetadata.operationToken, + protoLinks); + request.addCompletionCallbacks(cb); + } + StartActivityExecutionResponse response; try { response = genericClient.startActivity(request.build()); @@ -120,6 +157,10 @@ public StartActivityOutput startActivity(StartActivityInput input) { throw e; } + if (nexusOperationMetadata != null && response.hasLink()) { + nexusContext.addResponseLink(response.getLink()); + } + String runId = response.getRunId().isEmpty() ? null : response.getRunId(); return new StartActivityOutput(options.getId(), runId); } 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 72987e39c..08679ca10 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/common/LinkConverter.java b/temporal-sdk/src/main/java/io/temporal/internal/common/LinkConverter.java index 1eef63af2..8845bc65e 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/common/LinkConverter.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/common/LinkConverter.java @@ -22,6 +22,8 @@ public class LinkConverter { private static final String linkPathFormat = "temporal:///namespaces/%s/workflows/%s/%s/history"; private static final String nexusOperationLinkPathFormat = "temporal:///namespaces/%s/nexus-operations/%s/%s/details"; + private static final String activityLinkPathFormat = + "temporal:///namespaces/%s/activities/%s/%s/details"; private static final String linkReferenceTypeKey = "referenceType"; private static final String linkEventIDKey = "eventID"; private static final String linkEventTypeKey = "eventType"; @@ -35,6 +37,7 @@ public class LinkConverter { Link.WorkflowEvent.getDescriptor().getFullName(); private static final String nexusOperationLinkType = Link.NexusOperation.getDescriptor().getFullName(); + private static final String activityLinkType = Link.Activity.getDescriptor().getFullName(); public static io.temporal.api.nexus.v1.Link workflowEventToNexusLink(Link.WorkflowEvent we) { try { @@ -177,6 +180,9 @@ public static io.temporal.api.nexus.v1.Link linkToNexusLink(Link commonLink) { if (commonLink.hasNexusOperation()) { return nexusOperationToNexusLink(commonLink.getNexusOperation()); } + if (commonLink.hasActivity()) { + return activityToNexusLink(commonLink.getActivity()); + } return null; } @@ -192,10 +198,79 @@ public static Link nexusLinkToLink(io.temporal.api.nexus.v1.Link nexusLink) { if (nexusOperationLinkType.equals(type)) { return nexusLinkToNexusOperation(nexusLink); } + if (activityLinkType.equals(type)) { + return nexusLinkToActivity(nexusLink); + } log.warn("ignoring unsupported nexus link type: {}", type); return null; } + public static io.temporal.api.nexus.v1.Link activityToNexusLink(Link.Activity activity) { + try { + String url = + String.format( + activityLinkPathFormat, + URLEncoder.encode(activity.getNamespace(), StandardCharsets.UTF_8.toString()), + URLEncoder.encode(activity.getActivityId(), StandardCharsets.UTF_8.toString()) + .replace("+", "%20"), + URLEncoder.encode(activity.getRunId(), StandardCharsets.UTF_8.toString())); + return io.temporal.api.nexus.v1.Link.newBuilder() + .setUrl(url) + .setType(activityLinkType) + .build(); + } catch (Exception e) { + log.error("Failed to encode activity Nexus link URL", e); + } + return null; + } + + public static Link nexusLinkToActivity(io.temporal.api.nexus.v1.Link nexusLink) { + if (!activityLinkType.equals(nexusLink.getType())) { + log.error( + "Failed to parse Nexus link URL: cannot parse link type {} to {}", + nexusLink.getType(), + activityLinkType); + return null; + } + Link.Builder link = Link.newBuilder(); + try { + URI uri = new URI(nexusLink.getUrl()); + if (!"temporal".equals(uri.getScheme())) { + log.error("Failed to parse Nexus link URL: invalid scheme: {}", uri.getScheme()); + return null; + } + StringTokenizer st = new StringTokenizer(uri.getRawPath(), "/"); + if (!st.hasMoreTokens() || !st.nextToken().equals("namespaces")) { + log.error("Failed to parse Nexus link URL: invalid path: {}", uri.getRawPath()); + return null; + } + String namespace = URLDecoder.decode(st.nextToken(), StandardCharsets.UTF_8.toString()); + if (!st.hasMoreTokens() || !st.nextToken().equals("activities")) { + log.error("Failed to parse Nexus link URL: invalid path: {}", uri.getRawPath()); + return null; + } + String activityId = URLDecoder.decode(st.nextToken(), StandardCharsets.UTF_8.toString()); + if (!st.hasMoreTokens()) { + log.error("Failed to parse Nexus link URL: invalid path: {}", uri.getRawPath()); + return null; + } + String runId = URLDecoder.decode(st.nextToken(), StandardCharsets.UTF_8.toString()); + if (!st.hasMoreTokens() || !st.nextToken().equals("details")) { + log.error("Failed to parse Nexus link URL: invalid path: {}", uri.getRawPath()); + return null; + } + link.setActivity( + Link.Activity.newBuilder() + .setNamespace(namespace) + .setActivityId(activityId) + .setRunId(runId)); + } catch (Exception e) { + log.error("Failed to parse activity Nexus link URL", e); + return null; + } + return link.build(); + } + public static io.temporal.api.nexus.v1.Link nexusOperationToNexusLink(Link.NexusOperation no) { try { String url = 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 030868385..d712627e1 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 000000000..99b601fff --- /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/internal/nexus/NexusStartActivityHelper.java b/temporal-sdk/src/main/java/io/temporal/internal/nexus/NexusStartActivityHelper.java new file mode 100644 index 000000000..9cf346c8f --- /dev/null +++ b/temporal-sdk/src/main/java/io/temporal/internal/nexus/NexusStartActivityHelper.java @@ -0,0 +1,57 @@ +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. */ +@Experimental +public class NexusStartActivityHelper { + + /** + * Starts an activity via the provided invoker function and returns the response. The root + * activity invoker records the link from {@code StartActivityExecutionResponse}; the Nexus task + * handler attaches that link to the operation response. + * + * @param ctx the operation context + * @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); + + return invoker.apply(nexusRequest); + } + + 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 4bd5635e9..a5f7ec997 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 11aa57a81..952d784be 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 737a84aad..447eb04f1 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 000000000..679403381 --- /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 4eed1fe35..58bf25bb6 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 43e29e18f..d762f339c 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,38 @@ 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.api.common.v1.Payload; +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.context.ContextPropagator; +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.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; +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.HashMap; +import java.util.List; +import java.util.Map; import java.util.Objects; import java.util.concurrent.atomic.AtomicBoolean; @@ -244,13 +266,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 +281,300 @@ 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(); + InternalNexusOperationContext nexusContext = CurrentNexusOperationContext.get(); + try { + NexusOperationMetadata nexusOperationMetadata = + new NexusOperationMetadata( + operationStartDetails.getRequestId(), + operationStartDetails.getCallbackUrl(), + operationStartDetails.getCallbackHeaders()); + nexusContext.setNexusOperationMetadata(nexusOperationMetadata); + NexusStartActivityResponse response = + NexusStartActivityHelper.startActivityAndAttachLinks( + operationContext, + operationStartDetails, + activityType, + args, + options, + propagatedHeader(), + request -> { + ActivityClientCallsInterceptor.StartActivityInput input = + new ActivityClientCallsInterceptor.StartActivityInput( + request.getActivityType(), + request.getArgs(), + request.getOptions(), + 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(). + 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 = 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( + "operationToken is null on NexusOperationMetadata after activity start")); + } + 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; + } finally { + nexusContext.setNexusOperationMetadata(null); + } + } + + 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.")); + } + } + + private Header propagatedHeader() { + List propagators = client.getOptions().getContextPropagators(); + if (propagators.isEmpty()) { + return Header.empty(); + } + Map result = new HashMap<>(); + for (ContextPropagator propagator : propagators) { + result.putAll(propagator.serializeContext(propagator.getCurrentContext())); + } + return new Header(result); + } } 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 6a01d11fc..584d70222 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/client/RootActivityClientInvokerTest.java b/temporal-sdk/src/test/java/io/temporal/internal/client/RootActivityClientInvokerTest.java new file mode 100644 index 000000000..48af446a4 --- /dev/null +++ b/temporal-sdk/src/test/java/io/temporal/internal/client/RootActivityClientInvokerTest.java @@ -0,0 +1,154 @@ +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") + .setLink(activityLink()) + .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())); + Assert.assertEquals(Collections.singletonList(activityLink()), nexusContext.getResponseLinks()); + } + + @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()); + Assert.assertTrue(nexusContext.getResponseLinks().isEmpty()); + } + + 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(); + } + + private static Link activityLink() { + return Link.newBuilder() + .setActivity( + Link.Activity.newBuilder() + .setNamespace(NAMESPACE) + .setActivityId("activity-id") + .setRunId("activity-run-id")) + .build(); + } +} diff --git a/temporal-sdk/src/test/java/io/temporal/internal/common/LinkConverterTest.java b/temporal-sdk/src/test/java/io/temporal/internal/common/LinkConverterTest.java index 9024cbff6..34868a247 100644 --- a/temporal-sdk/src/test/java/io/temporal/internal/common/LinkConverterTest.java +++ b/temporal-sdk/src/test/java/io/temporal/internal/common/LinkConverterTest.java @@ -1,6 +1,8 @@ package io.temporal.internal.common; +import static io.temporal.internal.common.LinkConverter.activityToNexusLink; import static io.temporal.internal.common.LinkConverter.linkToNexusLink; +import static io.temporal.internal.common.LinkConverter.nexusLinkToActivity; import static io.temporal.internal.common.LinkConverter.nexusLinkToLink; import static io.temporal.internal.common.LinkConverter.nexusLinkToNexusOperation; import static io.temporal.internal.common.LinkConverter.nexusLinkToWorkflowEvent; @@ -466,6 +468,57 @@ public void testConvertNexusToNexusOperation_InvalidPathMissingDetails() { assertNull(nexusLinkToNexusOperation(input)); } + @Test + public void testConvertActivityToNexus_Valid() { + Link.Activity input = + Link.Activity.newBuilder() + .setNamespace("ns") + .setActivityId("act id/with+characters") + .setRunId("run-id") + .build(); + + io.temporal.api.nexus.v1.Link expected = + io.temporal.api.nexus.v1.Link.newBuilder() + .setUrl( + "temporal:///namespaces/ns/activities/act%20id%2Fwith%2Bcharacters/run-id/details") + .setType("temporal.api.common.v1.Link.Activity") + .build(); + + assertEquals(expected, activityToNexusLink(input)); + } + + @Test + public void testConvertNexusToActivity_Valid() { + io.temporal.api.nexus.v1.Link input = + io.temporal.api.nexus.v1.Link.newBuilder() + .setUrl( + "temporal:///namespaces/ns/activities/act%20id%2Fwith%2Bcharacters/run-id/details") + .setType("temporal.api.common.v1.Link.Activity") + .build(); + + Link expected = + Link.newBuilder() + .setActivity( + Link.Activity.newBuilder() + .setNamespace("ns") + .setActivityId("act id/with+characters") + .setRunId("run-id")) + .build(); + + assertEquals(expected, nexusLinkToActivity(input)); + } + + @Test + public void testConvertNexusToActivity_InvalidPath() { + io.temporal.api.nexus.v1.Link input = + io.temporal.api.nexus.v1.Link.newBuilder() + .setUrl("temporal:///namespaces/ns/activities/act-id/run-id") + .setType("temporal.api.common.v1.Link.Activity") + .build(); + + assertNull(nexusLinkToActivity(input)); + } + @Test public void testNexusLinkToLink_WorkflowEventRoundTrip() { Link.WorkflowEvent we = @@ -507,6 +560,19 @@ public void testNexusLinkToLink_NexusOperation() { assertEquals(expected, nexusLinkToLink(nexusLink)); } + @Test + public void testNexusLinkToLink_ActivityRoundTrip() { + Link.Activity activity = + Link.Activity.newBuilder() + .setNamespace("ns") + .setActivityId("act-id") + .setRunId("run-id") + .build(); + + io.temporal.api.nexus.v1.Link nexusLink = activityToNexusLink(activity); + assertEquals(Link.newBuilder().setActivity(activity).build(), nexusLinkToLink(nexusLink)); + } + @Test public void testNexusLinkToLink_UnknownType() { io.temporal.api.nexus.v1.Link nexusLink = @@ -550,6 +616,20 @@ public void testLinkToNexusLink_NexusOperation() { assertEquals(nexusOperationToNexusLink(no), actual); } + @Test + public void testLinkToNexusLink_Activity() { + Link.Activity activity = + Link.Activity.newBuilder() + .setNamespace("ns") + .setActivityId("act-id") + .setRunId("run-id") + .build(); + + io.temporal.api.nexus.v1.Link actual = + linkToNexusLink(Link.newBuilder().setActivity(activity).build()); + assertEquals(activityToNexusLink(activity), actual); + } + @Test public void testLinkToNexusLink_Empty() { assertNull(linkToNexusLink(Link.newBuilder().build())); 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 1f22fe8c2..ad55f7a3a 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 000000000..2433db132 --- /dev/null +++ b/temporal-sdk/src/test/java/io/temporal/nexus/TemporalNexusClientImplTest.java @@ -0,0 +1,303 @@ +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.Payload; +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.context.ContextPropagator; +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 java.util.Collections; +import java.util.concurrent.atomic.AtomicReference; +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 + * 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; + private MockedStatic activityClientFactory; + private AtomicReference activityInput; + + @Before + public void setUp() { + 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"); + Payload propagatedPayload = Payload.newBuilder().build(); + ContextPropagator contextPropagator = mock(ContextPropagator.class); + when(contextPropagator.getCurrentContext()).thenReturn("test-context"); + when(contextPropagator.serializeContext("test-context")) + .thenReturn(Collections.singletonMap("propagated-key", propagatedPayload)); + when(clientOptions.getContextPropagators()) + .thenReturn(Collections.singletonList(contextPropagator)); + 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"); + 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); + internalCtx.setStartWorkflowResponseLink(Link.getDefaultInstance()); + CurrentNexusOperationContext.set(internalCtx); + + ActivityClient activityClient = + mock(ActivityClient.class, withSettings().extraInterfaces(ActivityClientInternal.class)); + ActivityClientCallsInterceptor activityInvoker = mock(ActivityClientCallsInterceptor.class); + activityInput = new AtomicReference<>(); + when(((ActivityClientInternal) activityClient).getInvoker()).thenReturn(activityInvoker); + when(activityInvoker.startActivity( + org.mockito.ArgumentMatchers.any( + ActivityClientCallsInterceptor.StartActivityInput.class))) + .thenAnswer( + invocation -> { + activityInput.set(invocation.getArgument(0)); + 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(); + } + + // ---------- Activity double-start ---------- + + @Test + public void startActivity_propagatesWorkflowClientContext() { + StartActivityOptions options = + StartActivityOptions.newBuilder() + .setId("act-context") + .setTaskQueue(TASK_QUEUE) + .setStartToCloseTimeout(Duration.ofSeconds(10)) + .build(); + + client.startActivity(TestActivity.class, TestActivity::doSomething, options); + + Assert.assertNotNull(activityInput.get()); + Assert.assertTrue(activityInput.get().getHeader().getValues().containsKey("propagated-key")); + } + + @Test + public void doubleStartActivity_secondCallThrowsBadRequest() { + StartActivityOptions options = + StartActivityOptions.newBuilder() + .setId("act-1") + .setTaskQueue(TASK_QUEUE) + .setStartToCloseTimeout(Duration.ofSeconds(10)) + .build(); + + client.startActivity(TestActivity.class, TestActivity::doSomething, options); + + // 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(); + + client.startWorkflow(BlockingWorkflow.class, BlockingWorkflow::execute, "input", options); + + // 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(); + + client.startWorkflow(BlockingWorkflow.class, BlockingWorkflow::execute, "in", wfOptions); + + 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(); + + client.startActivity(TestActivity.class, TestActivity::doSomething, actOptions); + + 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 000000000..455ea9f3d --- /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 000000000..45a2b7879 --- /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. + } + }; + } + } +}