diff --git a/CHANGELOG.md b/CHANGELOG.md index d426a29297..b93088d53f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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). 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 bbdc0b926e..5de80b1062 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 @@ -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; @@ -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). * - *

Four link types are supported, each with a fixed URL shape: + *

The following link types are supported, each with a fixed URL shape: * * 

  *   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}
  *
* *

Whoever decodes one of these URLs — the receiving server, or this class when a link arrives @@ -54,6 +58,18 @@ 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 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(); @@ -61,35 +77,44 @@ public class LinkConverter { 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). * - *

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. + *

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); } } @@ -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; } @@ -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> 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. // =============================================================================================== @@ -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; } @@ -245,12 +300,46 @@ 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> 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. * *

Concatenated rather than built with {@link URI}, because the segments are already * percent-encoded and {@link URI} would escape the escapes. @@ -258,9 +347,11 @@ public static Link nexusLinkToActivity(io.temporal.api.nexus.v1.Link nexusLink) @Nullable private static io.temporal.api.nexus.v1.Link encode( LinkType linkType, + String keyword, String namespace, String id, String runId, + @Nullable String tailId, List> queryParams) { try { StringBuilder url = @@ -270,7 +361,7 @@ private static io.temporal.api.nexus.v1.Link encode( .append('/') .append(encodePathSegment(namespace)) .append('/') - .append(linkType.keyword) + .append(keyword) .append('/') .append(encodePathSegment(id)) .append('/') @@ -278,6 +369,9 @@ private static io.temporal.api.nexus.v1.Link encode( 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)); } @@ -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 query; - - Decoded(String namespace, String id, String runId, Map query) { + final List 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); } } @@ -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); @@ -435,6 +550,26 @@ private static Map 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 decodeQueryValues(@Nullable String rawQuery) + throws UnsupportedEncodingException { + List 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. // diff --git a/temporal-sdk/src/test/java/io/temporal/internal/common/LinkConverterTest.java b/temporal-sdk/src/test/java/io/temporal/internal/common/LinkConverterTest.java index 8086b5c9a5..db000a1aed 100644 --- a/temporal-sdk/src/test/java/io/temporal/internal/common/LinkConverterTest.java +++ b/temporal-sdk/src/test/java/io/temporal/internal/common/LinkConverterTest.java @@ -2,6 +2,7 @@ import static io.temporal.internal.common.LinkConverter.linkToNexusLink; import static io.temporal.internal.common.LinkConverter.nexusLinkToActivity; +import static io.temporal.internal.common.LinkConverter.nexusLinkToCallback; import static io.temporal.internal.common.LinkConverter.nexusLinkToLink; import static io.temporal.internal.common.LinkConverter.nexusLinkToNexusOperation; import static io.temporal.internal.common.LinkConverter.nexusLinkToWorkflowEvent; @@ -11,8 +12,11 @@ import static org.junit.Assert.assertNull; import static org.junit.Assert.fail; +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.util.Arrays; import org.junit.Test; /** @@ -32,6 +36,7 @@ public class LinkConverterTest { private static final String WORKFLOW = Link.Workflow.getDescriptor().getFullName(); private static final String NEXUS_OPERATION = Link.NexusOperation.getDescriptor().getFullName(); private static final String ACTIVITY = Link.Activity.getDescriptor().getFullName(); + private static final String CALLBACK = Link.Callback.getDescriptor().getFullName(); // =============================================================================================== // Encode. @@ -174,6 +179,63 @@ public void encodesActivityLink() { ACTIVITY, activity("ns", "act/id", "run-id")); } + /** The path keyword names the execution type. */ + @Test + public void encodesCallbackForEachExecutionType() { + assertEncodes( + "temporal:///namespaces/ns/workflows/wf-id/run-id/callbacks/req-id", + CALLBACK, + callback("ns", ExecutionType.EXECUTION_TYPE_WORKFLOW, "wf-id", "run-id", "req-id")); + assertEncodes( + "temporal:///namespaces/ns/activities/act-id/run-id/callbacks/req-id", + CALLBACK, + callback("ns", ExecutionType.EXECUTION_TYPE_ACTIVITY, "act-id", "run-id", "req-id")); + assertEncodes( + "temporal:///namespaces/ns/nexus-operations/op-id/run-id/callbacks/req-id", + CALLBACK, + callback("ns", ExecutionType.EXECUTION_TYPE_NEXUS_OPERATION, "op-id", "run-id", "req-id")); + } + + /** + * One {@code componentPath} param per segment, in the path's order rather than sorted, with each + * segment form-encoded. + */ + @Test + public void encodesComponentPathAsRepeatedParamInOrder() { + String url = "temporal:///namespaces/ns/workflows/wf-id/run-id/callbacks/req-id"; + ExecutionType wf = ExecutionType.EXECUTION_TYPE_WORKFLOW; + assertEncodes( + url + "?componentPath=A%3AB&componentPath=C%2FD&componentPath=E_F", + CALLBACK, + callback("ns", wf, "wf-id", "run-id", "req-id", "A:B", "C/D", "E_F")); + assertEncodes( + url + "?componentPath=Updates&componentPath=id%2Cwith%2Ccommas", + CALLBACK, + callback("ns", wf, "wf-id", "run-id", "req-id", "Updates", "id,with,commas")); + assertEncodes( + url + "?componentPath=zulu&componentPath=alpha", + CALLBACK, + callback("ns", wf, "wf-id", "run-id", "req-id", "zulu", "alpha")); + } + + /** + * An encoded {@code /callbacks/} inside the business ID must not read as the tail, and the + * request ID is escaped like any other segment. + */ + @Test + public void encodesReservedCharactersInCallbackPath() { + assertEncodes( + "temporal:///namespaces/ns%2Fwith-slash/workflows/wf-id%2Fcallbacks%2Ffake/run-id" + + "/callbacks/request%20id%3Fx%3D1", + CALLBACK, + callback( + "ns/with-slash", + ExecutionType.EXECUTION_TYPE_WORKFLOW, + "wf-id/callbacks/fake", + "run-id", + "request id?x=1")); + } + /** The event type goes on the wire in the short PascalCase form. */ @Test public void encodesEventTypeInPascalCase() { @@ -314,6 +376,7 @@ public void rejectsWrongScheme() { assertRejected(WORKFLOW_EVENT, "https:///namespaces/ns/workflows/wf-id/run-id/history"); assertRejected(WORKFLOW, "https:///namespaces/ns/workflows/wf-id/run-id"); assertRejected(NEXUS_OPERATION, "https:///namespaces/ns/nexus-operations/op/run-id/details"); + assertRejected(CALLBACK, "https:///namespaces/ns/workflows/wf-id/run-id/callbacks/req-id"); } /** A workflow link ends at the run ID; a workflow-event link ends at {@code /history}. */ @@ -328,9 +391,12 @@ public void rejectsTrailingPathSegment() { assertRejected( WORKFLOW_EVENT, "temporal:///namespaces/ns/workflows/wf-id/run-id/history/extra"); assertRejected(WORKFLOW, "temporal:///namespaces/ns/workflows/wf-id/run-id/extra"); + assertRejected( + CALLBACK, "temporal:///namespaces/ns/workflows/wf-id/run-id/callbacks/req-id/extra"); // A trailing slash is an empty extra segment, not a no-op. assertRejected(WORKFLOW_EVENT, "temporal:///namespaces/ns/workflows/wf-id/run-id/history/"); assertRejected(WORKFLOW, "temporal:///namespaces/ns/workflows/wf-id/run-id/"); + assertRejected(CALLBACK, "temporal:///namespaces/ns/workflows/wf-id/run-id/callbacks/req-id/"); } /** A repeated key takes its first occurrence, as api-go does. No encoder emits one. */ @@ -347,17 +413,24 @@ public void rejectsMissingPathSegment() { assertRejected(WORKFLOW, "temporal:///namespaces/ns/workflows/wf-id"); assertRejected(NEXUS_OPERATION, "temporal:///namespaces/ns/nexus-operations/op-id/run-id"); assertRejected(ACTIVITY, "temporal:///namespaces/ns/activities/act-id/run-id"); + assertRejected(CALLBACK, "temporal:///namespaces/ns/workflows/wf-id/run-id/callbacks"); } @Test public void rejectsEmptyPathSegment() { assertRejected(WORKFLOW_EVENT, "temporal:///namespaces//workflows/wf-id/run-id/history"); assertRejected(WORKFLOW_EVENT, "temporal:///namespaces/ns/workflows//run-id/history"); + assertRejected(CALLBACK, "temporal:///namespaces/ns/workflows//run-id/callbacks/req-id"); + assertRejected(CALLBACK, "temporal:///namespaces/ns/workflows/wf-id/run-id/callbacks/"); } @Test public void rejectsWrongKindSegment() { assertRejected(WORKFLOW_EVENT, "temporal:///namespaces/ns/activities/wf-id/run-id/history"); + // A callback's keyword varies, but only over the execution types. + assertRejected( + CALLBACK, "temporal:///namespaces/ns/schedules/sched-id/run-id/callbacks/req-id"); + assertRejected(CALLBACK, "temporal:///namespaces/ns/workflows/wf-id/run-id/history/req-id"); } /** The declared type is authoritative on every decoder, not just the dispatcher. */ @@ -369,6 +442,9 @@ public void rejectsTypeThatDoesNotMatchThePath() { assertNull(nexusLinkToWorkflowLink(nexusLink(WORKFLOW_EVENT, workflowEventUrl))); assertNull(nexusLinkToActivity(nexusLink(WORKFLOW_EVENT, workflowEventUrl))); assertNull(nexusLinkToNexusOperation(nexusLink(WORKFLOW_EVENT, workflowEventUrl))); + String callbackUrl = "temporal:///namespaces/ns/workflows/wf-id/run-id/callbacks/req-id"; + assertRejected(ACTIVITY, callbackUrl); + assertNull(nexusLinkToCallback(nexusLink(ACTIVITY, callbackUrl))); } @Test @@ -421,11 +497,20 @@ public void rejectsMalformedUrlsWithoutThrowing() { // =============================================================================================== @Test - public void dispatchCoversAllFourLinkTypes() { + public void dispatchCoversAllLinkTypes() { assertNotNull(linkToNexusLink(eventRef("ns", "w", "r", 1, EventType.EVENT_TYPE_TIMER_STARTED))); assertNotNull(linkToNexusLink(workflow("ns", "w", "r", ""))); assertNotNull(linkToNexusLink(nexusOperation("ns", "o", "r"))); assertNotNull(linkToNexusLink(activity("ns", "a", "r"))); + assertNotNull( + linkToNexusLink(callback("ns", ExecutionType.EXECUTION_TYPE_WORKFLOW, "w", "r", "q"))); + } + + /** A callback on an execution type with no path keyword is dropped rather than misencoded. */ + @Test + public void unsupportedCallbackExecutionTypeEncodesToNull() { + assertNull( + linkToNexusLink(callback("ns", ExecutionType.EXECUTION_TYPE_UNSPECIFIED, "w", "r", "q"))); } /** @@ -553,4 +638,22 @@ private static Link activity(String ns, String actId, String runId) { Link.Activity.newBuilder().setNamespace(ns).setActivityId(actId).setRunId(runId)) .build(); } + + private static Link callback( + String ns, + ExecutionType type, + String businessId, + String runId, + String requestId, + String... componentPath) { + return Link.newBuilder() + .setCallback( + Link.Callback.newBuilder() + .setNamespace(ns) + .setExecution( + Execution.newBuilder().setType(type).setBusinessId(businessId).setRunId(runId)) + .setRequestId(requestId) + .addAllComponentPath(Arrays.asList(componentPath))) + .build(); + } }