diff --git a/temporal-sdk/src/main/java/io/temporal/internal/client/RootWorkflowClientInvoker.java b/temporal-sdk/src/main/java/io/temporal/internal/client/RootWorkflowClientInvoker.java index 510471f2e..5299b9be4 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/client/RootWorkflowClientInvoker.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/client/RootWorkflowClientInvoker.java @@ -5,6 +5,7 @@ import static io.temporal.internal.common.HeaderUtils.intoPayloadMap; import static io.temporal.internal.common.WorkflowExecutionUtils.makeUserMetaData; +import com.fasterxml.jackson.core.JsonProcessingException; import com.google.common.collect.Iterators; import io.grpc.Deadline; import io.grpc.Status; @@ -23,7 +24,11 @@ import io.temporal.common.interceptors.WorkflowClientCallsInterceptor; import io.temporal.internal.client.external.GenericWorkflowClient; import io.temporal.internal.common.HeaderUtils; +import io.temporal.internal.common.InternalUtils; import io.temporal.internal.nexus.CurrentNexusOperationContext; +import io.temporal.internal.nexus.InternalNexusOperationContext; +import io.temporal.internal.nexus.NexusOperationMetadata; +import io.temporal.internal.nexus.OperationTokenUtil; import io.temporal.internal.worker.WorkerVersioningProtoUtils; import io.temporal.payload.context.WorkflowSerializationContext; import io.temporal.serviceclient.StatusUtils; @@ -473,6 +478,19 @@ public WorkflowUpdateHandle startUpdate(StartUpdateInput input) { } } while (updateNotYetDurable(input, result)); + // If triggered by a Nexus Operation, set necessary fields- link, result + if (CurrentNexusOperationContext.isNexusContext()) { + NexusOperationMetadata nexusOperationMetadata = + CurrentNexusOperationContext.get().getNexusOperationMetadata(); + if (nexusOperationMetadata != null) { + if (result.hasLink()) { + // add forward links for caller->handler + CurrentNexusOperationContext.get().addResponseLink(result.getLink()); + } + nexusOperationMetadata.operationCompleted = result.hasOutcome(); + } + } + return toUpdateHandle(input, result, dataConverterWithWorkflowContext); } @@ -495,14 +513,49 @@ private UpdateWorkflowExecutionRequest toUpdateWorkflowExecutionRequest( .setName(input.getUpdateName()); inputArgs.ifPresent(updateInput::setArgs); - Request request = + Request.Builder requestBuilder = Request.newBuilder() .setMeta( Meta.newBuilder() .setUpdateId(input.getUpdateId()) .setIdentity(clientOptions.getIdentity())) - .setInput(updateInput) - .build(); + .setInput(updateInput); + + // If this update is being issued from inside a Nexus operation handler via + // TemporalNexusClientImpl.startWorkflowUpdate, set the fields the server needs + // to deliver the Nexus completion callback. + if (CurrentNexusOperationContext.isNexusContext()) { + InternalNexusOperationContext nexusContext = CurrentNexusOperationContext.get(); + // already in a Nexus operation context, dont need to check nexusContext again + NexusOperationMetadata nexusOperationMetadata = nexusContext.getNexusOperationMetadata(); + if (nexusOperationMetadata != null) { + try { + nexusOperationMetadata.operationToken = + OperationTokenUtil.generateWorkflowUpdateOperationToken( + clientOptions.getNamespace(), + input.getWorkflowExecution().getWorkflowId(), + input.getWorkflowExecution().getRunId(), + input.getUpdateId()); + } catch (JsonProcessingException e) { + throw new IllegalStateException("failed to generate update operation token", e); + } + List requestLinks = nexusContext.getRequestLinks(); + requestBuilder + .setRequestId(nexusOperationMetadata.requestId) + .addCompletionCallbacks( + InternalUtils.buildNexusCallback( + nexusOperationMetadata.callbackHeaders, + nexusOperationMetadata.callbackUrl, + nexusOperationMetadata.operationToken, + requestLinks)) + .addAllLinks(requestLinks); + } else { + log.error("unexpected error fetching nexusOperationMetadata"); + // maybe throw an exception but should never really happen as its all sdk + } + } + + Request request = requestBuilder.build(); return UpdateWorkflowExecutionRequest.newBuilder() .setNamespace(clientOptions.getNamespace()) 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 0886802de..0a859c48a 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 @@ -147,6 +147,31 @@ public static NexusWorkflowStarter createNexusBoundStub( return new NexusWorkflowStarter(stub.newInstance(nexusWorkflowOptions.build()), operationToken); } + /** Helper to build a Nexus Callback from the provided input. */ + public static Callback buildNexusCallback( + Map callbackHeaders, + String callbackUrl, + String operationToken, + List links) { + Map headers = + callbackHeaders.entrySet().stream() + .collect( + Collectors.toMap( + (k) -> k.getKey().toLowerCase(), + Map.Entry::getValue, + (a, b) -> a, + TreeMap::new)); + headers.put(Header.OPERATION_TOKEN.toLowerCase(), operationToken); + Callback.Builder cbBuilder = + Callback.newBuilder() + .setNexus( + Callback.Nexus.newBuilder().setUrl(callbackUrl).putAllHeader(headers).build()); + if (links != null) { + cbBuilder.addAllLinks(links); + } + return cbBuilder.build(); + } + /** Check the method name for reserved prefixes or names. */ public static void checkMethodName(POJOWorkflowMethodMetadata methodMetadata) { if (methodMetadata.getName().startsWith(TEMPORAL_RESERVED_PREFIX)) { 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..21294e010 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 @@ -19,6 +19,7 @@ public class LinkConverter { private static final Logger log = LoggerFactory.getLogger(LinkConverter.class); + private static final String temporalUrlScheme = "temporal"; 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"; @@ -35,6 +36,7 @@ public class LinkConverter { Link.WorkflowEvent.getDescriptor().getFullName(); private static final String nexusOperationLinkType = Link.NexusOperation.getDescriptor().getFullName(); + private static final String workflowLinkType = Link.Workflow.getDescriptor().getFullName(); public static io.temporal.api.nexus.v1.Link workflowEventToNexusLink(Link.WorkflowEvent we) { try { @@ -93,6 +95,24 @@ public static io.temporal.api.nexus.v1.Link workflowEventToNexusLink(Link.Workfl return null; } + public static io.temporal.api.nexus.v1.Link workflowLinkToNexusLink(Link.Workflow w) { + try { + String namespace = URLEncoder.encode(w.getNamespace(), StandardCharsets.UTF_8.toString()); + String workflowId = + URLEncoder.encode(w.getWorkflowId(), StandardCharsets.UTF_8.toString()) + .replace("+", "%20"); // handle workflowIds supporting spaces + String runId = URLEncoder.encode(w.getRunId(), StandardCharsets.UTF_8.toString()); + String url = String.format(linkPathFormat, namespace, workflowId, runId); + return io.temporal.api.nexus.v1.Link.newBuilder() + .setUrl(url) + .setType(workflowLinkType) + .build(); + } catch (Exception e) { + log.error("Failed to encode Nexus link URL", e); + return null; + } + } + public static Link nexusLinkToWorkflowEvent(io.temporal.api.nexus.v1.Link nexusLink) { Link.Builder link = Link.newBuilder(); try { @@ -166,6 +186,44 @@ public static Link nexusLinkToWorkflowEvent(io.temporal.api.nexus.v1.Link nexusL return link.build(); } + public static Link nexusLinkToWorkflowLink(io.temporal.api.nexus.v1.Link nexusLink) { + Link.Builder link = Link.newBuilder(); + try { + URI uri = new URI(nexusLink.getUrl()); + log.debug("Parsing nexus link URL: {}", uri.getRawPath()); + if (!uri.getScheme().equals(temporalUrlScheme)) { + log.error("Failed to parse Nexus link URL: invalid scheme: {}", uri.getScheme()); + return null; + } + StringTokenizer st = new StringTokenizer(uri.getRawPath(), "/"); + // maybe add constants for "namespaces", "workflows" too + if (!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.nextToken().equals("workflows")) { + log.error("Failed to parse Nexus link URL: invalid path: {}", uri.getRawPath()); + return null; + } + String workflowID = 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()); + link.setWorkflow( + Link.Workflow.newBuilder() + .setNamespace(namespace) + .setWorkflowId(workflowID) + .setRunId(runID)); + } catch (Exception e) { + log.error("Failed to parse Nexus link URL", e); + return null; + } + return link.build(); + } + /** * Dispatches on the oneof variant of {@code commonLink} and converts to the matching {@link * io.temporal.api.nexus.v1.Link}. Returns {@code null} if no variant is set or encoding fails. @@ -177,6 +235,9 @@ public static io.temporal.api.nexus.v1.Link linkToNexusLink(Link commonLink) { if (commonLink.hasNexusOperation()) { return nexusOperationToNexusLink(commonLink.getNexusOperation()); } + if (commonLink.hasWorkflow()) { + return workflowLinkToNexusLink(commonLink.getWorkflow()); + } return null; } @@ -192,6 +253,9 @@ public static Link nexusLinkToLink(io.temporal.api.nexus.v1.Link nexusLink) { if (nexusOperationLinkType.equals(type)) { return nexusLinkToNexusOperation(nexusLink); } + if (workflowLinkType.equals(type)) { + return nexusLinkToWorkflowLink(nexusLink); + } log.warn("ignoring unsupported nexus link type: {}", type); return null; } 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..8fd807439 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 @@ -38,6 +38,21 @@ public class InternalNexusOperationContext { private final Object responseLinksLock = new Object(); private final List responseLinks = new ArrayList<>(); + private NexusOperationMetadata nexusOperationMetadata; + + /** + * Set the Nexus operation metadata + * + * @param metadata {@link NexusOperationMetadata} to be set + */ + public void setNexusOperationMetadata(NexusOperationMetadata metadata) { + this.nexusOperationMetadata = metadata; + } + + public NexusOperationMetadata getNexusOperationMetadata() { + return nexusOperationMetadata; + } + public InternalNexusOperationContext( String namespace, String taskQueue, 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..50ebcb935 --- /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 an 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/OperationToken.java b/temporal-sdk/src/main/java/io/temporal/internal/nexus/OperationToken.java index 4bd5635e9..47a8217e7 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 @@ -18,22 +18,49 @@ public class OperationToken { @JsonProperty("wid") private final String workflowId; + @JsonProperty("rid") + @JsonInclude(JsonInclude.Include.NON_NULL) + // only set for updates and activities + private final String runId; + + @JsonProperty("uid") + @JsonInclude(JsonInclude.Include.NON_NULL) + // only set for updates + private final String updateId; + public OperationToken( @JsonProperty("t") Integer type, @JsonProperty("ns") String namespace, @JsonProperty("wid") String workflowId, + @JsonProperty("rid") String runId, + @JsonProperty("uid") String updateId, @JsonProperty("v") Integer version) { this.type = OperationTokenType.fromValue(type); this.namespace = namespace; this.workflowId = workflowId; + this.runId = runId; + this.updateId = updateId; this.version = version; } + /** Generate a token for a workflow run operation */ public OperationToken(OperationTokenType type, String namespace, String workflowId) { this.type = type; this.namespace = namespace; this.workflowId = workflowId; this.version = null; + this.runId = null; + this.updateId = null; + } + + /** Generate a token for a workflow update operation */ + public OperationToken(String namespace, String workflowId, String runId, String updateId) { + this.type = OperationTokenType.WORKFLOW_UPDATE; + this.namespace = namespace; + this.workflowId = workflowId; + this.runId = runId; + this.updateId = updateId; + this.version = null; } public Integer getVersion() { @@ -51,4 +78,12 @@ public String getNamespace() { public String getWorkflowId() { return workflowId; } + + public String getUpdateId() { + return updateId; + } + + 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..9a93111ab 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,8 @@ public enum OperationTokenType { UNKNOWN(0), - WORKFLOW_RUN(1); + WORKFLOW_RUN(1), + WORKFLOW_UPDATE(3); 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..c24662728 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,6 +31,9 @@ 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.getNamespace())) { + throw new IllegalArgumentException("Invalid operation token: missing namespace(ns)"); + } if (Strings.isNullOrEmpty(token.getWorkflowId())) { throw new IllegalArgumentException("Invalid operation token: missing workflow ID (wid)"); } @@ -52,6 +55,25 @@ public static OperationToken loadWorkflowRunOperationToken(String operationToken return token; } + /** + * Load a workflow update operation token, asserting that the token type is {@link + * OperationTokenType#WORKFLOW_UPDATE}. + * + * @throws IllegalArgumentException if the operation token is invalid or not a workflow update + * token + */ + public static OperationToken loadWorkflowUpdateOperationToken(String operationToken) { + OperationToken token = loadOperationToken(operationToken); + if (!token.getType().equals(OperationTokenType.WORKFLOW_UPDATE)) { + throw new IllegalArgumentException( + "Invalid workflow update token: incorrect operation token type: " + token.getType()); + } + if (Strings.isNullOrEmpty(token.getUpdateId())) { + throw new IllegalArgumentException("Invalid workflow update token: missing update ID (uid)"); + } + return token; + } + /** * Extract the workflow ID from a workflow run operation token. * @@ -70,5 +92,24 @@ public static String generateWorkflowRunOperationToken(String workflowId, String return encoder.encodeToString(json.getBytes()); } + /** Generate a workflow update operation token from namespace, workflowId, runId, updateId */ + public static String generateWorkflowUpdateOperationToken( + String namespace, String workflowId, String runId, String updateId) + throws JsonProcessingException { + if (Strings.isNullOrEmpty(namespace)) { + throw new IllegalArgumentException("Invalid workflow update token: missing namespace(ns)"); + } + if (Strings.isNullOrEmpty(workflowId)) { + throw new IllegalArgumentException( + "Invalid workflow update token: missing workflow ID (wid)"); + } + if (Strings.isNullOrEmpty(updateId)) { + throw new IllegalArgumentException("Invalid workflow update token: missing update ID (uid)"); + } + runId = Strings.emptyToNull(runId); // empty runId is allowed but should not be serialized + String json = ow.writeValueAsString(new OperationToken(namespace, workflowId, runId, updateId)); + return encoder.encodeToString(json.getBytes()); + } + private OperationTokenUtil() {} } diff --git a/temporal-sdk/src/main/java/io/temporal/nexus/CancelUpdateWorkflowInput.java b/temporal-sdk/src/main/java/io/temporal/nexus/CancelUpdateWorkflowInput.java new file mode 100644 index 000000000..3726693e0 --- /dev/null +++ b/temporal-sdk/src/main/java/io/temporal/nexus/CancelUpdateWorkflowInput.java @@ -0,0 +1,38 @@ +package io.temporal.nexus; + +import com.google.common.base.Strings; +import io.temporal.common.Experimental; +import java.util.Objects; + +/** + * Input to {@link TemporalOperationHandler#cancelUpdateWorkflow} describing the workflow update to + * cancel. + */ +@Experimental +public final class CancelUpdateWorkflowInput { + + private final String workflowId; + private final String runId; + private final String updateId; + + public CancelUpdateWorkflowInput(String workflowId, String runId, String updateId) { + this.workflowId = Objects.requireNonNull(workflowId); + this.runId = Strings.nullToEmpty(runId); + this.updateId = Objects.requireNonNull(updateId); + } + + /** Returns the workflow ID extracted from the operation token. */ + public String getWorkflowId() { + return workflowId; + } + + /** Returns the run ID extracted from the operation token, or empty if not present. */ + public String getRunId() { + return runId; + } + + /** Returns the update ID extracted from the operation token. */ + public String getUpdateId() { + return updateId; + } +} 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..88fafbacb 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,7 @@ package io.temporal.nexus; +import io.nexusrpc.OperationException; +import io.temporal.client.UpdateOptions; import io.temporal.client.WorkflowClient; import io.temporal.client.WorkflowOptions; import io.temporal.common.Experimental; @@ -507,4 +509,445 @@ TemporalOperationResult startWorkflow( Type resultType, WorkflowOptions options, Object... args); + + // draft-review: all these combinations are for 1-1 compatibility - check if they can be trimmed + /** + * Starts a workflow update on an existing workflow as a Nexus operation. The result is delivered + * asynchronously via the Nexus completion callback, unless the update RPC comes back already + * completed (e.g. a retried request, or a request that failed validation), in which case the + * result (or failure) is returned synchronously. + * + *

