Skip to content

Add Nexus link converter for Link.Callback #3156

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
atol wants to merge 1 commit into
base: main
Choose a base branch
from
+263 −23
Open

Add Nexus link converter for Link.Callback #3156

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions CHANGELOG.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -49,6 +49,8 @@ to docs, or any other relevant information.
- Workflow tasks taking longer than 5 seconds now log a `[TMPRL1104]` warning reporting the task duration and the
external storage downloads and uploads that contributed to it. The threshold is configurable with the
`TEMPORAL_WORKFLOW_TASK_DURATION_WARN_SECONDS` environment variable.
- Nexus links of type `Link.Callback` are now converted to and from their Nexus wire form, so they are carried
through Nexus requests and responses instead of being dropped with an "unsupported nexus link type" warning.

### Changed
- Moved the Spring AI integration to [temporalio/ai-integrations](https://github.com/temporalio/ai-integrations/tree/main/java/spring-ai).
Expand Down
179 changes: 157 additions & 22 deletions temporal-sdk/src/main/java/io/temporal/internal/common/LinkConverter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,8 +4,11 @@
import static io.temporal.internal.common.ProtoEnumNameUtils.simplifiedToUniqueName;
import static io.temporal.internal.common.ProtoEnumNameUtils.uniqueToSimplifiedName;

import com.google.common.collect.ImmutableBiMap;
import io.temporal.api.common.v1.Execution;
import io.temporal.api.common.v1.Link;
import io.temporal.api.enums.v1.EventType;
import io.temporal.api.enums.v1.ExecutionType;
import java.io.UnsupportedEncodingException;
import java.net.URI;
import java.net.URLDecoder;
Expand All @@ -24,13 +27,14 @@
* Converts between {@link Link} (used on history events and RPCs) and {@link
* io.temporal.api.nexus.v1.Link} (the Nexus wire form: a URL plus a type string).
*
* <p>Four link types are supported, each with a fixed URL shape:
* <p>The following link types are supported, each with a fixed URL shape:
*
* <pre>
* WorkflowEvent temporal:///namespaces/{ns}/workflows/{workflowId}/{runId}/history
* Workflow temporal:///namespaces/{ns}/workflows/{workflowId}/{runId}
* NexusOperation temporal:///namespaces/{ns}/nexus-operations/{operationId}/{runId}/details
* Activity temporal:///namespaces/{ns}/activities/{activityId}/{runId}/details
* Callback temporal:///namespaces/{ns}/{executionType}/{businessId}/{runId}/callbacks/{requestId}
* </pre>
*
* <p>Whoever decodes one of these URLs — the receiving server, or this class when a link arrives
Expand All @@ -54,42 +58,63 @@ public class LinkConverter {
private static final String EVENT_TYPE_KEY = "eventType";
private static final String REQUEST_ID_KEY = "requestID";
private static final String REASON_KEY = "reason";
private static final String COMPONENT_PATH_KEY = "componentPath";

/**
* The path keyword for each execution type a callback can be attached to, in both directions:
* encoding looks up the keyword, decoding validates it and maps it back. Matches the keywords of
* the other link types, so every link addresses an execution the same way.
*/
private static final ImmutableBiMap<ExecutionType, String> EXECUTION_TYPE_SEGMENTS =
ImmutableBiMap.of(
ExecutionType.EXECUTION_TYPE_WORKFLOW, "workflows",
ExecutionType.EXECUTION_TYPE_ACTIVITY, "activities",
ExecutionType.EXECUTION_TYPE_NEXUS_OPERATION, "nexus-operations");

private static final String EVENT_REFERENCE_TYPE =
Link.WorkflowEvent.EventReference.getDescriptor().getName();
private static final String REQUEST_ID_REFERENCE_TYPE =
Link.WorkflowEvent.RequestIdReference.getDescriptor().getName();

/**
* The four link types, as a table of (path keyword, path tail, proto type name).
* The link types, as a table of (path keyword, path tail, tail ID, proto type name).
*
* <p>The path is always {@code /namespaces/{ns}/{keyword}/{id}/{runId}[/{tail}]}, so a link has 5
* segments when {@link #tail} is null and 6 otherwise. Matching the count exactly is what keeps a
* Workflow link distinguishable from a WorkflowEvent link, since they share a keyword.
* <p>The path is always {@code /namespaces/{ns}/{keyword}/{id}/{runId}[/{tail}[/{tailId}]]}, so a
* link has 5 segments with no tail, 6 with a tail, and 7 with a tail ID. Matching the count
* exactly is what keeps a Workflow link distinguishable from a WorkflowEvent link, since they
* share a keyword.
*/
private enum LinkType {
WORKFLOW_EVENT("workflows", "history", Link.WorkflowEvent.getDescriptor().getFullName()),
WORKFLOW_EVENT("workflows", "history", false, Link.WorkflowEvent.getDescriptor().getFullName()),

/** A workflow execution as a whole, for when there is no history event to point at. */
WORKFLOW("workflows", null, Link.Workflow.getDescriptor().getFullName()),
WORKFLOW("workflows", null, false, Link.Workflow.getDescriptor().getFullName()),

NEXUS_OPERATION(
"nexus-operations", "details", Link.NexusOperation.getDescriptor().getFullName()),
"nexus-operations", "details", false, Link.NexusOperation.getDescriptor().getFullName()),

ACTIVITY("activities", "details", Link.Activity.getDescriptor().getFullName());
ACTIVITY("activities", "details", false, Link.Activity.getDescriptor().getFullName()),

private final String keyword;
/**
* A callback attached to an execution. The keyword names the execution's type, so it varies and
* is any value of {@link #EXECUTION_TYPE_SEGMENTS}. The tail ID is the callback's request ID.
*/
CALLBACK(null, "callbacks", true, Link.Callback.getDescriptor().getFullName());

@Nullable private final String keyword;
@Nullable private final String tail;
private final boolean hasTailId;
private final String type;

LinkType(String keyword, @Nullable String tail, String type) {
LinkType(@Nullable String keyword, @Nullable String tail, boolean hasTailId, String type) {
this.keyword = keyword;
this.tail = tail;
this.hasTailId = hasTailId;
this.type = type;
}

int segmentCount() {
return tail == null ? 5 : 6;
return 5 + (tail == null ? 0 : 1) + (hasTailId ? 1 : 0);
}
}

Expand All @@ -112,6 +137,9 @@ public static io.temporal.api.nexus.v1.Link linkToNexusLink(Link commonLink) {
if (commonLink.hasActivity()) {
return activityToNexusLink(commonLink.getActivity());
}
if (commonLink.hasCallback()) {
return callbackToNexusLink(commonLink.getCallback());
}
return null;
}

Expand Down Expand Up @@ -156,6 +184,30 @@ public static io.temporal.api.nexus.v1.Link activityToNexusLink(Link.Activity ac
Collections.emptyList());
}

@Nullable
public static io.temporal.api.nexus.v1.Link callbackToNexusLink(Link.Callback cb) {
Execution execution = cb.getExecution();
String executionType = EXECUTION_TYPE_SEGMENTS.get(execution.getType());
if (executionType == null) {
log.error(
"Failed to convert Callback link to Nexus link: unsupported execution type: {}",
execution.getType());
return null;
}
List<Map.Entry<String, String>> query = new ArrayList<>();
for (String segment : cb.getComponentPathList()) {
query.add(param(COMPONENT_PATH_KEY, segment));
}
return encode(
LinkType.CALLBACK,
executionType,
cb.getNamespace(),
execution.getBusinessId(),
execution.getRunId(),
cb.getRequestId(),
query);
}

// ===============================================================================================
// Decode: nexus.v1.Link -> Link.
// ===============================================================================================
Expand All @@ -176,6 +228,9 @@ public static Link nexusLinkToLink(io.temporal.api.nexus.v1.Link nexusLink) {
if (LinkType.ACTIVITY.type.equals(type)) {
return nexusLinkToActivity(nexusLink);
}
if (LinkType.CALLBACK.type.equals(type)) {
return nexusLinkToCallback(nexusLink);
}
log.warn("ignoring unsupported nexus link type: {}", type);
return null;
}
Expand Down Expand Up @@ -245,22 +300,58 @@ public static Link nexusLinkToActivity(io.temporal.api.nexus.v1.Link nexusLink)
.build();
}

