Skip to content

Commit 7852a29

Browse files
mavemurimavemuri
committed
NEXUS-485: Support Workflow Update as a Nexus Operation
1 parent e68853e commit 7852a29

14 files changed

Lines changed: 1357 additions & 20 deletions

‎temporal-sdk/src/main/java/io/temporal/internal/client/RootWorkflowClientInvoker.java‎

Lines changed: 56 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,7 @@
55
import static io.temporal.internal.common.HeaderUtils.intoPayloadMap;
66
import static io.temporal.internal.common.WorkflowExecutionUtils.makeUserMetaData;
77

8+
import com.fasterxml.jackson.core.JsonProcessingException;
89
import com.google.common.collect.Iterators;
910
import io.grpc.Deadline;
1011
import io.grpc.Status;
@@ -23,7 +24,11 @@
2324
import io.temporal.common.interceptors.WorkflowClientCallsInterceptor;
2425
import io.temporal.internal.client.external.GenericWorkflowClient;
2526
import io.temporal.internal.common.HeaderUtils;
27+
import io.temporal.internal.common.InternalUtils;
2628
import io.temporal.internal.nexus.CurrentNexusOperationContext;
29+
import io.temporal.internal.nexus.InternalNexusOperationContext;
30+
import io.temporal.internal.nexus.NexusOperationMetadata;
31+
import io.temporal.internal.nexus.OperationTokenUtil;
2732
import io.temporal.internal.worker.WorkerVersioningProtoUtils;
2833
import io.temporal.payload.context.WorkflowSerializationContext;
2934
import io.temporal.serviceclient.StatusUtils;
@@ -473,6 +478,19 @@ public <R> WorkflowUpdateHandle<R> startUpdate(StartUpdateInput<R> input) {
473478
}
474479
} while (updateNotYetDurable(input, result));
475480

481+
// If triggered by a Nexus Operation, set necessary fields- link, result
482+
if (CurrentNexusOperationContext.isNexusContext()) {
483+
NexusOperationMetadata nexusOperationMetadata =
484+
CurrentNexusOperationContext.get().getNexusOperationMetadata();
485+
if (nexusOperationMetadata != null) {
486+
if (result.hasLink()) {
487+
// add forward links for caller->handler
488+
CurrentNexusOperationContext.get().addResponseLink(result.getLink());
489+
}
490+
nexusOperationMetadata.operationCompleted = result.hasOutcome();
491+
}
492+
}
493+
476494
return toUpdateHandle(input, result, dataConverterWithWorkflowContext);
477495
}
478496