{@code updateMethod} must be an unbound method reference to a method on {@code + * workflowClass} (as opposed to {@link #startWorkflow}, which creates a new workflow, this + * targets the existing workflow identified by {@code workflowId}). + * + *

{@code options}' {@code waitForStage} has to be set to {@code WaitForStage = ACCEPTED} as + * Nexus Operations only support async Update requests. If not, the operation will be marked as + * failed. If {@code options} does not set an update ID, it defaults to the Nexus request ID which + * is consistent with other SDKs usage. + * + *

A Nexus callback URL is required for this operation; if the caller did not provide one, this + * method throws a {@code BAD_REQUEST} {@code HandlerException}. + * + *

Example: + * + * 

{@code
+   * client.startWorkflowUpdate(
+   *     MyWorkflow.class, input.getWorkflowId(),
+   *     MyWorkflow::myUpdate, input.getArg(),
+   *     UpdateOptions.newBuilder(String.class)
+   *         .setUpdateName("myUpdate")
+   *         .setWaitForStage(WorkflowUpdateStage.ACCEPTED)
+   *         .build())
+   * }
+ * + * @param workflowClass the workflow interface class + * @param workflowId the ID of the existing workflow to update + * @param updateMethod unbound method reference to the update method + * @param options update options (must include the update result class) + * @param the workflow interface type + * @param the update return type + * @return a {@link TemporalOperationResult}; sync if the update already completed, async + * (carrying the update-workflow operation token) otherwise + */ + TemporalOperationResult startWorkflowUpdate( + Class workflowClass, + String workflowId, + Functions.Func1 updateMethod, + UpdateOptions options) + throws OperationException; + + /** + * Starts a one-argument workflow update on an existing workflow as a Nexus operation. See {@link + * #startWorkflowUpdate(Class, String, Functions.Func1, UpdateOptions)} for the full behavior + * contract. + * + * @param workflowClass the workflow interface class + * @param workflowId the ID of the existing workflow to update + * @param updateMethod unbound method reference to the update method + * @param arg1 first update argument + * @param options update options (must include the update result class) + * @param the workflow interface type + * @param the type of the first update argument + * @param the update return type + * @return a {@link TemporalOperationResult}; sync if the update already completed, async + * (carrying the update-workflow operation token) otherwise + */ + TemporalOperationResult startWorkflowUpdate( + Class workflowClass, + String workflowId, + Functions.Func2 updateMethod, + A1 arg1, + UpdateOptions options) + throws OperationException; + + /** + * Starts a two-argument workflow update on an existing workflow as a Nexus operation. See {@link + * #startWorkflowUpdate(Class, String, Functions.Func1, UpdateOptions)} for the full behavior + * contract. + * + * @param workflowClass the workflow interface class + * @param workflowId the ID of the existing workflow to update + * @param updateMethod unbound method reference to the update method + * @param arg1 first update argument + * @param arg2 second update argument + * @param options update options (must include the update result class) + * @param the workflow interface type + * @param the type of the first update argument + * @param the type of the second update argument + * @param the update return type + * @return a {@link TemporalOperationResult}; sync if the update already completed, async + * (carrying the update-workflow operation token) otherwise + */ + TemporalOperationResult startWorkflowUpdate( + Class workflowClass, + String workflowId, + Functions.Func3 updateMethod, + A1 arg1, + A2 arg2, + UpdateOptions options) + throws OperationException; + + /** + * Starts a three-argument workflow update on an existing workflow as a Nexus operation. See + * {@link #startWorkflowUpdate(Class, String, Functions.Func1, UpdateOptions)} for the full + * behavior contract. + * + * @param workflowClass the workflow interface class + * @param workflowId the ID of the existing workflow to update + * @param updateMethod unbound method reference to the update method + * @param arg1 first update argument + * @param arg2 second update argument + * @param arg3 third update argument + * @param options update options (must include the update result class) + * @param the workflow interface type + * @param the type of the first update argument + * @param the type of the second update argument + * @param the type of the third update argument + * @param the update return type + * @return a {@link TemporalOperationResult}; sync if the update already completed, async + * (carrying the update-workflow operation token) otherwise + */ + TemporalOperationResult startWorkflowUpdate( + Class workflowClass, + String workflowId, + Functions.Func4 updateMethod, + A1 arg1, + A2 arg2, + A3 arg3, + UpdateOptions options) + throws OperationException; + + /** + * Starts a four-argument workflow update on an existing workflow as a Nexus operation. See {@link + * #startWorkflowUpdate(Class, String, Functions.Func1, UpdateOptions)} for the full behavior + * contract. + * + * @param workflowClass the workflow interface class + * @param workflowId the ID of the existing workflow to update + * @param updateMethod unbound method reference to the update method + * @param arg1 first update argument + * @param arg2 second update argument + * @param arg3 third update argument + * @param arg4 fourth update argument + * @param options update options (must include the update result class) + * @param the workflow interface type + * @param the type of the first update argument + * @param the type of the second update argument + * @param the type of the third update argument + * @param the type of the fourth update argument + * @param the update return type + * @return a {@link TemporalOperationResult}; sync if the update already completed, async + * (carrying the update-workflow operation token) otherwise + */ + TemporalOperationResult startWorkflowUpdate( + Class workflowClass, + String workflowId, + Functions.Func5 updateMethod, + A1 arg1, + A2 arg2, + A3 arg3, + A4 arg4, + UpdateOptions options) + throws OperationException; + + /** + * Starts a five-argument workflow update on an existing workflow as a Nexus operation. See {@link + * #startWorkflowUpdate(Class, String, Functions.Func1, UpdateOptions)} for the full behavior + * contract. + * + * @param workflowClass the workflow interface class + * @param workflowId the ID of the existing workflow to update + * @param updateMethod unbound method reference to the update method + * @param arg1 first update argument + * @param arg2 second update argument + * @param arg3 third update argument + * @param arg4 fourth update argument + * @param arg5 fifth update argument + * @param options update options (must include the update result class) + * @param the workflow interface type + * @param the type of the first update argument + * @param the type of the second update argument + * @param the type of the third update argument + * @param the type of the fourth update argument + * @param the type of the fifth update argument + * @param the update return type + * @return a {@link TemporalOperationResult}; sync if the update already completed, async + * (carrying the update-workflow operation token) otherwise + */ + TemporalOperationResult startWorkflowUpdate( + Class workflowClass, + String workflowId, + Functions.Func6 updateMethod, + A1 arg1, + A2 arg2, + A3 arg3, + A4 arg4, + A5 arg5, + UpdateOptions options) + throws OperationException; + + /** + * Starts a six-argument workflow update on an existing workflow as a Nexus operation. See {@link + * #startWorkflowUpdate(Class, String, Functions.Func1, UpdateOptions)} for the full behavior + * contract. + * + * @param workflowClass the workflow interface class + * @param workflowId the ID of the existing workflow to update + * @param updateMethod unbound method reference to the update method + * @param arg1 first update argument + * @param arg2 second update argument + * @param arg3 third update argument + * @param arg4 fourth update argument + * @param arg5 fifth update argument + * @param arg6 sixth update argument + * @param options update options (must include the update result class) + * @param the workflow interface type + * @param the type of the first update argument + * @param the type of the second update argument + * @param the type of the third update argument + * @param the type of the fourth update argument + * @param the type of the fifth update argument + * @param the type of the sixth update argument + * @param the update return type + * @return a {@link TemporalOperationResult}; sync if the update already completed, async + * (carrying the update-workflow operation token) otherwise + */ + TemporalOperationResult startWorkflowUpdate( + Class workflowClass, + String workflowId, + Functions.Func7 updateMethod, + A1 arg1, + A2 arg2, + A3 arg3, + A4 arg4, + A5 arg5, + A6 arg6, + UpdateOptions options) + throws OperationException; + + // draft-review: check if all these are also necessary to preserve 1:1 compatibility + + /** + * Starts a zero-argument workflow update with no return value on an existing workflow as a Nexus + * operation. See {@link #startWorkflowUpdate(Class, String, Functions.Func1, UpdateOptions)} for + * the full behavior contract. + * + * @param workflowClass the workflow interface class + * @param workflowId the ID of the existing workflow to update + * @param updateMethod unbound method reference to the update method + * @param options update options + * @param the workflow interface type + * @return a {@link TemporalOperationResult}; sync if the update already completed, async + * (carrying the update-workflow operation token) otherwise + */ + TemporalOperationResult startWorkflowUpdate( + Class workflowClass, + String workflowId, + Functions.Proc1 updateMethod, + UpdateOptions options) + throws OperationException; + + /** + * Starts a one-argument workflow update with no return value on an existing workflow as a Nexus + * operation. See {@link #startWorkflowUpdate(Class, String, Functions.Func1, UpdateOptions)} for + * the full behavior contract. + * + * @param workflowClass the workflow interface class + * @param workflowId the ID of the existing workflow to update + * @param updateMethod unbound method reference to the update method + * @param arg1 first update argument + * @param options update options + * @param the workflow interface type + * @param the type of the first update argument + * @return a {@link TemporalOperationResult}; sync if the update already completed, async + * (carrying the update-workflow operation token) otherwise + */ + TemporalOperationResult startWorkflowUpdate( + Class workflowClass, + String workflowId, + Functions.Proc2 updateMethod, + A1 arg1, + UpdateOptions options) + throws OperationException; + + /** + * Starts a two-argument workflow update with no return value on an existing workflow as a Nexus + * operation. See {@link #startWorkflowUpdate(Class, String, Functions.Func1, UpdateOptions)} for + * the full behavior contract. + * + * @param workflowClass the workflow interface class + * @param workflowId the ID of the existing workflow to update + * @param updateMethod unbound method reference to the update method + * @param arg1 first update argument + * @param arg2 second update argument + * @param options update options + * @param the workflow interface type + * @param the type of the first update argument + * @param the type of the second update argument + * @return a {@link TemporalOperationResult}; sync if the update already completed, async + * (carrying the update-workflow operation token) otherwise + */ + TemporalOperationResult startWorkflowUpdate( + Class workflowClass, + String workflowId, + Functions.Proc3 updateMethod, + A1 arg1, + A2 arg2, + UpdateOptions options) + throws OperationException; + + /** + * Starts a three-argument workflow update with no return value on an existing workflow as a Nexus + * operation. See {@link #startWorkflowUpdate(Class, String, Functions.Func1, UpdateOptions)} for + * the full behavior contract. + * + * @param workflowClass the workflow interface class + * @param workflowId the ID of the existing workflow to update + * @param updateMethod unbound method reference to the update method + * @param arg1 first update argument + * @param arg2 second update argument + * @param arg3 third update argument + * @param options update options + * @param the workflow interface type + * @param the type of the first update argument + * @param the type of the second update argument + * @param the type of the third update argument + * @return a {@link TemporalOperationResult}; sync if the update already completed, async + * (carrying the update-workflow operation token) otherwise + */ + TemporalOperationResult startWorkflowUpdate( + Class workflowClass, + String workflowId, + Functions.Proc4 updateMethod, + A1 arg1, + A2 arg2, + A3 arg3, + UpdateOptions options) + throws OperationException; + + /** + * Starts a four-argument workflow update with no return value on an existing workflow as a Nexus + * operation. See {@link #startWorkflowUpdate(Class, String, Functions.Func1, UpdateOptions)} for + * the full behavior contract. + * + * @param workflowClass the workflow interface class + * @param workflowId the ID of the existing workflow to update + * @param updateMethod unbound method reference to the update method + * @param arg1 first update argument + * @param arg2 second update argument + * @param arg3 third update argument + * @param arg4 fourth update argument + * @param options update options + * @param the workflow interface type + * @param the type of the first update argument + * @param the type of the second update argument + * @param the type of the third update argument + * @param the type of the fourth update argument + * @return a {@link TemporalOperationResult}; sync if the update already completed, async + * (carrying the update-workflow operation token) otherwise + */ + TemporalOperationResult startWorkflowUpdate( + Class workflowClass, + String workflowId, + Functions.Proc5 updateMethod, + A1 arg1, + A2 arg2, + A3 arg3, + A4 arg4, + UpdateOptions options) + throws OperationException; + + /** + * Starts a five-argument workflow update with no return value on an existing workflow as a Nexus + * operation. See {@link #startWorkflowUpdate(Class, String, Functions.Func1, UpdateOptions)} for + * the full behavior contract. + * + * @param workflowClass the workflow interface class + * @param workflowId the ID of the existing workflow to update + * @param updateMethod unbound method reference to the update method + * @param arg1 first update argument + * @param arg2 second update argument + * @param arg3 third update argument + * @param arg4 fourth update argument + * @param arg5 fifth update argument + * @param options update options + * @param the workflow interface type + * @param the type of the first update argument + * @param the type of the second update argument + * @param the type of the third update argument + * @param the type of the fourth update argument + * @param the type of the fifth update argument + * @return a {@link TemporalOperationResult}; sync if the update already completed, async + * (carrying the update-workflow operation token) otherwise + */ + TemporalOperationResult startWorkflowUpdate( + Class workflowClass, + String workflowId, + Functions.Proc6 updateMethod, + A1 arg1, + A2 arg2, + A3 arg3, + A4 arg4, + A5 arg5, + UpdateOptions options) + throws OperationException; + + /** + * Starts a six-argument workflow update with no return value on an existing workflow as a Nexus + * operation. See {@link #startWorkflowUpdate(Class, String, Functions.Func1, UpdateOptions)} for + * the full behavior contract. + * + * @param workflowClass the workflow interface class + * @param workflowId the ID of the existing workflow to update + * @param updateMethod unbound method reference to the update method + * @param arg1 first update argument + * @param arg2 second update argument + * @param arg3 third update argument + * @param arg4 fourth update argument + * @param arg5 fifth update argument + * @param arg6 sixth update argument + * @param options update options + * @param the workflow interface type + * @param the type of the first update argument + * @param the type of the second update argument + * @param the type of the third update argument + * @param the type of the fourth update argument + * @param the type of the fifth update argument + * @param the type of the sixth update argument + * @return a {@link TemporalOperationResult}; sync if the update already completed, async + * (carrying the update-workflow operation token) otherwise + */ + TemporalOperationResult startWorkflowUpdate( + Class workflowClass, + String workflowId, + Functions.Proc7 updateMethod, + A1 arg1, + A2 arg2, + A3 arg3, + A4 arg4, + A5 arg5, + A6 arg6, + UpdateOptions options) + throws OperationException; } 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..14ac111f4 100644 --- a/temporal-sdk/src/main/java/io/temporal/nexus/TemporalNexusClientImpl.java +++ b/temporal-sdk/src/main/java/io/temporal/nexus/TemporalNexusClientImpl.java @@ -1,13 +1,22 @@ package io.temporal.nexus; +import com.google.common.base.Strings; +import io.nexusrpc.OperationException; import io.nexusrpc.handler.HandlerException; import io.nexusrpc.handler.OperationContext; import io.nexusrpc.handler.OperationStartDetails; +import io.temporal.client.UpdateOptions; import io.temporal.client.WorkflowClient; import io.temporal.client.WorkflowOptions; import io.temporal.client.WorkflowStub; +import io.temporal.client.WorkflowUpdateException; +import io.temporal.client.WorkflowUpdateHandle; +import io.temporal.client.WorkflowUpdateStage; import io.temporal.common.Experimental; 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.NexusStartWorkflowHelper; import io.temporal.workflow.Functions; import java.lang.reflect.Type; @@ -244,13 +253,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 +268,332 @@ private TemporalOperationResult invokeAndReturn(WorkflowHandle handle) throw t; } } + + // ---------- Update Workflow overloads ---------- + + @Override + public TemporalOperationResult startWorkflowUpdate( + Class workflowClass, + String workflowId, + Functions.Func1 updateMethod, + UpdateOptions options) + throws OperationException { + T stub = client.newWorkflowStub(workflowClass, workflowId); + return executeUpdate( + options, + effective -> WorkflowClient.startUpdate(() -> updateMethod.apply(stub), effective)); + } + + @Override + public TemporalOperationResult startWorkflowUpdate( + Class workflowClass, + String workflowId, + Functions.Func2 updateMethod, + A1 arg1, + UpdateOptions options) + throws OperationException { + T stub = client.newWorkflowStub(workflowClass, workflowId); + return executeUpdate( + options, + effective -> WorkflowClient.startUpdate(() -> updateMethod.apply(stub, arg1), effective)); + } + + @Override + public TemporalOperationResult startWorkflowUpdate( + Class workflowClass, + String workflowId, + Functions.Func3 updateMethod, + A1 arg1, + A2 arg2, + UpdateOptions options) + throws OperationException { + T stub = client.newWorkflowStub(workflowClass, workflowId); + return executeUpdate( + options, + effective -> + WorkflowClient.startUpdate(() -> updateMethod.apply(stub, arg1, arg2), effective)); + } + + @Override + public TemporalOperationResult startWorkflowUpdate( + Class workflowClass, + String workflowId, + Functions.Func4 updateMethod, + A1 arg1, + A2 arg2, + A3 arg3, + UpdateOptions options) + throws OperationException { + T stub = client.newWorkflowStub(workflowClass, workflowId); + return executeUpdate( + options, + effective -> + WorkflowClient.startUpdate( + () -> updateMethod.apply(stub, arg1, arg2, arg3), effective)); + } + + @Override + public TemporalOperationResult startWorkflowUpdate( + Class workflowClass, + String workflowId, + Functions.Func5 updateMethod, + A1 arg1, + A2 arg2, + A3 arg3, + A4 arg4, + UpdateOptions options) + throws OperationException { + T stub = client.newWorkflowStub(workflowClass, workflowId); + return executeUpdate( + options, + effective -> + WorkflowClient.startUpdate( + () -> updateMethod.apply(stub, arg1, arg2, arg3, arg4), effective)); + } + + @Override + public TemporalOperationResult startWorkflowUpdate( + Class workflowClass, + String workflowId, + Functions.Func6 updateMethod, + A1 arg1, + A2 arg2, + A3 arg3, + A4 arg4, + A5 arg5, + UpdateOptions options) + throws OperationException { + T stub = client.newWorkflowStub(workflowClass, workflowId); + return executeUpdate( + options, + effective -> + WorkflowClient.startUpdate( + () -> updateMethod.apply(stub, arg1, arg2, arg3, arg4, arg5), effective)); + } + + @Override + public TemporalOperationResult startWorkflowUpdate( + Class workflowClass, + String workflowId, + Functions.Func7 updateMethod, + A1 arg1, + A2 arg2, + A3 arg3, + A4 arg4, + A5 arg5, + A6 arg6, + UpdateOptions options) + throws OperationException { + T stub = client.newWorkflowStub(workflowClass, workflowId); + return executeUpdate( + options, + effective -> + WorkflowClient.startUpdate( + () -> updateMethod.apply(stub, arg1, arg2, arg3, arg4, arg5, arg6), effective)); + } + + @Override + public TemporalOperationResult startWorkflowUpdate( + Class workflowClass, + String workflowId, + Functions.Proc1 updateMethod, + UpdateOptions options) + throws OperationException { + T stub = client.newWorkflowStub(workflowClass, workflowId); + return executeUpdate( + options, + effective -> WorkflowClient.startUpdate(() -> updateMethod.apply(stub), effective)); + } + + @Override + public TemporalOperationResult startWorkflowUpdate( + Class workflowClass, + String workflowId, + Functions.Proc2 updateMethod, + A1 arg1, + UpdateOptions options) + throws OperationException { + T stub = client.newWorkflowStub(workflowClass, workflowId); + return executeUpdate( + options, + effective -> WorkflowClient.startUpdate(() -> updateMethod.apply(stub, arg1), effective)); + } + + @Override + public TemporalOperationResult startWorkflowUpdate( + Class workflowClass, + String workflowId, + Functions.Proc3 updateMethod, + A1 arg1, + A2 arg2, + UpdateOptions options) + throws OperationException { + T stub = client.newWorkflowStub(workflowClass, workflowId); + return executeUpdate( + options, + effective -> + WorkflowClient.startUpdate(() -> updateMethod.apply(stub, arg1, arg2), effective)); + } + + @Override + public TemporalOperationResult startWorkflowUpdate( + Class workflowClass, + String workflowId, + Functions.Proc4 updateMethod, + A1 arg1, + A2 arg2, + A3 arg3, + UpdateOptions options) + throws OperationException { + T stub = client.newWorkflowStub(workflowClass, workflowId); + return executeUpdate( + options, + effective -> + WorkflowClient.startUpdate( + () -> updateMethod.apply(stub, arg1, arg2, arg3), effective)); + } + + @Override + public TemporalOperationResult startWorkflowUpdate( + Class workflowClass, + String workflowId, + Functions.Proc5 updateMethod, + A1 arg1, + A2 arg2, + A3 arg3, + A4 arg4, + UpdateOptions options) + throws OperationException { + T stub = client.newWorkflowStub(workflowClass, workflowId); + return executeUpdate( + options, + effective -> + WorkflowClient.startUpdate( + () -> updateMethod.apply(stub, arg1, arg2, arg3, arg4), effective)); + } + + @Override + public TemporalOperationResult startWorkflowUpdate( + Class workflowClass, + String workflowId, + Functions.Proc6 updateMethod, + A1 arg1, + A2 arg2, + A3 arg3, + A4 arg4, + A5 arg5, + UpdateOptions options) + throws OperationException { + T stub = client.newWorkflowStub(workflowClass, workflowId); + return executeUpdate( + options, + effective -> + WorkflowClient.startUpdate( + () -> updateMethod.apply(stub, arg1, arg2, arg3, arg4, arg5), effective)); + } + + @Override + public TemporalOperationResult startWorkflowUpdate( + Class workflowClass, + String workflowId, + Functions.Proc7 updateMethod, + A1 arg1, + A2 arg2, + A3 arg3, + A4 arg4, + A5 arg5, + A6 arg6, + UpdateOptions options) + throws OperationException { + T stub = client.newWorkflowStub(workflowClass, workflowId); + return executeUpdate( + options, + effective -> + WorkflowClient.startUpdate( + () -> updateMethod.apply(stub, arg1, arg2, arg3, arg4, arg5, arg6), effective)); + } + + /** Function that will trigger {@code startUpdate} on overloads */ + @FunctionalInterface + private interface UpdateCommand { + WorkflowUpdateHandle triggerUpdate(UpdateOptions options); + } + + /** Common code for all {@code startWorkflowUpdate} overloads. */ + private TemporalOperationResult executeUpdate( + UpdateOptions options, UpdateCommand updateWrapper) throws OperationException { + + UpdateOptions.Builder effectiveOptsBuilder = UpdateOptions.newBuilder(options); + String requestId = operationStartDetails.getRequestId(); + if (Strings.isNullOrEmpty(options.getUpdateId())) { + // if updateId is unset, use requestId - consistent with other SDKs + effectiveOptsBuilder.setUpdateId(requestId); + } + options = effectiveOptsBuilder.build(); + checkNexusUpdateOptionsValid(options); + markAsyncOperationStarted(); + + InternalNexusOperationContext nexusContext = CurrentNexusOperationContext.get(); + try { + String callbackUrl = operationStartDetails.getCallbackUrl(); + if (Strings.isNullOrEmpty(callbackUrl)) { + throw new HandlerException( + HandlerException.ErrorType.BAD_REQUEST, + new IllegalArgumentException("callback URL is required for a Nexus operation")); + } + NexusOperationMetadata nexusOperationMetadata = + new NexusOperationMetadata( + requestId, callbackUrl, operationStartDetails.getCallbackHeaders()); + // set the nexusOperationMetadata and capture operationCompleted + nexusContext.setNexusOperationMetadata(nexusOperationMetadata); + WorkflowUpdateHandle handle = updateWrapper.triggerUpdate(options); + if (nexusOperationMetadata.operationCompleted) { + try { + R value = handle.getResult(); + return TemporalOperationResult.sync(value); + } catch (WorkflowUpdateException e) { + // Only case where operation is completed but getResult fails is if the update + // fails non-retriably - validation failure - so fail the operation immediately + throw OperationException.failed(e); + } + } + return TemporalOperationResult.async(nexusOperationMetadata.operationToken); + } catch (Throwable t) { + // Reset on failure so that if the update RPC throws, the handler can retry without being + // blocked by the guard. + asyncOperationStarted.set(false); + throw t; + } finally { + nexusContext.setNexusOperationMetadata(null); + } + } + + /** + * @throws OperationException if the options provided are invalid like missing + * UpdateName/WorkflowID/etc + */ + private void checkNexusUpdateOptionsValid(UpdateOptions options) + throws OperationException { + if (options.getWaitForStage() != WorkflowUpdateStage.ACCEPTED) { + throw OperationException.failed( + "workflow update Nexus operation only support WaitForStage Accepted"); + } + // draft-review: TBD, this is still under discussion as we may want to let + // handlers trigger invalid updates in some cases and just retry forever + try { + options.validate(); + } catch (IllegalStateException e) { + throw OperationException.failed(e); + } + } + + private void markAsyncOperationStarted() { + if (!asyncOperationStarted.compareAndSet(false, true)) { + throw new HandlerException( + HandlerException.ErrorType.BAD_REQUEST, + new IllegalStateException( + "Only one async operation can be started per operation handler invocation. " + + "Use getWorkflowClient() for additional workflow interactions.")); + } + } } diff --git a/temporal-sdk/src/main/java/io/temporal/nexus/TemporalOperationHandler.java b/temporal-sdk/src/main/java/io/temporal/nexus/TemporalOperationHandler.java index 6a01d11fc..84720408e 100644 --- a/temporal-sdk/src/main/java/io/temporal/nexus/TemporalOperationHandler.java +++ b/temporal-sdk/src/main/java/io/temporal/nexus/TemporalOperationHandler.java @@ -1,12 +1,12 @@ package io.temporal.nexus; +import io.nexusrpc.OperationException; import io.nexusrpc.handler.*; 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; /** @@ -48,7 +48,8 @@ public class TemporalOperationHandler implements OperationHandler { @FunctionalInterface public interface StartHandler { TemporalOperationResult apply( - TemporalOperationStartContext context, TemporalNexusClient client, T input); + TemporalOperationStartContext context, TemporalNexusClient client, T input) + throws OperationException; } private final StartHandler startHandler; @@ -70,7 +71,7 @@ public static TemporalOperationHandler create(StartHandler st @Override public final OperationStartResult start( - OperationContext ctx, OperationStartDetails details, T input) { + OperationContext ctx, OperationStartDetails details, T input) throws OperationException { InternalNexusOperationContext nexusCtx = CurrentNexusOperationContext.get(); TemporalNexusClient client = new TemporalNexusClientImpl(nexusCtx.getWorkflowClient(), ctx, details); @@ -100,12 +101,20 @@ 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 WORKFLOW_UPDATE: + cancelUpdateWorkflow( + cancelContext, + new CancelUpdateWorkflowInput( + token.getWorkflowId(), token.getRunId(), token.getUpdateId())); + break; + default: + throw new HandlerException( + HandlerException.ErrorType.BAD_REQUEST, + new IllegalArgumentException("unsupported operation token type: " + token.getType())); } } @@ -123,4 +132,21 @@ protected void cancelWorkflowRun( WorkflowClient client = CurrentNexusOperationContext.get().getWorkflowClient(); client.newUntypedWorkflowStub(input.getWorkflowId()).cancel(); } + + /** + * Called when a cancel request is received for a workflow update token. Override to customize + * cancel behavior. + * + *

Default behavior: not implemented. There is no server primitive to cancel an in-flight + * workflow update. + * + * @param context the cancel context + * @param input describes the update to cancel + */ + protected void cancelUpdateWorkflow( + TemporalOperationCancelContext context, CancelUpdateWorkflowInput input) { + throw new HandlerException( + HandlerException.ErrorType.NOT_IMPLEMENTED, + new UnsupportedOperationException("cannot cancel an UpdateWorkflow operation")); + } } diff --git a/temporal-sdk/src/test/java/io/temporal/internal/nexus/UpdateRunTokenTest.java b/temporal-sdk/src/test/java/io/temporal/internal/nexus/UpdateRunTokenTest.java new file mode 100644 index 000000000..5895d2916 --- /dev/null +++ b/temporal-sdk/src/test/java/io/temporal/internal/nexus/UpdateRunTokenTest.java @@ -0,0 +1,106 @@ +package io.temporal.internal.nexus; + +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.*; +import com.fasterxml.jackson.datatype.jdk8.Jdk8Module; +import java.util.Base64; +import org.junit.Assert; +import org.junit.Test; + +public class UpdateRunTokenTest { + private static final ObjectWriter ow = + new ObjectMapper().registerModule(new Jdk8Module()).writer(); + private static final Base64.Encoder encoder = Base64.getUrlEncoder().withoutPadding(); + + private static final String namespace = "ns"; + private static final String workflowId = "w"; + private static final String runId = "r"; + private static final String updateId = "u"; + + @Test + public void serializeWorkflowUpdateTokenOmitsEmptyRunId() throws JsonProcessingException { + OperationToken token = new OperationToken("ns", "wid", null, "uid"); + String json = ow.writeValueAsString(token); + JsonNode node = new ObjectMapper().readTree(json); + Assert.assertFalse(node.has("rid")); + } + + @Test + public void testEncodeDecode() throws JsonProcessingException { + String encoded = + OperationTokenUtil.generateWorkflowUpdateOperationToken( + namespace, workflowId, runId, updateId); + + OperationToken token = OperationTokenUtil.loadWorkflowUpdateOperationToken(encoded); + Assert.assertEquals(OperationTokenType.WORKFLOW_UPDATE, token.getType()); + Assert.assertEquals(namespace, token.getNamespace()); + Assert.assertEquals(workflowId, token.getWorkflowId()); + Assert.assertEquals(runId, token.getRunId()); + Assert.assertEquals(updateId, token.getUpdateId()); + Assert.assertNull(token.getVersion()); + } + + @Test + public void generateUpdateTokenRejectInvalid() { + // missing ns + Assert.assertThrows( + IllegalArgumentException.class, + () -> OperationTokenUtil.generateWorkflowUpdateOperationToken("", "w", "", "u")); + // missing workflowId + Assert.assertThrows( + IllegalArgumentException.class, + () -> OperationTokenUtil.generateWorkflowUpdateOperationToken("n", "", "", "u")); + // missing updateId + Assert.assertThrows( + IllegalArgumentException.class, + () -> OperationTokenUtil.generateWorkflowUpdateOperationToken("n", "w", "", "")); + } + + @Test + public void loadUpdateTokenRejectInvalid() { + // missing namespace + Assert.assertThrows( + IllegalArgumentException.class, + () -> + OperationTokenUtil.loadWorkflowUpdateOperationToken( + encoder.encodeToString( + "{\"t\":3,\"ns\":\"\",\"wid\":\"w\",\"uid\":\"u\"}".getBytes()))); + // missing workflowId + Assert.assertThrows( + IllegalArgumentException.class, + () -> + OperationTokenUtil.loadWorkflowUpdateOperationToken( + encoder.encodeToString( + "{\"t\":3,\"ns\":\"n\",\"wid\":\"\",\"uid\":\"u\"}".getBytes()))); + // missing updateId + Assert.assertThrows( + IllegalArgumentException.class, + () -> + OperationTokenUtil.loadWorkflowUpdateOperationToken( + encoder.encodeToString( + "{\"t\":3,\"ns\":\"n\",\"wid\":\"w\",\"uid\":\"\"}".getBytes()))); + } + + @Test + public void rejectInvalidUpdateTokenLoads() throws JsonProcessingException { + // loading update token from an encoded workflow run token should fail + Assert.assertThrows( + IllegalArgumentException.class, + () -> + OperationTokenUtil.loadWorkflowUpdateOperationToken( + OperationTokenUtil.generateWorkflowRunOperationToken("w", "n"))); + } + + @Test + public void loadWorkflowUpdateOperationTokenFromEncodedToken() { + // {"t":3,"ns":"ns","wid":"w","rid":"r","uid":"u"} + String encodedToken = "eyJ0IjozLCJucyI6Im5zIiwid2lkIjoidyIsInJpZCI6InIiLCJ1aWQiOiJ1In0"; + + OperationToken token = OperationTokenUtil.loadWorkflowUpdateOperationToken(encodedToken); + Assert.assertEquals(OperationTokenType.WORKFLOW_UPDATE, token.getType()); + Assert.assertEquals("ns", token.getNamespace()); + Assert.assertEquals("w", token.getWorkflowId()); + Assert.assertEquals("r", token.getRunId()); + Assert.assertEquals("u", token.getUpdateId()); + } +} diff --git a/temporal-sdk/src/test/java/io/temporal/workflow/nexus/UpdateWorkflowOperationTest.java b/temporal-sdk/src/test/java/io/temporal/workflow/nexus/UpdateWorkflowOperationTest.java new file mode 100644 index 000000000..97f3f7be5 --- /dev/null +++ b/temporal-sdk/src/test/java/io/temporal/workflow/nexus/UpdateWorkflowOperationTest.java @@ -0,0 +1,274 @@ +package io.temporal.workflow.nexus; + +import static org.junit.Assume.assumeTrue; + +import io.nexusrpc.Operation; +import io.nexusrpc.Service; +import io.nexusrpc.handler.OperationHandler; +import io.nexusrpc.handler.OperationImpl; +import io.nexusrpc.handler.ServiceImpl; +import io.temporal.client.UpdateOptions; +import io.temporal.client.WorkflowFailedException; +import io.temporal.client.WorkflowOptions; +import io.temporal.client.WorkflowStub; +import io.temporal.client.WorkflowUpdateStage; +import io.temporal.failure.ApplicationFailure; +import io.temporal.failure.NexusOperationFailure; +import io.temporal.internal.common.env.EnvironmentVariableUtils; +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.TestWorkflows; +import java.time.Duration; +import java.util.UUID; +import org.junit.AfterClass; +import org.junit.Assert; +import org.junit.BeforeClass; +import org.junit.ClassRule; +import org.junit.Test; + +public class UpdateWorkflowOperationTest extends BaseNexusTest { + + private static final String asyncVal = "async"; + private static final String doneSignal = "done"; + private static final String targetWorkflowId = "update-handler-workflow-" + UUID.randomUUID(); + + @ClassRule + public static SDKTestWorkflowRule testWorkflowRule = + SDKTestWorkflowRule.newBuilder() + .setWorkflowTypes(CallerWorkflow.class, HandlerWorkflowImpl.class) + .setNexusServiceImplementation(new TestNexusServiceImpl()) + .setUseTimeskipping(false) + .build(); + + private static WorkflowStub targetStub; + + @BeforeClass + public static void startTargetWorkflow() { + // start the handler workflow + targetStub = + testWorkflowRule + .getWorkflowClient() + .newUntypedWorkflowStub( + "HandlerWorkflow", + WorkflowOptions.newBuilder() + .setWorkflowId(targetWorkflowId) + .setTaskQueue(testWorkflowRule.getTaskQueue()) + .build()); + targetStub.start(); + } + + @AfterClass + public static void completeTargetWorkflow() { + // complete the handler workflow + targetStub.signal(doneSignal); + } + + @Override + protected SDKTestWorkflowRule getTestWorkflowRule() { + return testWorkflowRule; + } + + @Test + public void syncValidationFailureFailsOperation() { + TestWorkflows.TestWorkflow1 workflowStub = + testWorkflowRule.newWorkflowStubTimeoutOptions( + TestWorkflows.TestWorkflow1.class, "caller-validation-failure"); + + // empty value fails setValueValidator -> synchronous operation failure, no token issued + WorkflowFailedException e = + Assert.assertThrows( + WorkflowFailedException.class, () -> workflowStub.execute(targetWorkflowId + "|")); + + Assert.assertTrue(e.getCause() instanceof NexusOperationFailure); + NexusOperationFailure nexusFailure = (NexusOperationFailure) e.getCause(); + Assert.assertTrue(nexusFailure.getCause() instanceof ApplicationFailure); + } + + @Test + public void syncCompletedUpdateReturnsResult() { + // NOTE: this test makes no assumptions about whether update callbacks are enabled + // If enabled, the first call will resolve asynchronously due to NEXUS-489. If not, + // it resolves sync. In either case, the second call dedups and resolves synchronously + String fixedUpdateId = "fixed-update-id-" + testWorkflowRule.getTaskQueue(); + + TestWorkflows.TestWorkflow1 first = + testWorkflowRule.newWorkflowStubTimeoutOptions( + TestWorkflows.TestWorkflow1.class, "caller-dedup-completed-first"); + Assert.assertEquals( + "immediate-value", first.execute(targetWorkflowId + "|immediate-value|" + fixedUpdateId)); + + TestWorkflows.TestWorkflow1 second = + testWorkflowRule.newWorkflowStubTimeoutOptions( + TestWorkflows.TestWorkflow1.class, "caller-dedup-completed-second"); + Assert.assertEquals( + "immediate-value", + second.execute(targetWorkflowId + "|immediate-value|" + fixedUpdateId + "|dedup")); + } + + @Test + public void asyncUpdateWorkflowOperationCompletes() { + // only run this test if the server supports update completion callbacks + // currently only for local server manual checks until the in-memory test + // server can support update completion callbacks + assumeTrue( + "server does not support update completion callbacks", + EnvironmentVariableUtils.readBooleanFlag("UPDATE_CALLBACKS_SUPPORTED")); + + TestWorkflows.TestWorkflow1 workflowStub = + testWorkflowRule.newWorkflowStubTimeoutOptions( + TestWorkflows.TestWorkflow1.class, "caller-async-update"); + String result = workflowStub.execute(targetWorkflowId + "|" + asyncVal); + Assert.assertEquals(asyncVal, result); + } + + public static class CallerWorkflow implements TestWorkflows.TestWorkflow1 { + + @Override + public String execute(String input) { + String[] parts = input.split("\\|", 4); + String targetWorkflowId = parts[0]; + String value = parts.length > 1 ? parts[1] : ""; + String updateId = parts.length > 2 ? parts[2] : null; + boolean expectDedup = parts.length > 3 && "dedup".equals(parts[3]); + + NexusOperationOptions options = + NexusOperationOptions.newBuilder() + .setScheduleToCloseTimeout(Duration.ofSeconds(30)) + .build(); + NexusServiceOptions serviceOptions = + NexusServiceOptions.newBuilder() + .setEndpoint(getEndpointName()) + .setOperationOptions(options) + .build(); + + TestNexusUpdateService serviceStub = + Workflow.newNexusServiceStub(TestNexusUpdateService.class, serviceOptions); + + NexusOperationHandle handle = + Workflow.startNexusOperation( + serviceStub::update, new UpdateRequest(targetWorkflowId, value, updateId)); + + NexusOperationExecution execution = handle.getExecution().get(); + if (asyncVal.equals(value)) { + // must be issued with a token + Assert.assertTrue(execution.getOperationToken().isPresent()); + OperationToken token = + OperationTokenUtil.loadWorkflowUpdateOperationToken( + execution.getOperationToken().get()); + Assert.assertEquals(OperationTokenType.WORKFLOW_UPDATE, token.getType()); + Assert.assertEquals(targetWorkflowId, token.getWorkflowId()); + } else if (expectDedup) { + // dedup onto an already-completed update - sync, no token + Assert.assertFalse(execution.getOperationToken().isPresent()); + } + + return handle.getResult().get(); + } + } + + // Handler workflow + @WorkflowInterface + public interface HandlerWorkflow { + @WorkflowMethod + void execute(); + + @UpdateMethod(name = "setValue") + String setValue(String value); + + @UpdateValidatorMethod(updateName = "setValue") + void setValueValidator(String value); + + @SignalMethod(name = doneSignal) + void done(); + } + + public static class HandlerWorkflowImpl implements HandlerWorkflow { + private boolean completed; + + @Override + public void execute() { + Workflow.await(() -> completed); + } + + @Override + public void done() { + completed = true; + } + + @Override + public String setValue(String value) { + if (asyncVal.equals(value)) { + Workflow.sleep(Duration.ofSeconds(1)); + } + return value; + } + + @Override + public void setValueValidator(String value) { + if (value == null || value.isEmpty()) { + throw new IllegalArgumentException("value required"); + } + } + } + + // Nexus Service + @Service + public interface TestNexusUpdateService { + @Operation + String update(UpdateRequest input); + } + + @ServiceImpl(service = TestNexusUpdateService.class) + public static class TestNexusServiceImpl { + @OperationImpl + public OperationHandler update() { + return TemporalOperationHandler.create( + (context, client, input) -> { + UpdateOptions.Builder optionsBuilder = + UpdateOptions.newBuilder(String.class) + .setUpdateName("setValue") + .setWaitForStage(WorkflowUpdateStage.ACCEPTED); + if (input.getUpdateId() != null) { + optionsBuilder.setUpdateId(input.getUpdateId()); + } + return client.startWorkflowUpdate( + HandlerWorkflow.class, + input.getTargetWorkflowId(), + HandlerWorkflow::setValue, + input.getValue(), + optionsBuilder.build()); + }); + } + } + + // container for the update input + public static final class UpdateRequest { + public String targetWorkflowId; + public String value; + public String updateId; + + public UpdateRequest() {} + + public UpdateRequest(String targetWorkflowId, String value, String updateId) { + this.targetWorkflowId = targetWorkflowId; + this.value = value; + this.updateId = updateId; + } + + public String getTargetWorkflowId() { + return targetWorkflowId; + } + + public String getValue() { + return value; + } + + public String getUpdateId() { + return updateId; + } + } +} diff --git a/temporal-serviceclient/src/main/proto b/temporal-serviceclient/src/main/proto index d2fc34ab8..7e04182a8 160000 --- a/temporal-serviceclient/src/main/proto +++ b/temporal-serviceclient/src/main/proto @@ -1 +1 @@ -Subproject commit d2fc34ab844603f50e41365f46c7fb82bdedffe6 +Subproject commit 7e04182a8df692eb1608d7e26d00aca9f58cb7a1