Workflow run token (#2421)

Add WorkflowOperationToken

Author: Quinn-With-Two-Ns

temporal-opentracing/src/test/java/io/temporal/opentracing/NexusOperationTest.java

@@ -111,6 +111,7 @@ public interface TestOtherWorkflow {
   public static class OtherWorkflowImpl implements TestOtherWorkflow {
     @Override
     public String workflow(String input) {
+      Workflow.sleep(Duration.ofSeconds(1));
       return "Hello, " + input + "!";
     }
   }

temporal-sdk/src/main/java/io/temporal/internal/nexus/OperationTokenType.java

@@ -0,0 +1,53 @@
+/*
+ * Copyright (C) 2022 Temporal Technologies, Inc. All Rights Reserved.
+ *
+ * Copyright (C) 2012-2016 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ * Modifications copyright (C) 2017 Uber Technologies, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this material except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.temporal.internal.nexus;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonValue;
+
+public enum OperationTokenType {
+  UNKNOWN(0),
+  WORKFLOW_RUN(1);
+
+  private final int value;
+
+  OperationTokenType(int i) {
+    this.value = i;
+  }
+
+  @JsonValue
+  public int toValue() {
+    return value;
+  }
+
+  @JsonCreator
+  public static OperationTokenType fromValue(Integer value) {
+    if (value == null) {
+      return UNKNOWN;
+    }
+    for (OperationTokenType b : OperationTokenType.values()) {
+      if (b.value == value) {
+        return b;
+      }
+    }
+    return UNKNOWN;
+  }
+}

temporal-sdk/src/main/java/io/temporal/internal/nexus/OperationTokenUtil.java

@@ -0,0 +1,97 @@
+/*
+ * Copyright (C) 2022 Temporal Technologies, Inc. All Rights Reserved.
+ *
+ * Copyright (C) 2012-2016 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ * Modifications copyright (C) 2017 Uber Technologies, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this material except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.temporal.internal.nexus;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.JavaType;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.ObjectWriter;
+import com.fasterxml.jackson.datatype.jdk8.Jdk8Module;
+import com.google.common.base.Strings;
+import java.util.Base64;
+
+public class OperationTokenUtil {
+  private static final ObjectMapper mapper = new ObjectMapper().registerModule(new Jdk8Module());
+  private static final ObjectWriter ow = mapper.writer();
+  private static final Base64.Decoder decoder = Base64.getUrlDecoder();
+  private static final Base64.Encoder encoder = Base64.getUrlEncoder().withoutPadding();
+
+  /**
+   * Load a workflow run operation token from an operation token.
+   *
+   * @throws FallbackToWorkflowIdException if the operation token is not a workflow run token
+   * @throws IllegalArgumentException if the operation token is invalid
+   */
+  public static WorkflowRunOperationToken loadWorkflowRunOperationToken(String operationToken)
+      throws FallbackToWorkflowIdException {
+    WorkflowRunOperationToken token;
+    try {
+      JavaType reference = mapper.getTypeFactory().constructType(WorkflowRunOperationToken.class);
+      token = mapper.readValue(decoder.decode(operationToken), reference);
+    } catch (Exception e) {
+      throw new FallbackToWorkflowIdException("Failed to parse operation token: " + e.getMessage());
+    }
+    if (!token.getType().equals(OperationTokenType.WORKFLOW_RUN)) {
+      throw new IllegalArgumentException(
+          "Invalid workflow run token: incorrect operation token type: " + token.getType());
+    }
+    if (token.getVersion() != null) {
+      throw new IllegalArgumentException("Invalid workflow run token: unexpected version field");
+    }
+    if (Strings.isNullOrEmpty(token.getWorkflowId())) {
+      throw new IllegalArgumentException("Invalid workflow run token: missing workflow ID (wid)");
+    }
+    return token;
+  }
+
+  /**
+   * Attempt to extract the workflow Id from an operation token.
+   *
+   * @throws IllegalArgumentException if the operation token is invalid
+   */
+  public static String loadWorkflowIdFromOperationToken(String operationToken) {
+    try {
+      WorkflowRunOperationToken token = loadWorkflowRunOperationToken(operationToken);
+      return token.getWorkflowId();
+    } catch (OperationTokenUtil.FallbackToWorkflowIdException e) {
+      // Previous versions of the SDK simply used the workflow ID as the operation token
+      // This fallback is provided for backwards compatibility for those cases.
+      // This fallback will be removed in a future release.
+      // See: https://github.com/temporalio/sdk-java/issues/2423
+      return operationToken;
+    }
+  }
+
+  /** Generate a workflow run operation token from a workflow ID and namespace. */
+  public static String generateWorkflowRunOperationToken(String workflowId, String namespace)
+      throws JsonProcessingException {
+    String json = ow.writeValueAsString(new WorkflowRunOperationToken(namespace, workflowId));
+    return encoder.encodeToString(json.getBytes());
+  }
+
+  public static class FallbackToWorkflowIdException extends RuntimeException {
+    public FallbackToWorkflowIdException(String message) {
+      super(message);
+    }
+  }
+
+  private OperationTokenUtil() {}
+}

temporal-sdk/src/main/java/io/temporal/internal/nexus/WorkflowRunOperationToken.java

@@ -0,0 +1,73 @@
+/*
+ * Copyright (C) 2022 Temporal Technologies, Inc. All Rights Reserved.
+ *
+ * Copyright (C) 2012-2016 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ * Modifications copyright (C) 2017 Uber Technologies, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this material except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.temporal.internal.nexus;
+
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+public class WorkflowRunOperationToken {
+  @JsonProperty("v")
+  @JsonInclude(JsonInclude.Include.NON_NULL)
+  private final Integer version;
+
+  @JsonProperty("t")
+  private final OperationTokenType type;
+
+  @JsonProperty("ns")
+  private final String namespace;
+
+  @JsonProperty("wid")
+  private final String workflowId;
+
+  public WorkflowRunOperationToken(
+      @JsonProperty("t") Integer type,
+      @JsonProperty("ns") String namespace,
+      @JsonProperty("wid") String workflowId,
+      @JsonProperty("v") Integer version) {
+    this.type = OperationTokenType.fromValue(type);
+    this.namespace = namespace;
+    this.workflowId = workflowId;
+    this.version = version;
+  }
+
+  public WorkflowRunOperationToken(String namespace, String workflowId) {
+    this.type = OperationTokenType.WORKFLOW_RUN;
+    this.namespace = namespace;
+    this.workflowId = workflowId;
+    this.version = null;
+  }
+
+  public Integer getVersion() {
+    return version;
+  }
+
+  public OperationTokenType getType() {
+    return type;
+  }
+
+  public String getNamespace() {
+    return namespace;
+  }
+
+  public String getWorkflowId() {
+    return workflowId;
+  }
+}

temporal-sdk/src/main/java/io/temporal/nexus/WorkflowRunOperationImpl.java

@@ -23,6 +23,7 @@
 import static io.temporal.internal.common.LinkConverter.workflowEventToNexusLink;
 import static io.temporal.internal.common.NexusUtil.nexusProtoLinkToLink;
 
+import com.fasterxml.jackson.core.JsonProcessingException;
 import io.nexusrpc.OperationInfo;
 import io.nexusrpc.handler.*;
 import io.nexusrpc.handler.OperationHandler;
@@ -33,6 +34,7 @@
 import io.temporal.internal.client.NexusStartWorkflowRequest;
 import io.temporal.internal.nexus.CurrentNexusOperationContext;
 import io.temporal.internal.nexus.InternalNexusOperationContext;
+import io.temporal.internal.nexus.OperationTokenUtil;
 import java.net.URISyntaxException;
 
 class WorkflowRunOperationImpl<T, R> implements OperationHandler<T, R> {
@@ -70,19 +72,31 @@ public OperationStartResult<R> start(
                     .setEventType(EventType.EVENT_TYPE_WORKFLOW_EXECUTION_STARTED))
             .build();
     io.temporal.api.nexus.v1.Link nexusLink = workflowEventToNexusLink(workflowEventLink);
+    // Generate the operation token for the new workflow.
+    String operationToken;
     try {
-      OperationStartResult.Builder<R> result =
-          OperationStartResult.newAsyncBuilder(workflowExec.getWorkflowId());
-      if (nexusLink != null) {
-        ctx.addLinks(nexusProtoLinkToLink(nexusLink));
-      }
-      return result.build();
-    } catch (URISyntaxException e) {
+      operationToken =
+          OperationTokenUtil.generateWorkflowRunOperationToken(
+              workflowExec.getWorkflowId(), nexusCtx.getNamespace());
+    } catch (JsonProcessingException e) {
       // Not expected as the link is constructed by the SDK.
       throw new HandlerException(
-          HandlerException.ErrorType.INTERNAL,
-          new IllegalArgumentException("failed to parse URI", e));
+          HandlerException.ErrorType.BAD_REQUEST,
+          new IllegalArgumentException("failed to generate workflow operation token", e));
+    }
+    // Attach the link to the operation result.
+    OperationStartResult.Builder<R> result = OperationStartResult.newAsyncBuilder(operationToken);
+    if (nexusLink != null) {
+      try {
+        ctx.addLinks(nexusProtoLinkToLink(nexusLink));
+      } catch (URISyntaxException e) {
+        // Not expected as the link is constructed by the SDK.
+        throw new HandlerException(
+            HandlerException.ErrorType.INTERNAL,
+            new IllegalArgumentException("failed to parse URI", e));
+      }
     }
+    return result.build();
   }
 
   @Override
@@ -100,7 +114,18 @@ public OperationInfo fetchInfo(
   @Override
   public void cancel(
       OperationContext operationContext, OperationCancelDetails operationCancelDetails) {
+    String workflowId;
+    try {
+      workflowId =
+          OperationTokenUtil.loadWorkflowIdFromOperationToken(
+              operationCancelDetails.getOperationToken());
+    } catch (IllegalArgumentException e) {
+      throw new HandlerException(
+          HandlerException.ErrorType.BAD_REQUEST,
+          new IllegalArgumentException("failed to parse operation token", e));
+    }
+
     WorkflowClient client = CurrentNexusOperationContext.get().getWorkflowClient();
-    client.newUntypedWorkflowStub(operationCancelDetails.getOperationToken()).cancel();
+    client.newUntypedWorkflowStub(workflowId).cancel();
   }
 }