Skip to content

Commit 93b84f2

Browse files
Quinn-With-Two-NsVegetarianOrc
Quinn-With-Two-Ns
and
VegetarianOrc
authored
Add Standalone Nexus Operation links (#2932)
* Add SANO links * Test more SANO links * Clean up * Fix link converter * Use BeforeClass for checking for SANO support in tests --------- Co-authored-by: Alex Mazzeo <alex.mazzeo@temporal.io>
1 parent 6e23861 commit 93b84f2

8 files changed

Lines changed: 775 additions & 180 deletions

File tree

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

Lines changed: 3 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -22,14 +22,11 @@
2222
import io.temporal.internal.nexus.OperationTokenUtil;
2323
import java.util.*;
2424
import java.util.stream.Collectors;
25-
import org.slf4j.Logger;
26-
import org.slf4j.LoggerFactory;
2725

2826
/** Utility functions shared by the implementation code. */
2927
public final class InternalUtils {
3028
public static String TEMPORAL_RESERVED_PREFIX = "__temporal_";
3129

32-
private static final Logger log = LoggerFactory.getLogger(InternalUtils.class);
3330
private static String QUERY_TYPE_STACK_TRACE = "__stack_trace";
3431
private static String ENHANCED_QUERY_TYPE_STACK_TRACE = "__enhanced_stack_trace";
3532

@@ -93,21 +90,12 @@ public static NexusWorkflowStarter createNexusBoundStub(
9390
? null
9491
: request.getLinks().stream()
9592
.map(
96-
(link) -> {
97-
if (io.temporal.api.common.v1.Link.WorkflowEvent.getDescriptor()
98-
.getFullName()
99-
.equals(link.getType())) {
100-
io.temporal.api.nexus.v1.Link nexusLink =
93+
(link) ->
94+
LinkConverter.nexusLinkToLink(
10195
io.temporal.api.nexus.v1.Link.newBuilder()
10296
.setType(link.getType())
10397
.setUrl(link.getUri().toString())
104-
.build();
105-
return LinkConverter.nexusLinkToWorkflowEvent(nexusLink);
106-
} else {
107-
log.warn("ignoring unsupported link data type: {}", link.getType());
108-
return null;
109-
}
110-
})
98+
.build()))
11199
.filter(Objects::nonNull)
112100
.collect(Collectors.toList());
113101
WorkflowOptions.Builder nexusWorkflowOptions =

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

Lines changed: 107 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,8 @@ public class LinkConverter {
2020
private static final Logger log = LoggerFactory.getLogger(LinkConverter.class);
2121

2222
private static final String linkPathFormat = "temporal:///namespaces/%s/workflows/%s/%s/history";
23+
private static final String nexusOperationLinkPathFormat =
24+
"temporal:///namespaces/%s/nexus-operations/%s/%s/details";
2325
private static final String linkReferenceTypeKey = "referenceType";
2426
private static final String linkEventIDKey = "eventID";
2527
private static final String linkEventTypeKey = "eventType";
@@ -29,6 +31,10 @@ public class LinkConverter {
2931
Link.WorkflowEvent.EventReference.getDescriptor().getName();
3032
private static final String requestIDReferenceType =
3133
Link.WorkflowEvent.RequestIdReference.getDescriptor().getName();
34+
private static final String workflowEventLinkType =
35+
Link.WorkflowEvent.getDescriptor().getFullName();
36+
private static final String nexusOperationLinkType =
37+
Link.NexusOperation.getDescriptor().getFullName();
3238

3339
public static io.temporal.api.nexus.v1.Link workflowEventToNexusLink(Link.WorkflowEvent we) {
3440
try {
@@ -160,6 +166,107 @@ public static Link nexusLinkToWorkflowEvent(io.temporal.api.nexus.v1.Link nexusL
160166
return link.build();
161167
}
162168

169+
/**
170+
* Dispatches on the oneof variant of {@code commonLink} and converts to the matching {@link
171+
* io.temporal.api.nexus.v1.Link}. Returns {@code null} if no variant is set or encoding fails.
172+
*/
173+
public static io.temporal.api.nexus.v1.Link linkToNexusLink(Link commonLink) {
174+
if (commonLink.hasWorkflowEvent()) {
175+
return workflowEventToNexusLink(commonLink.getWorkflowEvent());
176+
}
177+
if (commonLink.hasNexusOperation()) {
178+
return nexusOperationToNexusLink(commonLink.getNexusOperation());
179+
}
180+
return null;
181+
}
182+
183+
/**
184+
* Dispatches on {@link io.temporal.api.nexus.v1.Link#getType()} and converts to the matching
185+
* {@link Link} variant. Returns {@code null} for unknown or unparseable types.
186+
*/
187+
public static Link nexusLinkToLink(io.temporal.api.nexus.v1.Link nexusLink) {
188+
String type = nexusLink.getType();
189+
if (workflowEventLinkType.equals(type)) {
190+
return nexusLinkToWorkflowEvent(nexusLink);
191+
}
192+
if (nexusOperationLinkType.equals(type)) {
193+
return nexusLinkToNexusOperation(nexusLink);
194+
}
195+
log.warn("ignoring unsupported nexus link type: {}", type);
196+
return null;
197+
}
198+
199+
public static io.temporal.api.nexus.v1.Link nexusOperationToNexusLink(Link.NexusOperation no) {
200+
try {
201+
String url =
202+
String.format(
203+
nexusOperationLinkPathFormat,
204+
URLEncoder.encode(no.getNamespace(), StandardCharsets.UTF_8.toString()),
205+
// See the WorkflowId comment in workflowEventToNexusLink for why '+' is rewritten to
206+
// '%20'. OperationId is user-supplied and can legally contain spaces.
207+
URLEncoder.encode(no.getOperationId(), StandardCharsets.UTF_8.toString())
208+
.replace("+", "%20"),
209+
URLEncoder.encode(no.getRunId(), StandardCharsets.UTF_8.toString()));
210+
return io.temporal.api.nexus.v1.Link.newBuilder()
211+
.setUrl(url)
212+
.setType(nexusOperationLinkType)
213+
.build();
214+
} catch (Exception e) {
215+
log.error("Failed to encode Nexus operation link URL", e);
216+
}
217+
return null;
218+
}
219+
220+
public static Link nexusLinkToNexusOperation(io.temporal.api.nexus.v1.Link nexusLink) {
221+
if (!nexusOperationLinkType.equals(nexusLink.getType())) {
222+
log.error(
223+
"Failed to parse Nexus link URL: cannot parse link type {} to {}",
224+
nexusLink.getType(),
225+
nexusOperationLinkType);
226+
return null;
227+
}
228+
Link.Builder link = Link.newBuilder();
229+
try {
230+
URI uri = new URI(nexusLink.getUrl());
231+
232+
if (!"temporal".equals(uri.getScheme())) {
233+
log.error("Failed to parse Nexus link URL: invalid scheme: {}", uri.getScheme());
234+
return null;
235+
}
236+
237+
StringTokenizer st = new StringTokenizer(uri.getRawPath(), "/");
238+
if (!st.hasMoreTokens() || !st.nextToken().equals("namespaces")) {
239+
log.error("Failed to parse Nexus link URL: invalid path: {}", uri.getRawPath());
240+
return null;
241+
}
242+
String namespace = URLDecoder.decode(st.nextToken(), StandardCharsets.UTF_8.toString());
243+
if (!st.hasMoreTokens() || !st.nextToken().equals("nexus-operations")) {
244+
log.error("Failed to parse Nexus link URL: invalid path: {}", uri.getRawPath());
245+
return null;
246+
}
247+
String operationId = URLDecoder.decode(st.nextToken(), StandardCharsets.UTF_8.toString());
248+
if (!st.hasMoreTokens()) {
249+
log.error("Failed to parse Nexus link URL: invalid path: {}", uri.getRawPath());
250+
return null;
251+
}
252+
String runId = URLDecoder.decode(st.nextToken(), StandardCharsets.UTF_8.toString());
253+
if (!st.hasMoreTokens() || !st.nextToken().equals("details")) {
254+
log.error("Failed to parse Nexus link URL: invalid path: {}", uri.getRawPath());
255+
return null;
256+
}
257+
258+
link.setNexusOperation(
259+
Link.NexusOperation.newBuilder()
260+
.setNamespace(namespace)
261+
.setOperationId(operationId)
262+
.setRunId(runId));
263+
} catch (Exception e) {
264+
log.error("Failed to parse Nexus link URL", e);
265+
return null;
266+
}
267+
return link.build();
268+
}
269+
163270
private static Map<String, String> parseQueryParams(URI uri) throws UnsupportedEncodingException {
164271
final String query = uri.getQuery();
165272
if (query == null || query.isEmpty()) {

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

Lines changed: 6 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,6 @@
11
package io.temporal.internal.nexus;
22

3+
import static io.temporal.internal.common.LinkConverter.linkToNexusLink;
34
import static io.temporal.internal.common.LinkConverter.workflowEventToNexusLink;
45
import static io.temporal.internal.common.NexusUtil.nexusProtoLinkToLink;
56

@@ -50,12 +51,10 @@ public static NexusStartWorkflowResponse startWorkflowAndAttachLinks(
5051

5152
// If the start workflow response returned a link use it, otherwise
5253
// create the link information about the new workflow and return to the caller.
53-
Link.WorkflowEvent workflowEventLink =
54-
nexusCtx.getStartWorkflowResponseLink().hasWorkflowEvent()
55-
? nexusCtx.getStartWorkflowResponseLink().getWorkflowEvent()
56-
: null;
57-
if (workflowEventLink == null) {
58-
workflowEventLink =
54+
io.temporal.api.nexus.v1.Link nexusLink =
55+
linkToNexusLink(nexusCtx.getStartWorkflowResponseLink());
56+
if (nexusLink == null) {
57+
Link.WorkflowEvent synthesized =
5958
Link.WorkflowEvent.newBuilder()
6059
.setNamespace(nexusCtx.getNamespace())
6160
.setWorkflowId(workflowExec.getWorkflowId())
@@ -64,8 +63,8 @@ public static NexusStartWorkflowResponse startWorkflowAndAttachLinks(
6463
Link.WorkflowEvent.EventReference.newBuilder()
6564
.setEventType(EventType.EVENT_TYPE_WORKFLOW_EXECUTION_STARTED))
6665
.build();
66+
nexusLink = workflowEventToNexusLink(synthesized);
6767
}
68-
io.temporal.api.nexus.v1.Link nexusLink = workflowEventToNexusLink(workflowEventLink);
6968
if (nexusLink != null) {
7069
try {
7170
ctx.addLinks(nexusProtoLinkToLink(nexusLink));

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

Lines changed: 6 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -301,20 +301,13 @@ private StartOperationResponse handleStartOperation(
301301
"Invalid link URL: " + link.getUrl(),
302302
e);
303303
}
304-
// LinkConverter only returns a WorkflowEvent-shaped common.v1.Link; nexus links of
305-
// other shapes (e.g. non-temporal URLs) come back null and are intentionally not
306-
// forwarded onto the RPCs the handler issues, which require the WorkflowEvent
307-
// variant. Log so a debugging session can see what was dropped.
308-
io.temporal.api.common.v1.Link commonLink =
309-
LinkConverter.nexusLinkToWorkflowEvent(link);
304+
// Convert inbound Nexus links into common.v1.Link so RPCs issued by the handler
305+
// (e.g. signal, signalWithStart) can attach them as request links. Both
306+
// WorkflowEvent (caller workflow → nexus op scheduled) and NexusOperation (SANO
307+
// record) variants flow through; other shapes are dropped.
308+
io.temporal.api.common.v1.Link commonLink = LinkConverter.nexusLinkToLink(link);
310309
if (commonLink != null) {
311310
inboundCommonLinks.add(commonLink);
312-
} else {
313-
log.warn(
314-
"Dropping inbound Nexus link from outbound link propagation: type='{}',"
315-
+ " url='{}' (not a parseable temporal WorkflowEvent link)",
316-
link.getType(),
317-
link.getUrl());
318311
}
319312
});
320313
CurrentNexusOperationContext.get().setRequestLinks(inboundCommonLinks);
@@ -335,11 +328,7 @@ private StartOperationResponse handleStartOperation(
335328
List<io.temporal.api.nexus.v1.Link> responseLinks = new ArrayList<>();
336329
for (io.temporal.api.common.v1.Link responseLink :
337330
CurrentNexusOperationContext.get().getResponseLinks()) {
338-
if (!responseLink.hasWorkflowEvent()) {
339-
continue;
340-
}
341-
io.temporal.api.nexus.v1.Link converted =
342-
LinkConverter.workflowEventToNexusLink(responseLink.getWorkflowEvent());
331+
io.temporal.api.nexus.v1.Link converted = LinkConverter.linkToNexusLink(responseLink);
343332
if (converted != null) {
344333
responseLinks.add(converted);
345334
}

0 commit comments

Comments
 (0)