/** Inverse of {@link #callbackToNexusLink}. A missing component path decodes to an empty one. */
@Nullable
public static Link nexusLinkToCallback(io.temporal.api.nexus.v1.Link nexusLink) {
Decoded decoded = decode(LinkType.CALLBACK, nexusLink);
if (decoded == null) {
return null;
}
ExecutionType executionType = EXECUTION_TYPE_SEGMENTS.inverse().get(decoded.keyword);
return Link.newBuilder()
.setCallback(
Link.Callback.newBuilder()
.setNamespace(decoded.namespace)
.setExecution(
Execution.newBuilder()
.setType(executionType)
.setBusinessId(decoded.id)
.setRunId(decoded.runId))
.setRequestId(decoded.tailId)
.addAllComponentPath(decoded.componentPath))
.build();
}

// ===============================================================================================
// Shared encode/decode.
// ===============================================================================================

/** Builds a Nexus link URL for {@code linkType}, which must have a fixed keyword. */
@Nullable
private static io.temporal.api.nexus.v1.Link encode(
LinkType linkType,
String namespace,
String id,
String runId,
List<Map.Entry<String, String>> queryParams) {
return encode(linkType, linkType.keyword, namespace, id, runId, null, queryParams);
}

/**
* Builds a Nexus link URL for {@code linkType}.
* Builds a Nexus link URL for {@code linkType}, with {@code keyword} in place of the type's own
* for a type whose keyword varies, and {@code tailId} after the tail for a type that has one.
*
* <p>Concatenated rather than built with {@link URI}, because the segments are already
* percent-encoded and {@link URI} would escape the escapes.
*/
@Nullable
private static io.temporal.api.nexus.v1.Link encode(
LinkType linkType,
String keyword,
String namespace,
String id,
String runId,
@Nullable String tailId,
List<Map.Entry<String, String>> queryParams) {
try {
StringBuilder url =
Expand All @@ -270,14 +361,17 @@ private static io.temporal.api.nexus.v1.Link encode(
.append('/')
.append(encodePathSegment(namespace))
.append('/')
.append(linkType.keyword)
.append(keyword)
.append('/')
.append(encodePathSegment(id))
.append('/')
.append(encodePathSegment(runId));
if (linkType.tail != null) {
url.append('/').append(linkType.tail);
}
if (tailId != null) {
url.append('/').append(encodePathSegment(tailId));
}
if (!queryParams.isEmpty()) {
url.append('?').append(encodeQuery(queryParams));
}
Expand All @@ -291,18 +385,31 @@ private static io.temporal.api.nexus.v1.Link encode(
}
}

/** The three path IDs plus the decoded query. */
/** The path IDs, the execution keyword and the tail ID if any, plus the query. */
private static final class Decoded {
final String namespace;
final String keyword;
final String id;
final String runId;
@Nullable final String tailId;
final Map<String, String> query;

Decoded(String namespace, String id, String runId, Map<String, String> query) {
final List<String> componentPath;

Decoded(
String namespace,
String keyword,
String id,
String runId,
@Nullable String tailId,
@Nullable String rawQuery)
throws UnsupportedEncodingException {
this.namespace = namespace;
this.keyword = keyword;
this.id = id;
this.runId = runId;
this.query = query;
this.tailId = tailId;
this.query = decodeQuery(rawQuery);
this.componentPath = decodeQueryValues(rawQuery);
}
}

Expand Down Expand Up @@ -339,21 +446,29 @@ private static Decoded decode(LinkType linkType, io.temporal.api.nexus.v1.Link n

if (segments.length != linkType.segmentCount()
|| !NAMESPACES_SEGMENT.equals(segments[0])
|| !linkType.keyword.equals(segments[2])
|| (linkType.keyword != null && !linkType.keyword.equals(segments[2]))
|| (linkType.tail != null && !linkType.tail.equals(segments[5]))) {
log.error("Failed to parse Nexus link URL: invalid path: {}", rawPath);
return null;
}
if (segments[1].isEmpty() || segments[3].isEmpty() || segments[4].isEmpty()) {
log.error("Failed to parse Nexus link URL: empty path segment: {}", rawPath);
for (String segment : segments) {
if (segment.isEmpty()) {
log.error("Failed to parse Nexus link URL: empty path segment: {}", rawPath);
return null;
}
}
if (linkType.keyword == null && !EXECUTION_TYPE_SEGMENTS.containsValue(segments[2])) {
log.error("Failed to parse Nexus link URL: unsupported execution type: {}", segments[2]);
return null;
}

return new Decoded(
decodePathSegment(segments[1]),
segments[2],
decodePathSegment(segments[3]),
decodePathSegment(segments[4]),
decodeQuery(uri.getRawQuery()));
linkType.hasTailId ? decodePathSegment(segments[6]) : null,
uri.getRawQuery());
} catch (Exception e) {
// Swallow un-parsable links since they are not critical to processing.
log.error("Failed to parse Nexus link URL", e);
Expand Down Expand Up @@ -435,6 +550,26 @@ private static Map<String, String> decodeQuery(@Nullable String rawQuery)
return params;
}

/**
* Form-decodes every {@code componentPath} value in a raw query, in order. {@link #decodeQuery}
* keeps only the first value of a repeated key, so this repeated param needs its own pass. A key
* with no value yields an empty string, matching api-go's {@code Query()}.
*/
private static List<String> decodeQueryValues(@Nullable String rawQuery)
throws UnsupportedEncodingException {
List<String> values = new ArrayList<>();
if (rawQuery == null || rawQuery.isEmpty()) {
return values;
}
for (String pair : rawQuery.split("&")) {
String[] kv = pair.split("=", 2);
if (!pair.isEmpty() && COMPONENT_PATH_KEY.equals(URLDecoder.decode(kv[0], UTF_8))) {
values.add(kv.length == 2 ? URLDecoder.decode(kv[1], UTF_8) : "");
}
}
return values;
}

// ===============================================================================================
// The WorkflowEvent "reference": WHICH event in the workflow's history the link points at.
//
Expand Down
Loading
Loading