@@ -495,14 +513,49 @@ private <R> UpdateWorkflowExecutionRequest toUpdateWorkflowExecutionRequest(
495513
.setName(input.getUpdateName());
496514
inputArgs.ifPresent(updateInput::setArgs);
497515

498-
Request request =
516+
Request.Builder requestBuilder =
499517
Request.newBuilder()
500518
.setMeta(
501519
Meta.newBuilder()
502520
.setUpdateId(input.getUpdateId())
503521
.setIdentity(clientOptions.getIdentity()))
504-
.setInput(updateInput)
505-
.build();
522+
.setInput(updateInput);
523+
524+
// If this update is being issued from inside a Nexus operation handler via
525+
// TemporalNexusClientImpl.startWorkflowUpdate, set the fields the server needs
526+
// to deliver the Nexus completion callback.
527+
if (CurrentNexusOperationContext.isNexusContext()) {
528+
InternalNexusOperationContext nexusContext = CurrentNexusOperationContext.get();
529+
// already in a Nexus operation context, dont need to check nexusContext again
530+
NexusOperationMetadata nexusOperationMetadata = nexusContext.getNexusOperationMetadata();
531+
if (nexusOperationMetadata != null) {
532+
try {
533+
nexusOperationMetadata.operationToken =
534+
OperationTokenUtil.generateWorkflowUpdateOperationToken(
535+
clientOptions.getNamespace(),
536+
input.getWorkflowExecution().getWorkflowId(),
537+
input.getWorkflowExecution().getRunId(),
538+
input.getUpdateId());
539+
} catch (JsonProcessingException e) {
540+
throw new IllegalStateException("failed to generate update operation token", e);
541+
}
542+
List<Link> requestLinks = nexusContext.getRequestLinks();
543+
requestBuilder
544+
.setRequestId(nexusOperationMetadata.requestId)
545+
.addCompletionCallbacks(
546+
InternalUtils.buildNexusCallback(
547+
nexusOperationMetadata.callbackHeaders,
548+
nexusOperationMetadata.callbackUrl,
549+
nexusOperationMetadata.operationToken,
550+
requestLinks))
551+
.addAllLinks(requestLinks);
552+
} else {
553+
log.error("unexpected error fetching nexusOperationMetadata");
554+
// maybe throw an exception but should never really happen as its all sdk
555+
}
556+
}
557+
558+
Request request = requestBuilder.build();
506559

507560
return UpdateWorkflowExecutionRequest.newBuilder()
508561
.setNamespace(clientOptions.getNamespace())

‎temporal-sdk/src/main/java/io/temporal/internal/common/InternalUtils.java‎

Lines changed: 25 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -147,6 +147,31 @@ public static NexusWorkflowStarter createNexusBoundStub(
147147
return new NexusWorkflowStarter(stub.newInstance(nexusWorkflowOptions.build()), operationToken);
148148
}
149149

150+
/** Helper to build a Nexus Callback from the provided input. */
151+
public static Callback buildNexusCallback(
152+
Map<String, String> callbackHeaders,
153+
String callbackUrl,
154+
String operationToken,
155+
List<Link> links) {
156+
Map<String, String> headers =
157+
callbackHeaders.entrySet().stream()
158+
.collect(
159+
Collectors.toMap(
160+
(k) -> k.getKey().toLowerCase(),
161+
Map.Entry::getValue,
162+
(a, b) -> a,
163+
TreeMap::new));
164+
headers.put(Header.OPERATION_TOKEN.toLowerCase(), operationToken);
165+
Callback.Builder cbBuilder =
166+
Callback.newBuilder()
167+
.setNexus(
168+
Callback.Nexus.newBuilder().setUrl(callbackUrl).putAllHeader(headers).build());
169+
if (links != null) {
170+
cbBuilder.addAllLinks(links);
171+
}
172+
return cbBuilder.build();
173+
}
174+
150175
/** Check the method name for reserved prefixes or names. */
151176
public static void checkMethodName(POJOWorkflowMethodMetadata methodMetadata) {
152177
if (methodMetadata.getName().startsWith(TEMPORAL_RESERVED_PREFIX)) {

‎temporal-sdk/src/main/java/io/temporal/internal/common/LinkConverter.java‎

Lines changed: 64 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,7 @@ public class LinkConverter {
1919

2020
private static final Logger log = LoggerFactory.getLogger(LinkConverter.class);
2121

22+
private static final String temporalUrlScheme = "temporal";
2223
private static final String linkPathFormat = "temporal:///namespaces/%s/workflows/%s/%s/history";
2324
private static final String nexusOperationLinkPathFormat =
2425
"temporal:///namespaces/%s/nexus-operations/%s/%s/details";
@@ -35,6 +36,7 @@ public class LinkConverter {
3536
Link.WorkflowEvent.getDescriptor().getFullName();
3637
private static final String nexusOperationLinkType =
3738
Link.NexusOperation.getDescriptor().getFullName();
39+
private static final String workflowLinkType = Link.Workflow.getDescriptor().getFullName();
3840

3941
public static io.temporal.api.nexus.v1.Link workflowEventToNexusLink(Link.WorkflowEvent we) {
4042
try {
@@ -93,6 +95,24 @@ public static io.temporal.api.nexus.v1.Link workflowEventToNexusLink(Link.Workfl
9395
return null;
9496
}
9597

98+
public static io.temporal.api.nexus.v1.Link workflowLinkToNexusLink(Link.Workflow w) {
99+
try {
100+
String namespace = URLEncoder.encode(w.getNamespace(), StandardCharsets.UTF_8.toString());
101+
String workflowId =
102+
URLEncoder.encode(w.getWorkflowId(), StandardCharsets.UTF_8.toString())
103+
.replace("+", "%20"); // handle workflowIds supporting spaces
104+
String runId = URLEncoder.encode(w.getRunId(), StandardCharsets.UTF_8.toString());
105+
String url = String.format(linkPathFormat, namespace, workflowId, runId);
106+
return io.temporal.api.nexus.v1.Link.newBuilder()
107+
.setUrl(url)
108+
.setType(workflowLinkType)
109+
.build();
110+
} catch (UnsupportedEncodingException e) {
111+
log.error("Failed to encode Nexus link URL", e);
112+
return null;
113+
}
114+
}
115+
96116
public static Link nexusLinkToWorkflowEvent(io.temporal.api.nexus.v1.Link nexusLink) {
97117
Link.Builder link = Link.newBuilder();
98118
try {
@@ -166,6 +186,43 @@ public static Link nexusLinkToWorkflowEvent(io.temporal.api.nexus.v1.Link nexusL
166186
return link.build();
167187
}
168188

189+
public static Link nexusLinkToWorkflowLink(io.temporal.api.nexus.v1.Link nexusLink) {
190+
Link.Builder link = Link.newBuilder();
191+
try {
192+
URI uri = new URI(nexusLink.getUrl());
193+
if (!uri.getScheme().equals(temporalUrlScheme)) {
194+
log.error("Failed to parse Nexus link URL: invalid scheme: {}", uri.getScheme());
195+
return null;
196+
}
197+
StringTokenizer st = new StringTokenizer(uri.getRawPath(), "/");
198+
// maybe add constants for "namespaces", "workflows" too
199+
if (!st.nextToken().equals("namespaces")) {
200+
log.error("Failed to parse Nexus link URL: invalid path: {}", uri.getRawPath());
201+
return null;
202+
}
203+
String namespace = URLDecoder.decode(st.nextToken(), StandardCharsets.UTF_8.toString());
204+
if (!st.nextToken().equals("workflows")) {
205+
log.error("Failed to parse Nexus link URL: invalid path: {}", uri.getRawPath());
206+
return null;
207+
}
208+
String workflowID = URLDecoder.decode(st.nextToken(), StandardCharsets.UTF_8.toString());
209+
if (!st.hasMoreTokens()) {
210+
log.error("Failed to parse Nexus link URL: invalid path: {}", uri.getRawPath());
211+
return null;
212+
}
213+
String runID = URLDecoder.decode(st.nextToken(), StandardCharsets.UTF_8.toString());
214+
link.setWorkflow(
215+
Link.Workflow.newBuilder()
216+
.setNamespace(namespace)
217+
.setWorkflowId(workflowID)
218+
.setRunId(runID));
219+
} catch (Exception e) {
220+
log.error("Failed to parse Nexus link URL", e);
221+
return null;
222+
}
223+
return link.build();
224+
}
225+
169226
/**
170227
* Dispatches on the oneof variant of {@code commonLink} and converts to the matching {@link
171228
* io.temporal.api.nexus.v1.Link}. Returns {@code null} if no variant is set or encoding fails.
@@ -177,6 +234,10 @@ public static io.temporal.api.nexus.v1.Link linkToNexusLink(Link commonLink) {
177234
if (commonLink.hasNexusOperation()) {
178235
return nexusOperationToNexusLink(commonLink.getNexusOperation());
179236
}
237+
// check for workflow link only if a more specific variant is not set
238+
if (commonLink.hasWorkflow()) {
239+
return workflowLinkToNexusLink(commonLink.getWorkflow());
240+
}
180241
return null;
181242
}
182243

@@ -192,6 +253,9 @@ public static Link nexusLinkToLink(io.temporal.api.nexus.v1.Link nexusLink) {
192253
if (nexusOperationLinkType.equals(type)) {
193254
return nexusLinkToNexusOperation(nexusLink);
194255
}
256+
if (workflowLinkType.equals(type)) {
257+
return nexusLinkToWorkflowLink(nexusLink);
258+
}
195259
log.warn("ignoring unsupported nexus link type: {}", type);
196260
return null;
197261
}

‎temporal-sdk/src/main/java/io/temporal/internal/nexus/InternalNexusOperationContext.java‎

Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -38,6 +38,21 @@ public class InternalNexusOperationContext {
3838
private final Object responseLinksLock = new Object();
3939
private final List<Link> responseLinks = new ArrayList<>();
4040

41+
private NexusOperationMetadata nexusOperationMetadata;
42+
43+
/**
44+
* Set the Nexus operation metadata
45+
*
46+
* @param metadata {@link NexusOperationMetadata} to be set
47+
*/
48+
public void setNexusOperationMetadata(NexusOperationMetadata metadata) {
49+
this.nexusOperationMetadata = metadata;
50+
}
51+
52+
public NexusOperationMetadata getNexusOperationMetadata() {
53+
return nexusOperationMetadata;
54+
}
55+
4156
public InternalNexusOperationContext(
4257
String namespace,
4358
String taskQueue,

‎temporal-sdk/src/main/java/io/temporal/internal/nexus/NexusOperationMetadata.java‎

Lines changed: 22 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,22 @@
1+
package io.temporal.internal.nexus;
2+
3+
import io.temporal.common.Experimental;
4+
import java.util.Map;
5+
6+
/** Container for an in-flight Nexus operation metadata. */
7+
@Experimental
8+
public final class NexusOperationMetadata {
9+
public final String requestId;
10+
public final String callbackUrl;
11+
public final Map<String, String> callbackHeaders;
12+
13+
public String operationToken;
14+
public boolean operationCompleted;
15+
16+
public NexusOperationMetadata(
17+
String requestId, String callbackUrl, Map<String, String> callbackHeaders) {
18+
this.requestId = requestId;
19+
this.callbackUrl = callbackUrl;
20+
this.callbackHeaders = callbackHeaders;
21+
}
22+
}

‎temporal-sdk/src/main/java/io/temporal/internal/nexus/OperationToken.java‎

Lines changed: 35 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -18,22 +18,49 @@ public class OperationToken {
1818
@JsonProperty("wid")
1919
private final String workflowId;
2020

21+
@JsonProperty("rid")
22+
@JsonInclude(JsonInclude.Include.NON_NULL)
23+
// only set for updates and activities
24+
private final String runId;
25+
26+
@JsonProperty("uid")
27+
@JsonInclude(JsonInclude.Include.NON_NULL)
28+
// only set for updates
29+
private final String updateId;
30+
2131
public OperationToken(
2232
@JsonProperty("t") Integer type,
2333
@JsonProperty("ns") String namespace,
2434
@JsonProperty("wid") String workflowId,
35+
@JsonProperty("rid") String runId,
36+
@JsonProperty("uid") String updateId,
2537
@JsonProperty("v") Integer version) {
2638
this.type = OperationTokenType.fromValue(type);
2739
this.namespace = namespace;
2840
this.workflowId = workflowId;
41+
this.runId = runId;
42+
this.updateId = updateId;
2943
this.version = version;
3044
}
3145

46+
/** Generate a token for a workflow run operation */
3247
public OperationToken(OperationTokenType type, String namespace, String workflowId) {
3348
this.type = type;
3449
this.namespace = namespace;
3550
this.workflowId = workflowId;
3651
this.version = null;
52+
this.runId = null;
53+
this.updateId = null;
54+
}
55+
56+
/** Generate a token for a workflow update operation */
57+
public OperationToken(String namespace, String workflowId, String runId, String updateId) {
58+
this.type = OperationTokenType.WORKFLOW_UPDATE;
59+
this.namespace = namespace;
60+
this.workflowId = workflowId;
61+
this.runId = runId;
62+
this.updateId = updateId;
63+
this.version = null;
3764
}
3865

3966
public Integer getVersion() {
@@ -51,4 +78,12 @@ public String getNamespace() {
5178
public String getWorkflowId() {
5279
return workflowId;
5380
}
81+
82+
public String getUpdateId() {
83+
return updateId;
84+
}
85+
86+
public String getRunId() {
87+
return runId;
88+
}
5489
}

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

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -5,7 +5,8 @@
55

66
public enum OperationTokenType {
77
UNKNOWN(0),
8-
WORKFLOW_RUN(1);
8+
WORKFLOW_RUN(1),
9+
WORKFLOW_UPDATE(3);
910

1011
private final int value;
1112

0 commit comments

Comments
 (0)