diff --git a/temporal-sdk/src/main/java/io/temporal/internal/client/RootWorkflowClientInvoker.java b/temporal-sdk/src/main/java/io/temporal/internal/client/RootWorkflowClientInvoker.java index 502c12e8e..121c62fc1 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/client/RootWorkflowClientInvoker.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/client/RootWorkflowClientInvoker.java @@ -434,6 +434,14 @@ public QueryOutput query(QueryInput input) { QueryWorkflowResponse result; result = genericClient.query(request); + // A query writes nothing to history, so the server returns a link to the workflow execution + // that processed it rather than to an event. When the query is issued from inside a Nexus + // operation handler, propagate that link so the caller's Nexus operation event points at the + // queried workflow. Older servers leave it unset. + if (CurrentNexusOperationContext.isNexusContext() && result.hasLink()) { + CurrentNexusOperationContext.get().addResponseLink(result.getLink()); + } + boolean queryRejected = result.hasQueryRejected(); WorkflowExecutionStatus rejectStatus = queryRejected ? result.getQueryRejected().getStatus() : null; 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 54b3cdc72..5505270ff 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 @@ -25,10 +25,12 @@ public class LinkConverter { "temporal:///namespaces/%s/nexus-operations/%s/%s/details"; private static final String activityLinkPathFormat = "temporal:///namespaces/%s/activities/%s/%s/details"; + private static final String workflowLinkPathFormat = "temporal:///namespaces/%s/workflows/%s/%s"; private static final String linkReferenceTypeKey = "referenceType"; private static final String linkEventIDKey = "eventID"; private static final String linkEventTypeKey = "eventType"; private static final String linkRequestIDKey = "requestID"; + private static final String linkReasonKey = "reason"; private static final String eventReferenceType = Link.WorkflowEvent.EventReference.getDescriptor().getName(); @@ -98,14 +100,28 @@ public static io.temporal.api.nexus.v1.Link workflowEventToNexusLink(Link.Workfl return null; } + /** + * Converts a {@link Link.Workflow} to a Nexus link. A workflow link addresses a workflow + * execution as a whole rather than one event within it, so the URL uses the workflow path and + * carries no event path suffix and no reference query params. It is used when there is no history + * event to point at, for example a Query or a rejected Update. The optional {@code reason} + * explaining why the link exists is carried as a query param. + */ public static io.temporal.api.nexus.v1.Link workflowLinkToNexusLink(Link.Workflow w) { try { - String namespace = URLEncoder.encode(w.getNamespace(), StandardCharsets.UTF_8.toString()); - String workflowId = - URLEncoder.encode(w.getWorkflowId(), StandardCharsets.UTF_8.toString()) - .replace("+", "%20"); // handle workflowIds supporting spaces - String runId = URLEncoder.encode(w.getRunId(), StandardCharsets.UTF_8.toString()); - String url = String.format(linkPathFormat, namespace, workflowId, runId); + String url = + String.format( + workflowLinkPathFormat, + encodePathSegment(w.getNamespace()), + encodePathSegment(w.getWorkflowId()), + encodePathSegment(w.getRunId())); + if (!w.getReason().isEmpty()) { + url += + "?" + + linkReasonKey + + "=" + + URLEncoder.encode(w.getReason(), StandardCharsets.UTF_8.toString()); + } return io.temporal.api.nexus.v1.Link.newBuilder() .setUrl(url) .setType(workflowLinkType) @@ -190,16 +206,25 @@ public static Link nexusLinkToWorkflowEvent(io.temporal.api.nexus.v1.Link nexusL } public static Link nexusLinkToWorkflowLink(io.temporal.api.nexus.v1.Link nexusLink) { + if (!workflowLinkType.equals(nexusLink.getType())) { + log.error( + "Failed to parse Nexus link URL: cannot parse link type {} to {}", + nexusLink.getType(), + workflowLinkType); + return null; + } Link.Builder link = Link.newBuilder(); try { URI uri = new URI(nexusLink.getUrl()); - log.debug("Parsing nexus link URL: {}", uri.getRawPath()); - if (!uri.getScheme().equals(temporalUrlScheme)) { + + // Compared in this order so a URL with no scheme at all reports the invalid scheme rather + // than throwing. + if (!temporalUrlScheme.equals(uri.getScheme())) { log.error("Failed to parse Nexus link URL: invalid scheme: {}", uri.getScheme()); return null; } + StringTokenizer st = new StringTokenizer(uri.getRawPath(), "/"); - // maybe add constants for "namespaces", "workflows" too if (!st.nextToken().equals("namespaces")) { log.error("Failed to parse Nexus link URL: invalid path: {}", uri.getRawPath()); return null; @@ -210,18 +235,28 @@ public static Link nexusLinkToWorkflowLink(io.temporal.api.nexus.v1.Link nexusLi return null; } String workflowID = URLDecoder.decode(st.nextToken(), StandardCharsets.UTF_8.toString()); - if (!st.hasMoreTokens()) { + String runID = URLDecoder.decode(st.nextToken(), StandardCharsets.UTF_8.toString()); + // The run ID ends a workflow link, so anything trailing means this is a different link + // shape. In particular this rejects the workflow-event form, which ends in "/history". + if (st.hasMoreTokens()) { log.error("Failed to parse Nexus link URL: invalid path: {}", uri.getRawPath()); return null; } - String runID = URLDecoder.decode(st.nextToken(), StandardCharsets.UTF_8.toString()); - link.setWorkflow( + + Link.Workflow.Builder w = Link.Workflow.newBuilder() .setNamespace(namespace) .setWorkflowId(workflowID) - .setRunId(runID)); + .setRunId(runID); + String reason = rawQueryParam(uri, linkReasonKey); + if (reason != null) { + w.setReason(reason); + } + + link.setWorkflow(w); } catch (Exception e) { - log.error("Failed to convert NexusLink {} to WorkflowLink", nexusLink, e); + // Swallow un-parsable links since they are not critical to processing. + log.error("Failed to parse Nexus link URL", e); return null; } return link.build(); @@ -406,6 +441,33 @@ public static Link nexusLinkToNexusOperation(io.temporal.api.nexus.v1.Link nexus return link.build(); } + /** + * Percent-encodes a single URL path segment. {@link URLEncoder} targets form encoding, where a + * space becomes '+', so rewrite it to "%20" as required for a path. + */ + private static String encodePathSegment(String value) throws UnsupportedEncodingException { + return URLEncoder.encode(value, StandardCharsets.UTF_8.toString()).replace("+", "%20"); + } + + /** + * Reads a single param out of the raw, still-encoded query string, or returns null when the param + * is absent. Unlike {@link #parseQueryParams} the value is decoded exactly once, so values that + * themselves contain '=' or '&' survive the round trip. + */ + private static String rawQueryParam(URI uri, String key) throws UnsupportedEncodingException { + final String rawQuery = uri.getRawQuery(); + if (rawQuery == null || rawQuery.isEmpty()) { + return null; + } + for (String pair : rawQuery.split("&")) { + final String[] kv = pair.split("=", 2); + if (kv[0].equals(key)) { + return kv.length == 2 ? URLDecoder.decode(kv[1], StandardCharsets.UTF_8.toString()) : ""; + } + } + return null; + } + private static Map parseQueryParams(URI uri) throws UnsupportedEncodingException { final String query = uri.getQuery(); if (query == null || query.isEmpty()) { diff --git a/temporal-sdk/src/test/java/io/temporal/internal/client/RootWorkflowClientInvokerLinkPropagationTest.java b/temporal-sdk/src/test/java/io/temporal/internal/client/RootWorkflowClientInvokerLinkPropagationTest.java index a597ae96d..e9cc66712 100644 --- a/temporal-sdk/src/test/java/io/temporal/internal/client/RootWorkflowClientInvokerLinkPropagationTest.java +++ b/temporal-sdk/src/test/java/io/temporal/internal/client/RootWorkflowClientInvokerLinkPropagationTest.java @@ -7,10 +7,15 @@ import com.uber.m3.tally.RootScopeBuilder; import com.uber.m3.tally.Scope; import io.temporal.api.common.v1.Link; +import io.temporal.api.common.v1.Payloads; import io.temporal.api.common.v1.WorkflowExecution; import io.temporal.api.enums.v1.EventType; import io.temporal.api.enums.v1.UpdateWorkflowExecutionLifecycleStage; +import io.temporal.api.enums.v1.WorkflowExecutionStatus; +import io.temporal.api.query.v1.QueryRejected; import io.temporal.api.update.v1.UpdateRef; +import io.temporal.api.workflowservice.v1.QueryWorkflowRequest; +import io.temporal.api.workflowservice.v1.QueryWorkflowResponse; import io.temporal.api.workflowservice.v1.SignalWithStartWorkflowExecutionRequest; import io.temporal.api.workflowservice.v1.SignalWithStartWorkflowExecutionResponse; import io.temporal.api.workflowservice.v1.SignalWorkflowExecutionRequest; @@ -23,7 +28,9 @@ import io.temporal.client.WorkflowClientOptions; import io.temporal.client.WorkflowOptions; import io.temporal.client.WorkflowUpdateStage; +import io.temporal.common.converter.DefaultDataConverter; import io.temporal.common.interceptors.Header; +import io.temporal.common.interceptors.WorkflowClientCallsInterceptor; import io.temporal.common.interceptors.WorkflowClientCallsInterceptor.StartUpdateInput; import io.temporal.common.interceptors.WorkflowClientCallsInterceptor.WorkflowSignalInput; import io.temporal.common.interceptors.WorkflowClientCallsInterceptor.WorkflowSignalWithStartInput; @@ -348,6 +355,116 @@ private static StartUpdateInput newStartUpdateInput() { .build()); } + /** + * A query never writes to history, so the server answers with a {@code Link.Workflow} naming the + * execution that processed it instead of a {@code Link.WorkflowEvent}. That link has to reach the + * operation context so the caller's Nexus operation event points back at the queried workflow. + */ + @Test + public void queryCapturesWorkflowResponseLink() { + Link responseLink = workflowLink(WORKFLOW_ID, "target-run", "Query processed"); + when(genericClient.query(any(QueryWorkflowRequest.class))) + .thenReturn( + QueryWorkflowResponse.newBuilder() + .setLink(responseLink) + .setQueryResult(queryResult("answer")) + .build()); + + WorkflowClientCallsInterceptor.QueryOutput output = invoker.query(newQueryInput()); + + List captured = nexusCtx.getResponseLinks(); + Assert.assertEquals("expected one captured response link", 1, captured.size()); + Assert.assertEquals(responseLink, captured.get(0)); + + // Capturing the link must not disturb the query's own result. + Assert.assertFalse(output.isQueryRejected()); + Assert.assertEquals("answer", output.getResult()); + } + + /** + * Two queries in a row each contribute a response link; both must accumulate in call order on the + * shared list, exactly as the signal path does. + */ + @Test + public void multipleQueriesAccumulateAllResponseLinks() { + Link firstResponseLink = workflowLink("callee-a", "run-a", "Query processed"); + Link secondResponseLink = workflowLink("callee-b", "run-b", "Query processed"); + when(genericClient.query(any(QueryWorkflowRequest.class))) + .thenReturn(QueryWorkflowResponse.newBuilder().setLink(firstResponseLink).build()) + .thenReturn(QueryWorkflowResponse.newBuilder().setLink(secondResponseLink).build()); + + invoker.query(newQueryInput()); + invoker.query(newQueryInput()); + + Assert.assertEquals( + "expected one response link per query call, in call order", + Arrays.asList(firstResponseLink, secondResponseLink), + nexusCtx.getResponseLinks()); + } + + /** + * A rejected query still carries a link to the workflow that rejected it, and the link is + * captured before the rejection is surfaced. This matches sdk-go, where the link is recorded + * ahead of the QueryRejected branch. Pins the ordering so it is not "fixed" into the wrong + * behavior later. + */ + @Test + public void rejectedQueryStillCapturesResponseLink() { + Link responseLink = workflowLink(WORKFLOW_ID, "target-run", "Query processed"); + when(genericClient.query(any(QueryWorkflowRequest.class))) + .thenReturn( + QueryWorkflowResponse.newBuilder() + .setLink(responseLink) + .setQueryRejected( + QueryRejected.newBuilder() + .setStatus(WorkflowExecutionStatus.WORKFLOW_EXECUTION_STATUS_COMPLETED)) + .build()); + + WorkflowClientCallsInterceptor.QueryOutput output = invoker.query(newQueryInput()); + + Assert.assertTrue("expected the query to be reported as rejected", output.isQueryRejected()); + Assert.assertEquals( + "expected the response link to be captured even for a rejected query", + Collections.singletonList(responseLink), + nexusCtx.getResponseLinks()); + } + + /** + * Older-server compatibility: {@code QueryWorkflowResponse.link} is unset, so nothing is captured + * and the query itself still succeeds. + */ + @Test + public void queryAgainstOlderServerCapturesNoResponseLink() { + when(genericClient.query(any(QueryWorkflowRequest.class))) + .thenReturn(QueryWorkflowResponse.getDefaultInstance()); + + invoker.query(newQueryInput()); + + Assert.assertTrue( + "expected no captured response link when server returned no link", + nexusCtx.getResponseLinks().isEmpty()); + } + + /** + * A query issued outside a Nexus operation handler must not touch the operation context at all. + * Guards against the propagation being reached without a context, which would throw. + */ + @Test + public void queryOutsideNexusContextIgnoresResponseLink() { + CurrentNexusOperationContext.unset(); + when(genericClient.query(any(QueryWorkflowRequest.class))) + .thenReturn( + QueryWorkflowResponse.newBuilder() + .setLink(workflowLink(WORKFLOW_ID, "target-run", "Query processed")) + .build()); + + invoker.query(newQueryInput()); + + Assert.assertTrue( + "a query outside a Nexus context must not record response links", + nexusCtx.getResponseLinks().isEmpty()); + } + // ── helpers ────────────────────────────────────────────────────────────────────────────── private static WorkflowSignalInput newSignalInput() { @@ -374,6 +491,33 @@ private static WorkflowSignalWithStartInput newSignalWithStartInput() { startInput, "test-signal", new Object[] {"signal-payload"}); } + private static WorkflowClientCallsInterceptor.QueryInput newQueryInput() { + return new WorkflowClientCallsInterceptor.QueryInput<>( + WorkflowExecution.newBuilder().setWorkflowId(WORKFLOW_ID).build(), + "test-query", + Header.empty(), + new Object[] {}, + String.class, + String.class); + } + + private static Payloads queryResult(String value) { + return DefaultDataConverter.STANDARD_INSTANCE + .toPayloads(value) + .orElseThrow(() -> new IllegalStateException("expected payloads")); + } + + private static Link workflowLink(String workflowId, String runId, String reason) { + return Link.newBuilder() + .setWorkflow( + Link.Workflow.newBuilder() + .setNamespace(NAMESPACE) + .setWorkflowId(workflowId) + .setRunId(runId) + .setReason(reason)) + .build(); + } + private static Link workflowEventLink(String workflowId, String runId, EventType eventType) { return Link.newBuilder() .setWorkflowEvent( 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 34868a247..2450e68d4 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 @@ -6,8 +6,10 @@ import static io.temporal.internal.common.LinkConverter.nexusLinkToLink; import static io.temporal.internal.common.LinkConverter.nexusLinkToNexusOperation; import static io.temporal.internal.common.LinkConverter.nexusLinkToWorkflowEvent; +import static io.temporal.internal.common.LinkConverter.nexusLinkToWorkflowLink; import static io.temporal.internal.common.LinkConverter.nexusOperationToNexusLink; import static io.temporal.internal.common.LinkConverter.workflowEventToNexusLink; +import static io.temporal.internal.common.LinkConverter.workflowLinkToNexusLink; import static org.junit.Assert.*; import io.temporal.api.common.v1.Link; @@ -634,4 +636,309 @@ public void testLinkToNexusLink_Activity() { public void testLinkToNexusLink_Empty() { assertNull(linkToNexusLink(Link.newBuilder().build())); } + + @Test + public void testConvertWorkflowToNexus_Valid() { + Link.Workflow input = + Link.Workflow.newBuilder() + .setNamespace("ns") + .setWorkflowId("wf-id") + .setRunId("run-id") + .build(); + + io.temporal.api.nexus.v1.Link expected = + io.temporal.api.nexus.v1.Link.newBuilder() + .setUrl("temporal:///namespaces/ns/workflows/wf-id/run-id") + .setType("temporal.api.common.v1.Link.Workflow") + .build(); + + assertEquals(expected, workflowLinkToNexusLink(input)); + } + + @Test + public void testConvertWorkflowToNexus_ValidReason() { + Link.Workflow input = + Link.Workflow.newBuilder() + .setNamespace("ns") + .setWorkflowId("wf-id") + .setRunId("run-id") + .setReason("rejected update") + .build(); + + io.temporal.api.nexus.v1.Link expected = + io.temporal.api.nexus.v1.Link.newBuilder() + .setUrl("temporal:///namespaces/ns/workflows/wf-id/run-id?reason=rejected+update") + .setType("temporal.api.common.v1.Link.Workflow") + .build(); + + assertEquals(expected, workflowLinkToNexusLink(input)); + } + + @Test + public void testConvertWorkflowToNexus_ValidSlash() { + Link.Workflow input = + Link.Workflow.newBuilder() + .setNamespace("ns") + .setWorkflowId("wf/id") + .setRunId("run-id") + .build(); + + io.temporal.api.nexus.v1.Link expected = + io.temporal.api.nexus.v1.Link.newBuilder() + .setUrl("temporal:///namespaces/ns/workflows/wf%2Fid/run-id") + .setType("temporal.api.common.v1.Link.Workflow") + .build(); + + assertEquals(expected, workflowLinkToNexusLink(input)); + } + + @Test + public void testConvertWorkflowToNexus_ValidSpace() throws UnsupportedEncodingException { + Link.Workflow input = + Link.Workflow.newBuilder() + .setNamespace("ns") + .setWorkflowId("wf id") + .setRunId("run-id") + .build(); + + io.temporal.api.nexus.v1.Link expected = + io.temporal.api.nexus.v1.Link.newBuilder() + .setUrl("temporal:///namespaces/ns/workflows/wf%20id/run-id") + .setType("temporal.api.common.v1.Link.Workflow") + .build(); + + io.temporal.api.nexus.v1.Link actual = workflowLinkToNexusLink(input); + assertEquals(expected, actual); + // A space in the path has to survive as %20 rather than the '+' that form encoding would + // produce, otherwise the link resolves to a different workflow ID. + assertEquals( + "temporal:///namespaces/ns/workflows/wf id/run-id", + URLDecoder.decode(actual.getUrl(), StandardCharsets.UTF_8.toString())); + } + + @Test + public void testConvertNexusToWorkflow_Valid() { + io.temporal.api.nexus.v1.Link input = + io.temporal.api.nexus.v1.Link.newBuilder() + .setUrl("temporal:///namespaces/ns/workflows/wf-id/run-id") + .setType("temporal.api.common.v1.Link.Workflow") + .build(); + + Link expected = + Link.newBuilder() + .setWorkflow( + Link.Workflow.newBuilder() + .setNamespace("ns") + .setWorkflowId("wf-id") + .setRunId("run-id")) + .build(); + + assertEquals(expected, nexusLinkToWorkflowLink(input)); + } + + @Test + public void testConvertNexusToWorkflow_ValidReason() { + io.temporal.api.nexus.v1.Link input = + io.temporal.api.nexus.v1.Link.newBuilder() + .setUrl("temporal:///namespaces/ns/workflows/wf-id/run-id?reason=rejected+update") + .setType("temporal.api.common.v1.Link.Workflow") + .build(); + + Link expected = + Link.newBuilder() + .setWorkflow( + Link.Workflow.newBuilder() + .setNamespace("ns") + .setWorkflowId("wf-id") + .setRunId("run-id") + .setReason("rejected update")) + .build(); + + assertEquals(expected, nexusLinkToWorkflowLink(input)); + } + + @Test + public void testConvertNexusToWorkflow_WrongType() { + io.temporal.api.nexus.v1.Link input = + io.temporal.api.nexus.v1.Link.newBuilder() + .setUrl("temporal:///namespaces/ns/workflows/wf-id/run-id") + .setType("temporal.api.common.v1.Link.WorkflowEvent") + .build(); + + assertNull(nexusLinkToWorkflowLink(input)); + } + + @Test + public void testConvertNexusToWorkflow_InvalidScheme() { + io.temporal.api.nexus.v1.Link input = + io.temporal.api.nexus.v1.Link.newBuilder() + .setUrl("random:///namespaces/ns/workflows/wf-id/run-id") + .setType("temporal.api.common.v1.Link.Workflow") + .build(); + + assertNull(nexusLinkToWorkflowLink(input)); + } + + @Test + public void testConvertNexusToWorkflow_InvalidPathTrailingSegment() { + // The workflow-event form addresses an event inside the workflow, so it must not be accepted + // as a workflow link even when the type says otherwise. + io.temporal.api.nexus.v1.Link input = + io.temporal.api.nexus.v1.Link.newBuilder() + .setUrl("temporal:///namespaces/ns/workflows/wf-id/run-id/history") + .setType("temporal.api.common.v1.Link.Workflow") + .build(); + + assertNull(nexusLinkToWorkflowLink(input)); + } + + @Test + public void testConvertNexusToWorkflow_ReasonNotFirstQueryParam() { + // The reason is located by key, not by position, so unrelated params ahead of it are skipped. + io.temporal.api.nexus.v1.Link input = + io.temporal.api.nexus.v1.Link.newBuilder() + .setUrl( + "temporal:///namespaces/ns/workflows/wf-id/run-id?foo=bar&reason=Query+processed") + .setType("temporal.api.common.v1.Link.Workflow") + .build(); + + assertEquals("Query processed", nexusLinkToWorkflowLink(input).getWorkflow().getReason()); + } + + @Test + public void testConvertNexusToWorkflow_EmptyReasonValue() { + io.temporal.api.nexus.v1.Link input = + io.temporal.api.nexus.v1.Link.newBuilder() + .setUrl("temporal:///namespaces/ns/workflows/wf-id/run-id?reason=") + .setType("temporal.api.common.v1.Link.Workflow") + .build(); + + assertEquals("", nexusLinkToWorkflowLink(input).getWorkflow().getReason()); + } + + @Test + public void testConvertNexusToWorkflow_BareReasonKey() { + // A key with no '=' must not blow up on the missing value. + io.temporal.api.nexus.v1.Link input = + io.temporal.api.nexus.v1.Link.newBuilder() + .setUrl("temporal:///namespaces/ns/workflows/wf-id/run-id?reason") + .setType("temporal.api.common.v1.Link.Workflow") + .build(); + + assertEquals("", nexusLinkToWorkflowLink(input).getWorkflow().getReason()); + } + + @Test + public void testConvertNexusToWorkflow_ReasonPrefixKeyIgnored() { + // "reasonx" must not be treated as "reason". + io.temporal.api.nexus.v1.Link input = + io.temporal.api.nexus.v1.Link.newBuilder() + .setUrl("temporal:///namespaces/ns/workflows/wf-id/run-id?reasonx=nope") + .setType("temporal.api.common.v1.Link.Workflow") + .build(); + + assertEquals("", nexusLinkToWorkflowLink(input).getWorkflow().getReason()); + } + + @Test + public void testConvertNexusToWorkflow_EmptyUrl() { + // A URL with no scheme must be reported as an invalid scheme rather than throwing. + io.temporal.api.nexus.v1.Link input = + io.temporal.api.nexus.v1.Link.newBuilder() + .setUrl("") + .setType("temporal.api.common.v1.Link.Workflow") + .build(); + + assertNull(nexusLinkToWorkflowLink(input)); + } + + /** + * Characterization test, not a statement of intent. {@link java.net.URLDecoder} performs form + * decoding, so it turns a literal '+' in a path segment into a space. Other SDKs escape paths + * with Go's {@code url.PathEscape}, which leaves '+' literal, so a link they produce for a + * workflow ID containing '+' is currently decoded incorrectly here. The same flaw exists in + * {@link LinkConverter#nexusLinkToWorkflowEvent} and {@link + * LinkConverter#nexusLinkToNexusOperation}. When path decoding is fixed across all three + * converters, invert this assertion to expect "a+b". + */ + @Test + public void testConvertNexusToWorkflow_LiteralPlusInPathDecodesToSpace() { + io.temporal.api.nexus.v1.Link input = + io.temporal.api.nexus.v1.Link.newBuilder() + .setUrl("temporal:///namespaces/ns/workflows/a+b/run-id") + .setType("temporal.api.common.v1.Link.Workflow") + .build(); + + assertEquals("a b", nexusLinkToWorkflowLink(input).getWorkflow().getWorkflowId()); + + // A '+' this SDK encoded itself does survive, because URLEncoder emits %2B. + Link.Workflow w = + Link.Workflow.newBuilder() + .setNamespace("ns") + .setWorkflowId("a+b") + .setRunId("run-id") + .build(); + assertEquals( + Link.newBuilder().setWorkflow(w).build(), + nexusLinkToWorkflowLink(workflowLinkToNexusLink(w))); + } + + @Test + public void testConvertNexusToWorkflow_InvalidPathMissingRunID() { + io.temporal.api.nexus.v1.Link input = + io.temporal.api.nexus.v1.Link.newBuilder() + .setUrl("temporal:///namespaces/ns/workflows/wf-id") + .setType("temporal.api.common.v1.Link.Workflow") + .build(); + + assertNull(nexusLinkToWorkflowLink(input)); + } + + @Test + public void testWorkflowLinkRoundTrip() { + // Reserved characters in every field at once: the path segments are percent-escaped and the + // reason is form-encoded, so a reason containing '=' and '&' must not be split as query syntax. + Link.Workflow w = + Link.Workflow.newBuilder() + .setNamespace("ns/with/slash") + .setWorkflowId("wf id with space") + .setRunId("run-id") + .setReason("reason with = and &") + .build(); + + io.temporal.api.nexus.v1.Link nexusLink = workflowLinkToNexusLink(w); + assertEquals("temporal.api.common.v1.Link.Workflow", nexusLink.getType()); + assertEquals(Link.newBuilder().setWorkflow(w).build(), nexusLinkToWorkflowLink(nexusLink)); + } + + @Test + public void testLinkToNexusLink_Workflow() { + Link.Workflow w = + Link.Workflow.newBuilder() + .setNamespace("ns") + .setWorkflowId("wf-id") + .setRunId("run-id") + .setReason("Query processed") + .build(); + + io.temporal.api.nexus.v1.Link actual = + linkToNexusLink(Link.newBuilder().setWorkflow(w).build()); + assertEquals(workflowLinkToNexusLink(w), actual); + } + + @Test + public void testNexusLinkToLink_WorkflowRoundTrip() { + Link.Workflow w = + Link.Workflow.newBuilder() + .setNamespace("ns") + .setWorkflowId("wf-id") + .setRunId("run-id") + .setReason("Query processed") + .build(); + + io.temporal.api.nexus.v1.Link nexusLink = workflowLinkToNexusLink(w); + Link converted = nexusLinkToLink(nexusLink); + assertNotNull(converted); + assertEquals(Link.newBuilder().setWorkflow(w).build(), converted); + } } diff --git a/temporal-sdk/src/test/java/io/temporal/workflow/nexus/QueryOperationTest.java b/temporal-sdk/src/test/java/io/temporal/workflow/nexus/QueryOperationTest.java new file mode 100644 index 000000000..709c1ecbd --- /dev/null +++ b/temporal-sdk/src/test/java/io/temporal/workflow/nexus/QueryOperationTest.java @@ -0,0 +1,316 @@ +package io.temporal.workflow.nexus; + +import io.nexusrpc.Operation; +import io.nexusrpc.Service; +import io.nexusrpc.handler.HandlerException; +import io.nexusrpc.handler.OperationHandler; +import io.nexusrpc.handler.OperationImpl; +import io.nexusrpc.handler.ServiceImpl; +import io.temporal.api.enums.v1.QueryRejectCondition; +import io.temporal.client.WorkflowClient; +import io.temporal.client.WorkflowClientOptions; +import io.temporal.client.WorkflowFailedException; +import io.temporal.client.WorkflowOptions; +import io.temporal.client.WorkflowStub; +import io.temporal.client.WorkflowTargetOptions; +import io.temporal.failure.NexusOperationFailure; +import io.temporal.nexus.Nexus; +import io.temporal.testing.internal.SDKTestWorkflowRule; +import io.temporal.workflow.NexusOperationOptions; +import io.temporal.workflow.NexusServiceOptions; +import io.temporal.workflow.QueryMethod; +import io.temporal.workflow.SignalMethod; +import io.temporal.workflow.Workflow; +import io.temporal.workflow.WorkflowInterface; +import io.temporal.workflow.WorkflowMethod; +import java.time.Duration; +import java.util.UUID; +import org.junit.Assert; +import org.junit.Rule; +import org.junit.Test; +import org.junit.function.ThrowingRunnable; + +/** + * A Nexus operation backed by a workflow Query. A Query is always synchronous and writes nothing to + * history, so the handler simply queries and returns the result; there is no operation token and no + * completion callback. + * + *

Covers the value round trip plus the failure modes a caller can observe: an unknown workflow, + * a query handler that throws, and a Query rejected by the client's reject condition. All of these + * must fail the caller's Nexus operation rather than hanging or returning a default. + * + *

The response link the server attaches to {@code QueryWorkflowResponse} is verified in {@link + * io.temporal.internal.client.RootWorkflowClientInvokerLinkPropagationTest} instead, since neither + * the in-memory test server nor a released real server populates that field yet. + */ +public class QueryOperationTest { + + private static final int BUMPS = 2; + + @Rule + public SDKTestWorkflowRule testWorkflowRule = + SDKTestWorkflowRule.newBuilder() + .setWorkflowTypes(QueryCallerWorkflowImpl.class, CounterWorkflowImpl.class) + .setNexusServiceImplementation(new QueryingNexusServiceImpl()) + // The workflow being queried parks on a signal, so time skipping would fast-forward it + // into its execution timeout and it would be gone before the Query lands. + .setUseTimeskipping(false) + // Matches the reject condition the Go test passes per request; in Java the condition is a + // client-level option. + .setWorkflowClientOptions( + WorkflowClientOptions.newBuilder() + .setQueryRejectCondition(QueryRejectCondition.QUERY_REJECT_CONDITION_NOT_OPEN) + .build()) + .build(); + + @Test + public void queryOperationReturnsResult() { + String targetWorkflowId = startCounterWorkflow(); + bumpCounter(targetWorkflowId, BUMPS); + + QueryCallerWorkflow caller = + testWorkflowRule.newWorkflowStubTimeoutOptions(QueryCallerWorkflow.class, "query-caller"); + Assert.assertEquals( + "the operation should return what the query handler computed from workflow state", + BUMPS, + caller.execute(new QueryRequest(targetWorkflowId, "", false))); + + completeCounterWorkflow(targetWorkflowId); + } + + @Test + public void queryOnUnknownWorkflowFailsOperation() { + QueryCallerWorkflow caller = + testWorkflowRule.newWorkflowStubTimeoutOptions( + QueryCallerWorkflow.class, "unknown-wid-caller"); + + assertOperationFailedWith( + HandlerException.ErrorType.NOT_FOUND, + () -> caller.execute(new QueryRequest("unknown-wid-" + UUID.randomUUID(), "", false))); + } + + @Test + public void queryOnUnknownRunFailsOperation() { + String targetWorkflowId = startCounterWorkflow(); + QueryCallerWorkflow caller = + testWorkflowRule.newWorkflowStubTimeoutOptions( + QueryCallerWorkflow.class, "unknown-rid-caller"); + + assertOperationFailedWith( + HandlerException.ErrorType.NOT_FOUND, + () -> + caller.execute( + new QueryRequest(targetWorkflowId, UUID.randomUUID().toString(), false))); + + completeCounterWorkflow(targetWorkflowId); + } + + @Test + public void failedQueryFailsOperation() { + String targetWorkflowId = startCounterWorkflow(); + QueryCallerWorkflow caller = + testWorkflowRule.newWorkflowStubTimeoutOptions( + QueryCallerWorkflow.class, "failed-query-caller"); + + assertOperationFailedWith( + HandlerException.ErrorType.BAD_REQUEST, + () -> caller.execute(new QueryRequest(targetWorkflowId, "", true))); + + completeCounterWorkflow(targetWorkflowId); + } + + @Test + public void rejectedQueryFailsOperation() { + // The reject condition is NOT_OPEN, so querying a workflow that has already closed is rejected + // and must surface as an operation failure. + String targetWorkflowId = startCounterWorkflow(); + completeCounterWorkflow(targetWorkflowId); + + QueryCallerWorkflow caller = + testWorkflowRule.newWorkflowStubTimeoutOptions( + QueryCallerWorkflow.class, "rejected-query-caller"); + + assertOperationFailedWith( + HandlerException.ErrorType.BAD_REQUEST, + () -> caller.execute(new QueryRequest(targetWorkflowId, "", false))); + } + + // ── helpers ────────────────────────────────────────────────────────────────────────────── + + /** + * Asserts the caller's operation failed, and that it failed with the specific handler error type + * the SDK is supposed to derive from what the handler threw. Asserting only {@code + * NexusOperationFailure} would still pass if every failure collapsed into one retryable type, so + * the mapping in {@code NexusTaskHandlerImpl.convertKnownFailures} is pinned here. + */ + private static void assertOperationFailedWith( + HandlerException.ErrorType expectedErrorType, ThrowingRunnable callerInvocation) { + WorkflowFailedException e = + Assert.assertThrows(WorkflowFailedException.class, callerInvocation); + Assert.assertTrue( + "expected the caller to fail with a NexusOperationFailure but got: " + e.getCause(), + e.getCause() instanceof NexusOperationFailure); + + Throwable handlerFailure = e.getCause().getCause(); + Assert.assertTrue( + "expected a HandlerException under the NexusOperationFailure but got: " + handlerFailure, + handlerFailure instanceof HandlerException); + Assert.assertEquals(expectedErrorType, ((HandlerException) handlerFailure).getErrorType()); + } + + private String startCounterWorkflow() { + String workflowId = "counter-" + UUID.randomUUID(); + WorkflowStub stub = + testWorkflowRule + .getWorkflowClient() + .newUntypedWorkflowStub( + "CounterWorkflow", + WorkflowOptions.newBuilder() + .setWorkflowId(workflowId) + .setTaskQueue(testWorkflowRule.getTaskQueue()) + .build()); + stub.start(); + return workflowId; + } + + private void bumpCounter(String workflowId, int times) { + CounterWorkflow stub = + testWorkflowRule.getWorkflowClient().newWorkflowStub(CounterWorkflow.class, workflowId); + for (int i = 0; i < times; i++) { + stub.bump(); + } + } + + private void completeCounterWorkflow(String workflowId) { + WorkflowStub stub = testWorkflowRule.getWorkflowClient().newUntypedWorkflowStub(workflowId); + stub.signal("done"); + stub.getResult(Integer.class); + } + + // ── workflows ──────────────────────────────────────────────────────────────────────────── + + /** Target of the Query: holds a counter that signals advance and a query reads. */ + @WorkflowInterface + public interface CounterWorkflow { + @WorkflowMethod + int execute(); + + @QueryMethod + int getCount(boolean fail); + + @SignalMethod + void bump(); + + @SignalMethod + void done(); + } + + public static class CounterWorkflowImpl implements CounterWorkflow { + private int counter; + private boolean completed; + + @Override + public int execute() { + Workflow.await(() -> completed); + return counter; + } + + @Override + public int getCount(boolean fail) { + if (fail) { + // A query handler that throws makes the server answer with a query failure, which the + // handler surfaces to the caller as a failed operation. + throw new IllegalStateException("query failed (for testing)"); + } + return counter; + } + + @Override + public void bump() { + counter++; + } + + @Override + public void done() { + completed = true; + } + } + + @WorkflowInterface + public interface QueryCallerWorkflow { + @WorkflowMethod + int execute(QueryRequest request); + } + + public static class QueryCallerWorkflowImpl implements QueryCallerWorkflow { + @Override + public int execute(QueryRequest request) { + TestNexusQueryService service = + Workflow.newNexusServiceStub( + TestNexusQueryService.class, + NexusServiceOptions.newBuilder() + .setOperationOptions( + NexusOperationOptions.newBuilder() + .setScheduleToCloseTimeout(Duration.ofSeconds(20)) + .build()) + .build()); + return service.query(request); + } + } + + // ── nexus service ──────────────────────────────────────────────────────────────────────── + + @Service + public interface TestNexusQueryService { + @Operation + Integer query(QueryRequest input); + } + + @ServiceImpl(service = TestNexusQueryService.class) + public static class QueryingNexusServiceImpl { + @OperationImpl + public OperationHandler query() { + // A Query resolves immediately, so this is a plain synchronous operation: no operation token, + // no completion callback, nothing to cancel. + return OperationHandler.sync( + (context, details, input) -> { + WorkflowClient client = Nexus.getOperationContext().getWorkflowClient(); + WorkflowTargetOptions.Builder target = + WorkflowTargetOptions.newBuilder().setWorkflowId(input.getWorkflowId()); + if (!input.getRunId().isEmpty()) { + target.setRunId(input.getRunId()); + } + return client + .newWorkflowStub(CounterWorkflow.class, target.build()) + .getCount(input.isFail()); + }); + } + } + + /** Input describing which workflow to query and how the query should behave. */ + public static final class QueryRequest { + private String workflowId; + private String runId; + private boolean fail; + + public QueryRequest() {} + + QueryRequest(String workflowId, String runId, boolean fail) { + this.workflowId = workflowId; + this.runId = runId; + this.fail = fail; + } + + public String getWorkflowId() { + return workflowId; + } + + public String getRunId() { + return runId; + } + + public boolean isFail() { + return fail; + } + } +} diff --git a/temporal-serviceclient/src/main/proto b/temporal-serviceclient/src/main/proto index f53963d44..3f611ea9c 160000 --- a/temporal-serviceclient/src/main/proto +++ b/temporal-serviceclient/src/main/proto @@ -1 +1 @@ -Subproject commit f53963d4489c8a73aa30bd7091fe758f9896c08c +Subproject commit 3f611ea9c6bde844cdf6a0fc2b02ab65f803a707